fix: include _stream_id in stream delta coalescing key (#4063)

ChannelManager coalesces _stream_delta messages by (channel, chat_id)
only. Overlapping streams in the same chat can be merged incorrectly
because deltas from distinct _stream_id values share one buffer.

Include _stream_id in the coalescing key so distinct streams in the
same channel and chat are delivered separately.
This commit is contained in:
axelray-dev
2026-06-27 16:47:53 +08:00
committed by Xubin Ren
parent 6a27c26257
commit 66fc54421c
+3 -3
View File
@@ -396,7 +396,7 @@ class ChannelManager:
def _coalesce_stream_deltas(
self, first_msg: OutboundMessage
) -> tuple[OutboundMessage, list[OutboundMessage]]:
"""Merge consecutive _stream_delta messages for the same (channel, chat_id).
"""Merge consecutive _stream_delta messages for the same (channel, chat_id, _stream_id).
This reduces the number of API calls when the queue has accumulated multiple
deltas, which happens when LLM generates faster than the channel can process.
@@ -404,7 +404,7 @@ class ChannelManager:
Returns:
tuple of (merged_message, list_of_non_matching_messages)
"""
target_key = (first_msg.channel, first_msg.chat_id)
target_key = (first_msg.channel, first_msg.chat_id, first_msg.metadata.get("_stream_id"))
combined_content = first_msg.content
final_metadata = dict(first_msg.metadata or {})
non_matching: list[OutboundMessage] = []
@@ -418,7 +418,7 @@ class ChannelManager:
break
# Check if this message belongs to the same stream
same_target = (next_msg.channel, next_msg.chat_id) == target_key
same_target = (next_msg.channel, next_msg.chat_id, next_msg.metadata.get("_stream_id") if next_msg.metadata else None) == target_key
is_delta = next_msg.metadata and next_msg.metadata.get("_stream_delta")
is_end = next_msg.metadata and next_msg.metadata.get("_stream_end")