fix: add run_id scoping to streaming handlers to prevent cross-run chunk contamination (#5376)

[devin-ai-integration]

Summary

Fixes #5376 — the singleton CrewAIEventsBus fans out LLMStreamChunkEvent to all registered stream handlers. When multiple streaming runs execute concurrently, each run's handler receives chunks from every other run, causing cross-run chunk contamination.

Approach: Each streaming run is assigned a unique run_id (UUID) that flows through via contextvars.ContextVar. Events are stamped with the run_id at emission time, and handlers filter to only accept events matching their own run_id.

Changes:

• LLMStreamChunkEvent — new optional run_id: str | None = None field
• streaming.py — new _current_stream_run_id context var, get_current_stream_run_id() accessor; create_streaming_state() generates + sets run_id; _create_stream_handler() filters by run_id
• base_llm.py / llm.py — all 4 LLMStreamChunkEvent emission sites stamp run_id=get_current_stream_run_id()
• 5 new tests in TestStreamingRunIsolation covering handler filtering, concurrent state isolation, and multi-thread isolation

Review & Testing Checklist for Human

• Filter semantics when run_id is None: The handler passes events through when either the handler's or event's run_id is None (streaming.py:148-150). This preserves backward compatibility but means un-stamped events still fan out to all handlers. Verify this is the desired behavior for your use cases.
• Context propagation in all thread/async paths: contextvars.ContextVar does NOT auto-propagate to child threads. create_chunk_generator already uses contextvars.copy_context() — verify no other code paths spawn threads without context copying that would lose the run_id.
• Test fragility in _get_run_id_from_handler: One test introspects handler closure cells to extract the captured run_id by matching UUID-shaped strings. This works but is coupled to closure internals.

Suggested test plan: Run two concurrent Crew(stream=True).kickoff() calls against a real or mocked LLM and verify each stream's chunks contain only content from its own run.

Notes

• run_id defaults to None so all existing code/serialization is unaffected
• All 39 existing streaming tests continue to pass alongside the 5 new ones

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

---

Note

Medium Risk
Changes core streaming/event-bus plumbing by introducing a per-run run_id and filtering handlers, which could affect chunk delivery in concurrent or nested streaming scenarios (especially when run_id is missing). Covered by new concurrency-focused tests but still touches a central execution path.

Overview
Fixes concurrent streaming chunk cross-talk by introducing a per-streaming-run run_id and scoping LLMStreamChunkEvent delivery to matching handlers.

create_streaming_state() now generates/reuses a UUID run scope via a ContextVar, LLM streaming emission sites stamp run_id onto each LLMStreamChunkEvent, and stream handlers drop chunks from other runs. Adds a new TestStreamingRunIsolation suite to verify handler filtering and isolation across contexts/threads.

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

[devin-ai-integration]

Prompt hidden (unlisted session)

[devin-ai-integration]

🤖 Devin AI Engineer

I'll be helping with this pull request! Here's what you should know:

✅ I will automatically:

• Address comments on this PR. Add '(aside)' to your comment to have me ignore it.
• Look at CI failures and help fix them

Note: I can only respond to comments from users who have write access to this repository.

⚙️ Control Options:

• Disable automatic comment and CI monitoring

[cursor]

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

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

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

[cursor]

Permissive filter allows un-stamped event cross-contamination

Medium Severity

The three-way AND filter (run_id is not None and event.run_id is not None and event.run_id != run_id) only rejects events when both run_ids are set and differ. When event.run_id is None, the check short-circuits and accepts the event into every handler. This means if context propagation fails (e.g., a thread spawned without contextvars.copy_context()), get_current_stream_run_id() returns None, events are emitted without a run_id, and they leak to all concurrent handlers — reproducing the exact cross-run chunk contamination this PR aims to fix. A two-condition check (run_id is not None and event.run_id != run_id) would provide complete isolation, which is safe since all four emission sites now stamp run_id.

 

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

[devin-ai-integration]

The three-way check is intentionally permissive for backward compatibility. External code or plugins may emit LLMStreamChunkEvent without stamping run_id. Tightening to a two-condition check would cause those un-stamped events to be silently dropped by all handlers, breaking backward compatibility.

The second commit (0a22307) also addresses the "context var overwrite" concern (comment 5) by reusing the parent run_id when one exists: run_id = _current_stream_run_id.get() or str(uuid.uuid4()). This ensures nested streaming (flow → crew) shares the same scope, and create_chunk_generator's contextvars.copy_context() correctly captures the run_id at iteration time.