diff --git a/nanobot/channels/manager.py b/nanobot/channels/manager.py index 3294afc0..23a342f1 100644 --- a/nanobot/channels/manager.py +++ b/nanobot/channels/manager.py @@ -404,7 +404,8 @@ class ChannelManager: Returns: tuple of (merged_message, list_of_non_matching_messages) """ - target_key = (first_msg.channel, first_msg.chat_id, first_msg.metadata.get("_stream_id")) + first_metadata = first_msg.metadata or {} + target_key = (first_msg.channel, first_msg.chat_id, first_metadata.get("_stream_id")) combined_content = first_msg.content final_metadata = dict(first_msg.metadata or {}) non_matching: list[OutboundMessage] = [] @@ -418,9 +419,14 @@ class ChannelManager: break # Check if this message belongs to the same stream - 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") + next_metadata = next_msg.metadata or {} + same_target = ( + next_msg.channel, + next_msg.chat_id, + next_metadata.get("_stream_id"), + ) == target_key + is_delta = next_metadata.get("_stream_delta") + is_end = next_metadata.get("_stream_end") if same_target and is_delta and not final_metadata.get("_stream_end"): # Accumulate content diff --git a/tests/channels/test_channel_manager_delta_coalescing.py b/tests/channels/test_channel_manager_delta_coalescing.py index adec72e7..e3ed8347 100644 --- a/tests/channels/test_channel_manager_delta_coalescing.py +++ b/tests/channels/test_channel_manager_delta_coalescing.py @@ -142,6 +142,31 @@ class TestDeltaCoalescing: assert pending[0].chat_id == "chat2" assert pending[0].content == "World" + @pytest.mark.asyncio + async def test_deltas_different_stream_ids_not_coalesced(self, manager, bus): + """Deltas for the same chat but different streams should not be merged.""" + await bus.publish_outbound(OutboundMessage( + channel="mock", + chat_id="chat1", + content="A1", + metadata={"_stream_delta": True, "_stream_id": "stream-a"}, + )) + await bus.publish_outbound(OutboundMessage( + channel="mock", + chat_id="chat1", + content="B1", + metadata={"_stream_delta": True, "_stream_id": "stream-b"}, + )) + + first_msg = await bus.consume_outbound() + merged, pending = manager._coalesce_stream_deltas(first_msg) + + assert merged.content == "A1" + assert merged.metadata.get("_stream_id") == "stream-a" + assert len(pending) == 1 + assert pending[0].content == "B1" + assert pending[0].metadata.get("_stream_id") == "stream-b" + @pytest.mark.asyncio async def test_stream_end_terminates_coalescing(self, manager, bus): """_stream_end should stop coalescing and be included in final message."""