diff --git a/gateway/response_filters.py b/gateway/response_filters.py new file mode 100644 index 000000000000..ad725d439dbf --- /dev/null +++ b/gateway/response_filters.py @@ -0,0 +1,69 @@ +from __future__ import annotations + +import re +from typing import Optional + +LIVE_GATEWAY_SILENT_MARKERS = frozenset( + { + "[silent]", + "silent", + "no message", + "no reply", + "no response", + "no response generated", + "empty", + } +) + + +def _unwrap_live_gateway_response_text(text: str) -> str: + normalized = text + for _ in range(6): + updated = normalized.strip() + changed = False + + for wrapper in ("**", "__", "~~", "`"): + if updated.startswith(wrapper) and updated.endswith(wrapper): + inner = updated[len(wrapper) : -len(wrapper)].strip() + if inner: + normalized = inner + changed = True + break + if changed: + continue + + for left, right in (("(", ")"), ("[", "]"), ("{", "}"), ('"', '"'), ("'", "'")): + if updated.startswith(left) and updated.endswith(right): + inner = updated[len(left) : -len(right)].strip() + if inner: + normalized = inner + changed = True + break + + if not changed: + normalized = updated + break + + return normalized + + +def _canonicalize_live_gateway_response(text: str) -> str: + normalized = _unwrap_live_gateway_response_text(text) + return re.sub(r"[\s\-_]+", " ", normalized).strip(" .!?:;").casefold() + + +def normalize_live_gateway_response( + response: Optional[str], *, failed: bool = False +) -> str: + """Suppress placeholder silence markers before live message delivery.""" + if response is None: + return "" + + text = str(response).strip() + if not text or failed: + return text + + if _canonicalize_live_gateway_response(text) in LIVE_GATEWAY_SILENT_MARKERS: + return "" + + return text diff --git a/gateway/run.py b/gateway/run.py index d137d73c3737..b6d16212364c 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -28,6 +28,8 @@ from datetime import datetime from typing import Dict, Optional, Any, List +from gateway.response_filters import normalize_live_gateway_response + # --------------------------------------------------------------------------- # SSL certificate auto-detection for NixOS and other non-standard systems. # Must run BEFORE any HTTP library (discord, aiohttp, etc.) is imported. @@ -3775,7 +3777,10 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str): except Exception: pass - response = agent_result.get("final_response") or "" + response = normalize_live_gateway_response( + agent_result.get("final_response"), + failed=bool(agent_result.get("failed")), + ) agent_messages = agent_result.get("messages", []) _response_time = time.time() - _msg_start_time _api_calls = agent_result.get("api_calls", 0) @@ -8330,6 +8335,7 @@ def _interim_assistant_cb(text: str, *, already_streamed: bool = False) -> None: else: _stream_consumer.on_commentary(text) return + text = normalize_live_gateway_response(text) if already_streamed or not _status_adapter or not str(text or "").strip(): return try: @@ -9139,7 +9145,10 @@ async def _notify_long_running(): ) ) ) - first_response = result.get("final_response", "") + first_response = normalize_live_gateway_response( + result.get("final_response"), + failed=bool(result.get("failed")), + ) if first_response and not _already_streamed: try: await adapter.send( diff --git a/gateway/stream_consumer.py b/gateway/stream_consumer.py index e6d96c802d29..c9b2e07b3cff 100644 --- a/gateway/stream_consumer.py +++ b/gateway/stream_consumer.py @@ -23,6 +23,8 @@ from dataclasses import dataclass from typing import Any, Optional +from gateway.response_filters import normalize_live_gateway_response + logger = logging.getLogger("gateway.stream_consumer") # Sentinel to signal the stream is complete @@ -444,12 +446,31 @@ def _clean_for_display(text: str) -> str: # Strip trailing whitespace/newlines but preserve leading content return cleaned.rstrip() + def _prepare_for_delivery(self, text: str) -> str: + cleaned = self._clean_for_display(text) + if not cleaned: + return cleaned + + cursor = self.cfg.cursor or "" + if cursor and cleaned.strip() == cursor.strip(): + return cleaned + + if cursor and cleaned.endswith(cursor): + body = cleaned[: -len(cursor)].rstrip() + if body and not normalize_live_gateway_response(body): + return "" + + if not normalize_live_gateway_response(cleaned): + return "" + + return cleaned + async def _send_new_chunk(self, text: str, reply_to_id: Optional[str]) -> Optional[str]: """Send a new message chunk, optionally threaded to a previous message. Returns the message_id so callers can thread subsequent chunks. """ - text = self._clean_for_display(text) + text = self._prepare_for_delivery(text) if not text.strip(): return reply_to_id try: @@ -508,7 +529,7 @@ async def _send_fallback_final(self, text: str) -> None: Retries each chunk once on flood-control failures with a short delay. """ - final_text = self._clean_for_display(text) + final_text = self._prepare_for_delivery(text) continuation = self._continuation_text(final_text) self._fallback_final_send = False if not continuation.strip(): @@ -600,7 +621,7 @@ async def _try_strip_cursor(self) -> None: async def _send_commentary(self, text: str) -> bool: """Send a completed interim assistant commentary message.""" - text = self._clean_for_display(text) + text = self._prepare_for_delivery(text) if not text.strip(): return False try: @@ -626,7 +647,7 @@ async def _send_or_edit(self, text: str) -> bool: # Strip MEDIA: directives so they don't appear as visible text. # Media files are delivered as native attachments after the stream # finishes (via _deliver_media_from_response in gateway/run.py). - text = self._clean_for_display(text) + text = self._prepare_for_delivery(text) # A bare streaming cursor is not meaningful user-visible content and # can render as a stray tofu/white-box message on some clients. visible_without_cursor = text diff --git a/tests/gateway/test_live_silent_responses.py b/tests/gateway/test_live_silent_responses.py new file mode 100644 index 000000000000..333e88e91928 --- /dev/null +++ b/tests/gateway/test_live_silent_responses.py @@ -0,0 +1,93 @@ +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from gateway.config import Platform +from gateway.platforms.base import MessageEvent, MessageType +from gateway.response_filters import normalize_live_gateway_response +from gateway.run import GatewayRunner +from gateway.session import SessionSource + + +@pytest.mark.parametrize( + ("raw_text", "expected"), + [ + ("(No message)", ""), + ("[SILENT]", ""), + ("`(No reply)`", ""), + ("**(No response generated)**", ""), + ("(empty)", ""), + ("[SILENT] means stay quiet", "[SILENT] means stay quiet"), + ("No message received from Discord", "No message received from Discord"), + ], +) +def test_normalize_live_gateway_response(raw_text, expected): + assert normalize_live_gateway_response(raw_text) == expected + + +def test_normalize_live_gateway_response_preserves_failed_output(): + assert normalize_live_gateway_response("[SILENT]", failed=True) == "[SILENT]" + + +def _make_runner(): + runner = GatewayRunner.__new__(GatewayRunner) + runner.config = MagicMock() + runner.session_store = MagicMock() + runner.hooks = SimpleNamespace(emit=AsyncMock()) + runner.adapters = {} + runner._show_reasoning = False + runner._session_db = None + runner._set_session_env = MagicMock(return_value=[]) + runner._clear_session_env = MagicMock() + runner._should_send_voice_reply = MagicMock(return_value=False) + runner._deliver_media_from_response = AsyncMock() + return runner + + +@pytest.mark.asyncio +async def test_handle_message_with_agent_suppresses_placeholder(monkeypatch): + runner = _make_runner() + + session_entry = SimpleNamespace( + session_id="sess-1", + session_key="key-1", + created_at=1, + updated_at=2, + was_auto_reset=False, + last_prompt_tokens=0, + ) + history = [{"role": "assistant", "content": "Earlier reply"}] + + runner.session_store.get_or_create_session.return_value = session_entry + runner.session_store.load_transcript.return_value = history + runner.session_store.has_any_sessions.return_value = True + runner.session_store.append_to_transcript = MagicMock() + runner.session_store.update_session = MagicMock() + + runner._run_agent = AsyncMock( + return_value={ + "final_response": "(No message)", + "messages": history, + "api_calls": 1, + "last_prompt_tokens": 0, + } + ) + + monkeypatch.setattr("gateway.run.build_session_context", lambda *_a, **_kw: {}) + monkeypatch.setattr("gateway.run.build_session_context_prompt", lambda *_a, **_kw: "") + + source = SessionSource( + platform=Platform.LOCAL, + chat_id="chat-1", + user_id="user-1", + user_name="tester", + ) + event = MessageEvent(text="test", message_type=MessageType.TEXT, source=source) + + result = await runner._handle_message_with_agent(event, source, "key-1") + + assert result == "" + appended = [call.args[1] for call in runner.session_store.append_to_transcript.call_args_list] + assert any(entry["role"] == "user" for entry in appended) + assert not any(entry.get("content") == "(No message)" for entry in appended) diff --git a/tests/gateway/test_stream_consumer.py b/tests/gateway/test_stream_consumer.py index 38532e66be89..30f764751260 100644 --- a/tests/gateway/test_stream_consumer.py +++ b/tests/gateway/test_stream_consumer.py @@ -9,6 +9,16 @@ from gateway.stream_consumer import GatewayStreamConsumer, StreamConsumerConfig +async def _wait_for_mock_call_count(mock, count: int, *, timeout: float = 0.5) -> None: + deadline = asyncio.get_running_loop().time() + timeout + while mock.call_count < count and asyncio.get_running_loop().time() < deadline: + await asyncio.sleep(0.01) + + assert mock.call_count >= count, ( + f"Timed out waiting for {mock!r} to reach {count} calls; got {mock.call_count}" + ) + + # ── _clean_for_display unit tests ──────────────────────────────────────── @@ -239,6 +249,47 @@ async def test_short_text_cursor_edit_existing_message_allowed(self): assert result is True adapter.edit_message.assert_called_once() + @pytest.mark.asyncio + async def test_silent_placeholder_skips_send(self): + """Silent placeholder text should not be streamed to the platform.""" + adapter = MagicMock() + adapter.send = AsyncMock() + adapter.MAX_MESSAGE_LENGTH = 4096 + + consumer = GatewayStreamConsumer(adapter, "chat_123") + await consumer._send_or_edit("(No message)") + + adapter.send.assert_not_called() + + @pytest.mark.asyncio + async def test_silent_placeholder_with_cursor_skips_send(self): + """Cursor updates should still suppress exact silence markers.""" + adapter = MagicMock() + adapter.send = AsyncMock() + adapter.MAX_MESSAGE_LENGTH = 4096 + + consumer = GatewayStreamConsumer( + adapter, + "chat_123", + StreamConsumerConfig(cursor=" ..."), + ) + await consumer._send_or_edit("(No message) ...") + + adapter.send.assert_not_called() + + @pytest.mark.asyncio + async def test_commentary_silent_placeholder_skips_send(self): + """Completed commentary placeholders should also stay hidden.""" + adapter = MagicMock() + adapter.send = AsyncMock() + adapter.MAX_MESSAGE_LENGTH = 4096 + + consumer = GatewayStreamConsumer(adapter, "chat_123") + sent = await consumer._send_commentary("[SILENT]") + + assert sent is False + adapter.send.assert_not_called() + # ── Integration: full stream run ───────────────────────────────────────── @@ -278,6 +329,27 @@ async def test_stream_with_media_tag(self): assert consumer.already_sent + @pytest.mark.asyncio + async def test_stream_with_silent_placeholder_sends_nothing(self): + """A final streamed silence marker should produce no visible message.""" + adapter = MagicMock() + adapter.send = AsyncMock() + adapter.edit_message = AsyncMock() + adapter.MAX_MESSAGE_LENGTH = 4096 + + config = StreamConsumerConfig(edit_interval=0.01, buffer_threshold=5) + consumer = GatewayStreamConsumer(adapter, "chat_123", config) + + consumer.on_delta("(No message)") + consumer.finish() + + await consumer.run() + + adapter.send.assert_not_called() + adapter.edit_message.assert_not_called() + assert consumer.already_sent is False + assert consumer.final_response_sent is True + # ── Segment break (tool boundary) tests ────────────────────────────────── @@ -439,10 +511,11 @@ async def test_edit_failure_sends_only_unsent_tail_at_finish(self): config = StreamConsumerConfig(edit_interval=0.01, buffer_threshold=5, cursor=" ▉") consumer = GatewayStreamConsumer(adapter, "chat_123", config) + consumer._MAX_FLOOD_STRIKES = 1 consumer.on_delta("Hello") task = asyncio.create_task(consumer.run()) - await asyncio.sleep(0.08) + await _wait_for_mock_call_count(adapter.send, 1) consumer.on_delta(" world") await asyncio.sleep(0.08) consumer.finish() @@ -469,10 +542,11 @@ async def test_segment_break_clears_failed_edit_fallback_state(self): config = StreamConsumerConfig(edit_interval=0.01, buffer_threshold=5, cursor=" ▉") consumer = GatewayStreamConsumer(adapter, "chat_123", config) + consumer._MAX_FLOOD_STRIKES = 1 consumer.on_delta("Hello") task = asyncio.create_task(consumer.run()) - await asyncio.sleep(0.08) + await _wait_for_mock_call_count(adapter.send, 1) consumer.on_delta(" world") await asyncio.sleep(0.08) consumer.on_delta(None) @@ -502,7 +576,7 @@ async def test_no_message_id_enters_fallback_mode(self): consumer.on_delta("Hello") task = asyncio.create_task(consumer.run()) - await asyncio.sleep(0.08) + await _wait_for_mock_call_count(adapter.send, 1) consumer.on_delta(" world, this is a longer response.") await asyncio.sleep(0.08) consumer.finish() @@ -590,12 +664,13 @@ async def test_fallback_final_splits_long_continuation_without_dropping_text(sel config = StreamConsumerConfig(edit_interval=0.01, buffer_threshold=5, cursor=" ▉") consumer = GatewayStreamConsumer(adapter, "chat_123", config) + consumer._MAX_FLOOD_STRIKES = 1 prefix = "Hello world" tail = "x" * 620 consumer.on_delta(prefix) task = asyncio.create_task(consumer.run()) - await asyncio.sleep(0.08) + await _wait_for_mock_call_count(adapter.send, 1) consumer.on_delta(tail) await asyncio.sleep(0.08) consumer.finish() @@ -673,7 +748,7 @@ async def test_success_without_message_id_marks_visible_and_sends_only_tail(self consumer.on_delta("Hello") task = asyncio.create_task(consumer.run()) - await asyncio.sleep(0.08) + await _wait_for_mock_call_count(adapter.send, 1) consumer.on_delta(" world") await asyncio.sleep(0.08) consumer.finish()