From 66fc54421cc43c240a4200f55cf4e35d04b7e185 Mon Sep 17 00:00:00 2001 From: axelray-dev <110029405+axelray-dev@users.noreply.github.com> Date: Fri, 26 Jun 2026 03:42:44 +0800 Subject: [PATCH] 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. --- nanobot/channels/manager.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/nanobot/channels/manager.py b/nanobot/channels/manager.py index 9eff648e..3294afc0 100644 --- a/nanobot/channels/manager.py +++ b/nanobot/channels/manager.py @@ -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")