diff --git a/gateway/stream_consumer.py b/gateway/stream_consumer.py index 2ceb0fb1d815e..0790c4d8091ef 100644 --- a/gateway/stream_consumer.py +++ b/gateway/stream_consumer.py @@ -77,8 +77,19 @@ def already_sent(self) -> bool: return self._already_sent def on_delta(self, text: str) -> None: - """Thread-safe callback — called from the agent's worker thread.""" - if text: + """Thread-safe callback — called from the agent's worker thread. + + Passing None signals an iteration boundary (the agent finished + streaming one segment and is about to resume after tool calls). + A newline separator is injected so resumed text starts on a new line + instead of being concatenated directly onto the previous segment. + """ + if text is None: + # Iteration boundary: inject a separator so the next streamed + # segment starts on a new line rather than running together. + if self._accumulated and not self._accumulated.endswith(chr(10)): + self._queue.put(chr(10)) + elif text: self._queue.put(text) def finish(self) -> None: diff --git a/tests/gateway/test_stream_consumer.py b/tests/gateway/test_stream_consumer.py new file mode 100644 index 0000000000000..0bff64057c378 --- /dev/null +++ b/tests/gateway/test_stream_consumer.py @@ -0,0 +1,79 @@ +"""Tests for GatewayStreamConsumer iteration boundary behaviour. + +Covers the bug where resumed text after tool calls was concatenated +directly onto prior streamed content with no separator (issue #2177). +""" + +import queue +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from gateway.stream_consumer import GatewayStreamConsumer, StreamConsumerConfig, _DONE + + +def _make_consumer(accumulated=""): + adapter = MagicMock() + adapter.MAX_MESSAGE_LENGTH = 4096 + adapter.send = AsyncMock(return_value=MagicMock(success=True, message_id="msg_1")) + adapter.edit_message = AsyncMock(return_value=MagicMock(success=True)) + consumer = GatewayStreamConsumer( + adapter=adapter, + chat_id="chat_123", + config=StreamConsumerConfig(edit_interval=0.0, buffer_threshold=1), + ) + consumer._accumulated = accumulated + return consumer + + +class TestOnDeltaBasic: + def test_text_enqueued(self): + consumer = _make_consumer() + consumer.on_delta("hello") + assert consumer._queue.get_nowait() == "hello" + + def test_empty_string_not_enqueued(self): + consumer = _make_consumer() + consumer.on_delta("") + assert consumer._queue.empty() + + def test_none_no_separator_when_empty(self): + consumer = _make_consumer(accumulated="") + consumer.on_delta(None) + assert consumer._queue.empty() + + +class TestOnDeltaBoundary: + def test_none_injects_newline_when_no_trailing_newline(self): + """Core regression test for issue #2177.""" + consumer = _make_consumer(accumulated="Let me check the issues.") + consumer.on_delta(None) + item = consumer._queue.get_nowait() + assert item == "\n", f"Expected newline separator, got {item!r}" + + def test_none_no_newline_when_already_ends_with_newline(self): + consumer = _make_consumer(accumulated="Let me check.\n") + consumer.on_delta(None) + assert consumer._queue.empty() + + def test_resumed_text_follows_separator(self): + consumer = _make_consumer(accumulated="Let me check the issues.") + consumer.on_delta(None) + consumer.on_delta("Found 3 issues.") + newline = consumer._queue.get_nowait() + resumed = consumer._queue.get_nowait() + assert newline == "\n" + assert resumed == "Found 3 issues." + + +class TestFinish: + def test_finish_puts_done_sentinel(self): + consumer = _make_consumer() + consumer.finish() + assert consumer._queue.get_nowait() is _DONE + + +class TestAlreadySent: + def test_initially_false(self): + consumer = _make_consumer() + assert consumer.already_sent is False