"""Streaming intentional-silence suppression. When the agent chooses not to reply it emits a bare control marker (``NO_REPLY`` / ``[SILENT]`` / …). The gateway's whole-response filter (``gateway/response_filters.is_intentional_silence_agent_result``) suppresses this on the non-streaming delivery path, but the *streaming* path (``GatewayStreamConsumer``) previously had no silence awareness: it edited the raw marker onto the screen delta-by-delta and finalized it *before* the whole-response filter could run. On any streaming-capable adapter (Slack, Telegram, Discord, …) users saw a literal ``NO_REPLY`` bubble. These tests pin the two halves of the fix: * ``is_partial_silence_marker`` — the mid-stream hold-back predicate. * ``GatewayStreamConsumer`` — an exact-marker final buffer is suppressed and any already-shown preview is retracted, while substantive prose that merely mentions a marker is delivered normally. """ from __future__ import annotations from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock import pytest from gateway.response_filters import ( is_intentional_silence_response, is_partial_silence_marker, ) from gateway.stream_consumer import GatewayStreamConsumer, StreamConsumerConfig # -------------------------------------------------------------------------- # is_partial_silence_marker — mid-stream hold-back predicate # -------------------------------------------------------------------------- # Buffers that could still resolve to a marker → held back while streaming. PARTIAL_POSITIVE = [ "N", "NO", "NO_", "NO_REP", "NO_REPLY", # exact marker, not yet terminated by stream-end "NO REPLY", "no reply", # canonicalized (case/space-insensitive) " no_reply ", # surrounding whitespace stripped "[", "[SIL", "[SILENT]", "SILENT", "sil", ] # Buffers that have already diverged from every marker → stream normally. PARTIAL_NEGATIVE = [ "", " ", "No reply needed — here is the plan", # diverged past the marker "NO_REPLYING", # superset, not a prefix "Nope", "Hello there", "The NO_REPLY token means silence", # marker mentioned mid-prose "x" * 65, # over the 64-char cap "silence is golden", # 'SILENCE...' is not a marker prefix ] @pytest.mark.parametrize("text", PARTIAL_POSITIVE) def test_partial_silence_marker_positive(text): assert is_partial_silence_marker(text) is True @pytest.mark.parametrize("text", PARTIAL_NEGATIVE) def test_partial_silence_marker_negative(text): assert is_partial_silence_marker(text) is False def test_partial_silence_marker_none_safe(): assert is_partial_silence_marker(None) is False def test_partial_predicate_agrees_with_exact_on_full_markers(): """Every exact silence marker is also a (trivial) partial of itself.""" from gateway.response_filters import LIVE_GATEWAY_SILENT_MARKERS for marker in LIVE_GATEWAY_SILENT_MARKERS: assert is_partial_silence_marker(marker) is True assert is_intentional_silence_response(marker) is True # -------------------------------------------------------------------------- # GatewayStreamConsumer — end-to-end suppression through run() # -------------------------------------------------------------------------- def _make_adapter(*, supports_delete: bool = True) -> MagicMock: """Minimal MagicMock adapter wired for send/edit/delete.""" adapter = MagicMock() adapter.REQUIRES_EDIT_FINALIZE = False adapter.MAX_MESSAGE_LENGTH = 4096 adapter.send = AsyncMock(return_value=SimpleNamespace( success=True, message_id="preview_1", )) adapter.edit_message = AsyncMock(return_value=SimpleNamespace( success=True, message_id="preview_1", )) if supports_delete: adapter.delete_message = AsyncMock(return_value=True) else: del adapter.delete_message # type: ignore[attr-defined] return adapter def _sent_and_edited(adapter): texts = [] for call in adapter.send.call_args_list: texts.append(call.kwargs.get("content", "")) if getattr(adapter, "edit_message", None) is not None: for call in adapter.edit_message.call_args_list: texts.append(call.kwargs.get("content", "")) return texts class TestStreamedSilenceSuppression: @pytest.mark.asyncio async def test_no_reply_only_stream_is_fully_suppressed(self): """A stream whose entire content is NO_REPLY sends nothing visible.""" adapter = _make_adapter() consumer = GatewayStreamConsumer( adapter, "chat_1", StreamConsumerConfig(edit_interval=0.01, buffer_threshold=1), ) consumer.on_delta("NO_REPLY") consumer.finish() await consumer.run() # No marker text ever reached the platform. for text in _sent_and_edited(adapter): assert "NO_REPLY" not in text, f"marker leaked: {text!r}" # Delivery flags stay False so the gateway does not treat the marker # as a delivered reply (its whole-response filter then drops it too). assert consumer.final_response_sent is False assert consumer.final_content_delivered is False assert consumer.already_sent is False @pytest.mark.asyncio async def test_partial_marker_preview_is_retracted(self): """A marker flushed mid-stream as a preview is deleted on completion.""" adapter = _make_adapter() consumer = GatewayStreamConsumer( adapter, "chat_1", StreamConsumerConfig(edit_interval=0.01, buffer_threshold=1), ) # Force a mid-stream preview: pretend "NO_REPLY" was already put on # screen (the pre-fix behaviour) before got_done runs. consumer._message_id = "preview_1" consumer._preview_message_ids = {"preview_1"} consumer._already_sent = True consumer.on_delta("NO_REPLY") consumer.finish() await consumer.run() # The stale preview was best-effort deleted. adapter.delete_message.assert_awaited_once_with("chat_1", "preview_1") assert consumer.final_content_delivered is False assert consumer.already_sent is False @pytest.mark.asyncio async def test_suppression_without_delete_support_is_best_effort(self): """Adapter lacking delete_message still suppresses (leaves no new send).""" adapter = _make_adapter(supports_delete=False) consumer = GatewayStreamConsumer( adapter, "chat_1", StreamConsumerConfig(edit_interval=0.01, buffer_threshold=1), ) consumer.on_delta("NO_REPLY") consumer.finish() await consumer.run() for text in _sent_and_edited(adapter): assert "NO_REPLY" not in text assert consumer.final_content_delivered is False @pytest.mark.asyncio async def test_bracket_silent_marker_suppressed(self): """The [SILENT] marker is suppressed just like NO_REPLY.""" adapter = _make_adapter() consumer = GatewayStreamConsumer( adapter, "chat_1", StreamConsumerConfig(edit_interval=0.01, buffer_threshold=1), ) consumer.on_delta("[SILENT]") consumer.finish() await consumer.run() for text in _sent_and_edited(adapter): assert "[SILENT]" not in text assert consumer.final_content_delivered is False @pytest.mark.asyncio async def test_prose_mentioning_marker_is_delivered(self): """Substantive prose that merely mentions NO_REPLY is NOT suppressed.""" adapter = _make_adapter() consumer = GatewayStreamConsumer( adapter, "chat_1", StreamConsumerConfig(edit_interval=0.01, buffer_threshold=5), ) body = "The NO_REPLY token tells the gateway to stay silent." consumer.on_delta(body) consumer.finish() await consumer.run() delivered = "".join(_sent_and_edited(adapter)) assert "NO_REPLY" in delivered assert consumer.final_content_delivered is True @pytest.mark.asyncio async def test_marker_prefix_then_prose_is_delivered(self): """A reply that starts marker-like but continues is delivered whole. "NO REPLY needed …" passes through the mid-stream hold-back while the buffer is still a marker prefix, then flushes normally once it diverges. The final text is NOT an exact marker, so got_done does not suppress it. """ adapter = _make_adapter() consumer = GatewayStreamConsumer( adapter, "chat_1", StreamConsumerConfig(edit_interval=0.01, buffer_threshold=1), ) consumer.on_delta("NO REPLY") consumer.on_delta(" needed — the build is already green.") consumer.finish() await consumer.run() delivered = "".join(_sent_and_edited(adapter)) assert "the build is already green" in delivered assert consumer.final_content_delivered is True