From b59c1b724b79b9f352e30c3047f16f3ca3ac0b1d Mon Sep 17 00:00:00 2001 From: Jetha Chan Date: Sun, 19 Jul 2026 10:12:07 +0900 Subject: [PATCH 1/6] test(line): regression tests for typed media-cache routing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Rebase note: upstream 73e193c03 ("fix(line): normalize inbound media types and cache routing") independently shipped a superset of this branch's adapter fix — typed cache helper dispatch, fileName threading, and a (path, media_type) return from _download_media. The adapter changes are therefore dropped in favor of upstream's version. What remains from the original commit: the mocked regression tests covering _download_media routing for all four LINE content types, cache/fetch failure fallbacks, and fileName threading from the message event — adapted to upstream's keyword-only ``filename=`` parameter and tuple return, and to media_types now carrying MIME types. Co-Authored-By: Claude Fable 5 --- tests/gateway/test_line_plugin.py | 106 ++++++++++++++++++++++++++++++ 1 file changed, 106 insertions(+) diff --git a/tests/gateway/test_line_plugin.py b/tests/gateway/test_line_plugin.py index e59bd8286e97f..ae59c0aa281f9 100644 --- a/tests/gateway/test_line_plugin.py +++ b/tests/gateway/test_line_plugin.py @@ -507,3 +507,109 @@ def test_send_image_blocked_without_public_url(self, monkeypatch, tmp_path): assert not result.success assert "LINE_PUBLIC_URL" in (result.error or "") +class TestDownloadMediaRouting: + """``_download_media`` must dispatch each LINE content type to its typed + cache helper. Regression guard for the old code that pushed audio, video, + and file payloads through ``cache_image_from_bytes``, which rejected them + as non-image data and silently dropped the media.""" + + @pytest.fixture + def adapter(self, monkeypatch): + monkeypatch.delenv("LINE_CHANNEL_ACCESS_TOKEN", raising=False) + monkeypatch.delenv("LINE_CHANNEL_SECRET", raising=False) + from gateway.config import PlatformConfig + cfg = PlatformConfig(enabled=True, extra={ + "channel_access_token": "tok", + "channel_secret": "sec", + }) + ad = LineAdapter(cfg) + ad._client = MagicMock() + ad._client.fetch_content = AsyncMock(return_value=b"payload-bytes") + ad._client.loading = AsyncMock() + return ad + + @pytest.fixture + def cache_mocks(self, monkeypatch): + """Mock the typed cache helpers in the adapter's namespace; no disk IO.""" + mocks = { + "image": MagicMock(return_value="/cache/images/img_abc.jpg"), + "audio": MagicMock(return_value="/cache/audio/audio_abc.m4a"), + "video": MagicMock(return_value="/cache/videos/video_abc.mp4"), + "document": MagicMock(return_value="/cache/documents/doc_abc_x.bin"), + } + monkeypatch.setattr(_line, "cache_image_from_bytes", mocks["image"]) + monkeypatch.setattr(_line, "cache_audio_from_bytes", mocks["audio"]) + monkeypatch.setattr(_line, "cache_video_from_bytes", mocks["video"]) + monkeypatch.setattr(_line, "cache_document_from_bytes", mocks["document"]) + return mocks + + def _assert_only(self, cache_mocks, kind): + for other in set(cache_mocks) - {kind}: + cache_mocks[other].assert_not_called() + + def test_image_routes_to_image_cache(self, adapter, cache_mocks): + path, media_type = asyncio.run(adapter._download_media("mid-1", "image")) + assert path == "/cache/images/img_abc.jpg" + assert media_type == "image/jpeg" + cache_mocks["image"].assert_called_once_with(b"payload-bytes", ext=".jpg") + self._assert_only(cache_mocks, "image") + + def test_audio_routes_to_audio_cache(self, adapter, cache_mocks): + path, media_type = asyncio.run(adapter._download_media("mid-2", "audio")) + assert path == "/cache/audio/audio_abc.m4a" + # Exact subtype is platform mimetypes-dependent; family must be audio. + assert media_type.startswith("audio/") + cache_mocks["audio"].assert_called_once_with(b"payload-bytes", ext=".m4a") + self._assert_only(cache_mocks, "audio") + + def test_video_routes_to_video_cache(self, adapter, cache_mocks): + path, media_type = asyncio.run(adapter._download_media("mid-3", "video")) + assert path == "/cache/videos/video_abc.mp4" + assert media_type.startswith("video/") + cache_mocks["video"].assert_called_once_with(b"payload-bytes", ext=".mp4") + self._assert_only(cache_mocks, "video") + + def test_file_routes_to_document_cache_with_filename(self, adapter, cache_mocks): + path, media_type = asyncio.run( + adapter._download_media("mid-4", "file", filename="report.pdf") + ) + assert path == "/cache/documents/doc_abc_x.bin" + assert media_type == "application/pdf" + cache_mocks["document"].assert_called_once_with(b"payload-bytes", "report.pdf") + self._assert_only(cache_mocks, "document") + + def test_file_without_name_uses_fallback_filename(self, adapter, cache_mocks): + asyncio.run(adapter._download_media("mid-5", "file")) + cache_mocks["document"].assert_called_once_with( + b"payload-bytes", "line_file.bin" + ) + + def test_cache_failure_returns_none(self, adapter, cache_mocks): + cache_mocks["audio"].side_effect = ValueError("audio exceeds size limit") + assert asyncio.run(adapter._download_media("mid-6", "audio")) == (None, "") + + def test_fetch_failure_returns_none_without_caching(self, adapter, cache_mocks): + adapter._client.fetch_content = AsyncMock(side_effect=RuntimeError("boom")) + assert asyncio.run(adapter._download_media("mid-7", "image")) == (None, "") + for mock in cache_mocks.values(): + mock.assert_not_called() + + def test_message_event_threads_file_name_to_document_cache( + self, adapter, cache_mocks + ): + adapter.handle_message = AsyncMock() + event = { + "message": {"type": "file", "id": "mid-8", "fileName": "minutes.docx"}, + "source": {"type": "user", "userId": "U1"}, + "replyToken": "rt-1", + } + asyncio.run(adapter._handle_message_event(event)) + cache_mocks["document"].assert_called_once_with( + b"payload-bytes", "minutes.docx" + ) + event_obj = adapter.handle_message.call_args.args[0] + assert event_obj.media_urls == ["/cache/documents/doc_abc_x.bin"] + assert event_obj.media_types == [ + "application/vnd.openxmlformats-officedocument" + ".wordprocessingml.document" + ] From 054e4eb4b1d8d2777f821023b9f3799a96b5e94f Mon Sep 17 00:00:00 2001 From: Jetha Chan Date: Sun, 19 Jul 2026 10:12:07 +0900 Subject: [PATCH 2/6] chore(release): map LINE PR contributor email Map the contributor email used on the LINE media PR stack to the jethac GitHub handle, replacing the stale legacy-map entry. Co-Authored-By: Claude Fable 5 --- scripts/release.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/scripts/release.py b/scripts/release.py index 541ca2eefc3a8..ffab0981b0a57 100755 --- a/scripts/release.py +++ b/scripts/release.py @@ -282,6 +282,7 @@ "minz0721@outlook.com": "s010mn", # PR #29221 salvage (ollama-cloud reasoning_effort xhigh→max) "128256017+chriswesley4@users.noreply.github.com": "chriswesley4", # PR #53185 salvage (re-enable titleBarOverlay on plain Linux; missing min/max/close regression) "rafael.millan@gmail.com": "RafaelMiMi", # PR #42229 salvage (no-sandbox fallback for AppArmor-restricted Linux desktop launch) + "jethachan@gmail.com": "jethac", # PR #35785/#40931/#40933 (LINE media and reply modality stack) "jeevesassistant00@gmail.com": "jeeves-assistant", # PR #50771 (computer-use CuaDriver vision capture routing) "21178861+ScotterMonk@users.noreply.github.com": "ScotterMonk", # PR #50145 salvage (cron output truncation: adapter-aware chunking, #50126) "rrandqua@gmail.com": "TutkuEroglu", # PR #50481 salvage (AGENTS.md stale token-lock adapter path) @@ -862,7 +863,6 @@ "lazycat.manatee@gmail.com": "manateelazycat", "bzarnitz13@gmail.com": "Beandon13", "tony@tonysimons.dev": "asimons81", - "jetha@google.com": "jethac", "vishal.dharm@gmail.com": "vishal-dharm", "jani@0xhoneyjar.xyz": "deep-name", # LINE messaging plugin (synthesis PR) From a934b25986c8c33f9b8b05d85f6e428520b97c97 Mon Sep 17 00:00:00 2001 From: Jetha Chan Date: Sun, 7 Jun 2026 11:27:47 +0900 Subject: [PATCH 3/6] refactor(gateway): add adapter reply delivery policy MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Rebase note (onto f75b577b9): upstream's voice.auto_tts sync (_should_auto_tts_for_chat, "voice accompanies text replies unless the chat explicitly set off", with unset voice_mode None distinct from "off") and the #60671 streaming-TTS skip landed in these same paths since this branch's base. Adapted the graft to preserve both: the base adapter's default reply_delivery_policy and the runner's legacy fallback now include the auto_tts term, _reply_delivery_policy passes voice_mode through raw (None when unset) instead of defaulting to "off", and the runner call site keeps upstream's streaming-TTS guard while threading _voice_reply_sent/suppress-text through it. Co-Authored-By: Claude Fable 5 [rebase note 2026-08-01: re-resolved against upstream e444d1658 — adopted upstream's tightened voice.auto_tts precedence (fallback only when chat has no explicit voice mode, `voice_mode is None`, upstream changed from `!= "off"`) in both the base adapter's default reply_delivery_policy and the runner's legacy fallback] --- gateway/platforms/base.py | 61 +++++++++ gateway/run.py | 144 +++++++++++++------- tests/gateway/test_reply_delivery_policy.py | 125 +++++++++++++++++ 3 files changed, 284 insertions(+), 46 deletions(-) create mode 100644 tests/gateway/test_reply_delivery_policy.py diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index f224aab323fcd..1a0ee1c912d11 100644 --- a/gateway/platforms/base.py +++ b/gateway/platforms/base.py @@ -2407,6 +2407,14 @@ def get_command_args(self) -> str: return args +@dataclass +class ReplyDeliveryPolicy: + """Adapter-provided delivery policy for a completed assistant reply.""" + + send_voice_reply: bool = False + suppress_text_if_voice_reply_sent: bool = False + + @dataclass class TextDebounceState: event: MessageEvent @@ -3253,6 +3261,59 @@ def streaming_overflow_limit(self) -> Optional[int]: """ return None + def observe_inbound_message(self, event: MessageEvent) -> None: + """Observe inbound messages before gateway dispatch. + + Platform adapters can override this to maintain lightweight + conversation state used by later delivery decisions. The default is a + no-op so existing adapters keep their behavior unchanged. + """ + return None + + def reply_delivery_policy( + self, + event: MessageEvent, + response: str, + *, + voice_mode: Optional[str], + already_sent: bool, + ) -> ReplyDeliveryPolicy: + """Return how the gateway should deliver the final assistant reply. + + The default preserves the runner's legacy auto-voice behavior: + explicit ``/voice all`` or ``/voice voice_only`` opt-ins request + runner-side TTS, ``voice.auto_tts`` (synced into the adapter on + gateway startup via ``_should_auto_tts_for_chat``) is the fallback + only when the chat has no explicit mode — otherwise the chat-level + all/voice_only/off choice takes precedence — and voice-input turns + are skipped when the adapter's own post-processing can still + auto-TTS the text response. + + ``voice_mode`` is ``None`` when the chat never set a mode — distinct + from an explicit ``"off"``, which disables the auto_tts path. + """ + if not response or response.startswith("Error:"): + return ReplyDeliveryPolicy() + + is_voice_input = event.message_type == MessageType.VOICE + auto_tts = False + if hasattr(self, "_should_auto_tts_for_chat"): + try: + auto_tts = bool(self._should_auto_tts_for_chat(event.source.chat_id)) + except Exception: + auto_tts = False + send_voice = ( + voice_mode == "all" + or (voice_mode == "voice_only" and is_voice_input) + # The base adapter's own auto-TTS path only covers voice-input + # replies, so final text replies need the runner path here. + # Fallback only when the chat never set an explicit mode. + or (voice_mode is None and auto_tts) + ) + if is_voice_input and not already_sent: + send_voice = False + return ReplyDeliveryPolicy(send_voice_reply=send_voice) + async def send_draft( self, chat_id: str, diff --git a/gateway/run.py b/gateway/run.py index 11e6fd7f0fbfb..a33931e8d047e 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -2467,6 +2467,7 @@ def _platform_has_bot_credential(platform: "Platform", platform_config: "Platfor EphemeralReply, MessageEvent, MessageType, + ReplyDeliveryPolicy, _prefix_within_utf16_limit, _reply_anchor_for_event, build_auto_tts_output_path, @@ -14816,6 +14817,7 @@ async def _handle_message(self, event: MessageEvent) -> Optional[str]: 7. Return response """ source = event.source + self._observe_inbound_message(event) # 🔴 Cross-session leak guard. This handler runs inside a per-message # asyncio task created via create_task(), which snapshots the spawning @@ -18820,11 +18822,14 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g _stts_adapter is not None and bool(getattr(_stts_adapter, "_streaming_tts_turn_completed", lambda *_a, **_k: False)(session_key, run_generation)) ) + _voice_reply_sent = False if ( not _streaming_tts_done and self._should_send_voice_reply(event, response, agent_messages, already_sent=_already_sent) ): - await self._send_voice_reply(event, response) + _voice_reply_sent = await self._send_voice_reply(event, response) + if self._should_suppress_text_after_voice_reply(event, response, _voice_reply_sent, already_sent=_already_sent): + return None # If streaming already delivered the response, extract and # deliver any MEDIA: files before returning None. Streaming @@ -19887,53 +19892,79 @@ async def _handle_voice_channel_input( await adapter.handle_message(event) - def _should_send_voice_reply( + def _observe_inbound_message(self, event: MessageEvent) -> None: + """Let the source adapter observe an inbound event before dispatch.""" + adapter = self.adapters.get(event.source.platform) + if not adapter or not hasattr(adapter, "observe_inbound_message"): + return + try: + adapter.observe_inbound_message(event) + except Exception: + logger.debug( + "Adapter observe_inbound_message failed for %s", + getattr(event.source.platform, "value", event.source.platform), + exc_info=True, + ) + + def _reply_delivery_policy( self, event: MessageEvent, response: str, - agent_messages: list, + *, already_sent: bool = False, - ) -> bool: - """Decide whether the runner should send a TTS voice reply. - - Returns False when: - - voice_mode is off for this chat - - response is empty or an error - - agent already called text_to_speech tool (dedup) - - voice input and base adapter auto-TTS already handled it (skip_double) - UNLESS streaming already consumed the response (already_sent=True), - in which case the base adapter won't have text for auto-TTS so the - runner must handle it. - """ - if not response or response.startswith("Error:"): - return False - + ): + """Return the adapter's reply delivery policy for this turn.""" + adapter = self.adapters.get(event.source.platform) chat_id = event.source.chat_id - voice_key = self._voice_key(event.source.platform, chat_id) - voice_mode = self._voice_mode.get(voice_key) - is_voice_input = (event.message_type == MessageType.VOICE) + # Raw get — ``None`` (chat never set a voice mode) is distinct from an + # explicit ``"off"``: the ``voice.auto_tts`` fallback below (and in the + # base adapter's default policy) stays eligible only for ``None``. + voice_mode = self._voice_mode.get(self._voice_key(event.source.platform, chat_id)) - adapter = self.adapters.get(event.source.platform) + if not response or response.startswith("Error:"): + return ReplyDeliveryPolicy() + + if adapter and hasattr(adapter, "reply_delivery_policy"): + policy = adapter.reply_delivery_policy( + event, + response, + voice_mode=voice_mode, + already_sent=already_sent, + ) + if isinstance(policy, ReplyDeliveryPolicy): + return policy + + # Legacy fallback for adapters without a policy hook. Mirrors the + # pre-policy inline logic, including the ``voice.auto_tts`` term + # (synced into the adapter on gateway startup): it is the fallback + # only when the chat has no explicit mode; otherwise the chat-level + # all/voice_only/off choice takes precedence. adapter_auto_tts = False if adapter and hasattr(adapter, "_should_auto_tts_for_chat"): try: adapter_auto_tts = bool(adapter._should_auto_tts_for_chat(chat_id)) except Exception: adapter_auto_tts = False - - should = ( - (voice_mode == "all") + is_voice_input = event.message_type == MessageType.VOICE + send_voice = ( + voice_mode == "all" or (voice_mode == "voice_only" and is_voice_input) - # ``voice.auto_tts`` is synced into the adapter on gateway startup. - # It is the fallback only when the chat has no explicit mode; - # otherwise the chat-level all/voice_only/off choice takes precedence. or (voice_mode is None and adapter_auto_tts) ) - if not should: - logger.debug( - "Auto voice reply skipped: mode=%s adapter_auto_tts=%s chat=%s platform=%s", - voice_mode, adapter_auto_tts, chat_id, event.source.platform.value, - ) + if is_voice_input and not already_sent: + send_voice = False + return ReplyDeliveryPolicy(send_voice_reply=send_voice) + + def _should_send_voice_reply( + self, + event: MessageEvent, + response: str, + agent_messages: list, + already_sent: bool = False, + ) -> bool: + """Decide whether the runner should send a TTS voice reply.""" + policy = self._reply_delivery_policy(event, response, already_sent=already_sent) + if not getattr(policy, "send_voice_reply", False): return False # Dedup: agent already called TTS tool in THIS turn only @@ -19953,21 +19984,27 @@ def _should_send_voice_reply( if has_agent_tts: return False - # Dedup: base adapter auto-TTS already handles voice input - # (play_tts plays in VC when connected, so runner can skip). - # When streaming already delivered the text (already_sent=True), - # the base adapter will receive None and can't run auto-TTS, - # so the runner must take over. - if is_voice_input and not already_sent: - return False - return True def _should_echo_stt_transcripts(self) -> bool: """Return whether inbound voice/STT transcripts should be echoed to chat.""" return bool(getattr(self.config, "stt_echo_transcripts", True)) - async def _send_voice_reply(self, event: MessageEvent, text: str) -> None: + def _should_suppress_text_after_voice_reply( + self, + event: MessageEvent, + response: str, + voice_reply_sent: bool, + *, + already_sent: bool = False, + ) -> bool: + """Return True when adapter policy wants voice to replace text.""" + if not voice_reply_sent: + return False + policy = self._reply_delivery_policy(event, response, already_sent=already_sent) + return bool(getattr(policy, "suppress_text_if_voice_reply_sent", False)) + + async def _send_voice_reply(self, event: MessageEvent, text: str) -> bool: """Generate TTS audio and send as a voice message before the text reply.""" audio_path = None actual_paths: List[str] = [] @@ -19976,7 +20013,7 @@ async def _send_voice_reply(self, event: MessageEvent, text: str) -> None: tts_text = _strip_markdown_for_tts(text) if not tts_text: - return + return False # Platform-aware output path: platforms whose native voice # bubbles require Ogg/Opus (OPUS_VOICE_PLATFORMS — Telegram, @@ -19992,7 +20029,7 @@ async def _send_voice_reply(self, event: MessageEvent, text: str) -> None: result = json.loads(result_json) except (json.JSONDecodeError, TypeError): logger.warning("Auto voice reply TTS returned invalid JSON: %s", result_json[:200] if result_json else result_json) - return + return False # Final delivery may be one combined file or multiple separately # valid files when combination is unavailable or would exceed a @@ -20006,7 +20043,7 @@ async def _send_voice_reply(self, event: MessageEvent, text: str) -> None: ] if not result.get("success") or not actual_paths: logger.warning("Auto voice reply TTS failed: %s", result.get("error")) - return + return False adapter = self._adapter_for_source(event.source) @@ -20036,10 +20073,21 @@ async def _send_voice_reply(self, event: MessageEvent, text: str) -> None: thread_meta["notify"] = True else: thread_meta = {"notify": True} + elif not in_voice_channel: + # Neither a live voice channel nor a send_voice hook: adapter + # can't deliver voice at all for this turn. + return False + + # Track per-file delivery so the bool return (added by this PR's + # reply-delivery-policy refactor) reflects whether the adapter + # actually accepted the voice reply, matching the multi-file + # split support carried over from upstream. + sent_any = False for actual_path in actual_paths: if in_voice_channel: play_voice = cast(Callable[..., Awaitable[Any]], play_in_voice_channel) await play_voice(guild_id, actual_path) + sent_any = True elif callable(send_voice): send_voice_call = cast(Callable[..., Awaitable[Any]], send_voice) send_kwargs: Dict[str, Any] = { @@ -20048,9 +20096,13 @@ async def _send_voice_reply(self, event: MessageEvent, text: str) -> None: "reply_to": reply_anchor, "metadata": thread_meta, } - await send_voice_call(**send_kwargs) + send_result = await send_voice_call(**send_kwargs) + if send_result is None or bool(getattr(send_result, "success", False)): + sent_any = True + return sent_any except Exception as e: logger.warning("Auto voice reply failed: %s", e, exc_info=True) + return False finally: for p in ({audio_path, *actual_paths} - {None}): try: diff --git a/tests/gateway/test_reply_delivery_policy.py b/tests/gateway/test_reply_delivery_policy.py new file mode 100644 index 0000000000000..65b738e1c0674 --- /dev/null +++ b/tests/gateway/test_reply_delivery_policy.py @@ -0,0 +1,125 @@ +import pytest + +from gateway.config import GatewayConfig, Platform, PlatformConfig +from gateway.platforms.base import MessageEvent, MessageType, ReplyDeliveryPolicy, SendResult +from gateway.run import GatewayRunner +from gateway.session import SessionSource + + +class _PolicyAdapter: + def __init__(self, policy): + self.policy = policy + self.observed = [] + self.sent_voice = [] + + def observe_inbound_message(self, event): + self.observed.append(event) + + def reply_delivery_policy(self, event, response, *, voice_mode, already_sent): + return self.policy + + async def send_voice(self, **kwargs): + self.sent_voice.append(kwargs) + return SendResult(success=True, message_id="voice-1") + + +def _runner(adapter): + runner = GatewayRunner.__new__(GatewayRunner) + platform = Platform.TELEGRAM + runner.config = GatewayConfig( + platforms={platform: PlatformConfig(enabled=True, extra={})} + ) + runner.adapters = {platform: adapter} + runner._voice_mode = {} + return runner + + +def _event(message_type=MessageType.TEXT): + return MessageEvent( + text="hello", + message_type=message_type, + source=SessionSource( + platform=Platform.TELEGRAM, + chat_id="chat-1", + chat_type="dm", + user_id="user-1", + ), + ) + + +def test_runner_uses_adapter_reply_delivery_policy_for_voice_decision(): + adapter = _PolicyAdapter(ReplyDeliveryPolicy(send_voice_reply=True)) + runner = _runner(adapter) + event = _event(MessageType.TEXT) + + assert runner._should_send_voice_reply(event, "hi", [], already_sent=False) is True + + +def test_runner_ignores_non_policy_adapter_return_and_preserves_legacy_gate(): + adapter = _PolicyAdapter(object()) + runner = _runner(adapter) + event = _event(MessageType.TEXT) + + assert runner._should_send_voice_reply(event, "hi", [], already_sent=False) is False + + +def test_runner_observes_inbound_message_before_dispatch(): + adapter = _PolicyAdapter(ReplyDeliveryPolicy()) + runner = _runner(adapter) + event = _event(MessageType.TEXT) + + runner._observe_inbound_message(event) + + assert adapter.observed == [event] + + +@pytest.mark.asyncio +async def test_send_voice_reply_reports_success(monkeypatch, tmp_path): + adapter = _PolicyAdapter(ReplyDeliveryPolicy()) + runner = _runner(adapter) + event = _event(MessageType.TEXT) + audio_path = tmp_path / "voice.mp3" + audio_path.write_bytes(b"audio") + + monkeypatch.setattr("gateway.run.text_to_speech_tool", None, raising=False) + + def fake_tts(*, text, output_path): + return '{"success": true, "file_path": "' + str(audio_path) + '"}' + + monkeypatch.setattr("tools.tts_tool.text_to_speech_tool", fake_tts) + monkeypatch.setattr("tools.tts_tool._strip_markdown_for_tts", lambda text: text) + + assert await runner._send_voice_reply(event, "hello") is True + assert adapter.sent_voice + + +def test_default_policy_preserves_voice_mode_gate(): + from gateway.platforms.base import BasePlatformAdapter + + class _DefaultPolicyAdapter(BasePlatformAdapter): + async def connect(self): + return True + + async def disconnect(self): + return None + + async def send(self, chat_id, content, reply_to=None, metadata=None): + return SendResult(success=True) + + async def get_chat_info(self, chat_id): + return {} + + adapter = _DefaultPolicyAdapter(PlatformConfig(enabled=True, extra={}), Platform.TELEGRAM) + event = _event(MessageType.VOICE) + + off = adapter.reply_delivery_policy(event, "hi", voice_mode="off", already_sent=True) + voice_only_not_streamed = adapter.reply_delivery_policy( + event, "hi", voice_mode="voice_only", already_sent=False + ) + voice_only_streamed = adapter.reply_delivery_policy( + event, "hi", voice_mode="voice_only", already_sent=True + ) + + assert off.send_voice_reply is False + assert voice_only_not_streamed.send_voice_reply is False + assert voice_only_streamed.send_voice_reply is True From 18a655de3f9c9a76ceb99173aff9126b4fd4d332 Mon Sep 17 00:00:00 2001 From: Jetha Chan Date: Tue, 23 Jun 2026 19:58:57 +0900 Subject: [PATCH 4/6] fix(gateway): tolerate runner tests without adapters --- gateway/run.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/gateway/run.py b/gateway/run.py index a33931e8d047e..7693fdbf08c90 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -19894,7 +19894,10 @@ async def _handle_voice_channel_input( def _observe_inbound_message(self, event: MessageEvent) -> None: """Let the source adapter observe an inbound event before dispatch.""" - adapter = self.adapters.get(event.source.platform) + adapters = getattr(self, "adapters", None) + if not adapters: + return + adapter = adapters.get(event.source.platform) if not adapter or not hasattr(adapter, "observe_inbound_message"): return try: From 7c94062a82350c9de3a8dac277c0c0edc44e517b Mon Sep 17 00:00:00 2001 From: Jetha Chan Date: Sun, 19 Jul 2026 10:01:25 +0900 Subject: [PATCH 5/6] fix(gateway): resolve reply-policy adapter via _adapter_for_source self.adapters is the default profile's adapter map. Resolve the reply-delivery-policy and inbound-observe adapter through _adapter_for_source(event.source) so multiplex secondary profiles consult their own adapter's policy (aligns with 8a9bc38c). Co-Authored-By: Claude Fable 5 --- gateway/run.py | 11 +++---- tests/gateway/test_reply_delivery_policy.py | 32 ++++++++++++++++++++- 2 files changed, 37 insertions(+), 6 deletions(-) diff --git a/gateway/run.py b/gateway/run.py index 7693fdbf08c90..bc259b94bfe8f 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -19894,10 +19894,9 @@ async def _handle_voice_channel_input( def _observe_inbound_message(self, event: MessageEvent) -> None: """Let the source adapter observe an inbound event before dispatch.""" - adapters = getattr(self, "adapters", None) - if not adapters: - return - adapter = adapters.get(event.source.platform) + # Resolve through _adapter_for_source so multiplex secondary profiles + # observe their own adapter, never the default profile's (8a9bc38c). + adapter = self._adapter_for_source(event.source) if not adapter or not hasattr(adapter, "observe_inbound_message"): return try: @@ -19917,7 +19916,9 @@ def _reply_delivery_policy( already_sent: bool = False, ): """Return the adapter's reply delivery policy for this turn.""" - adapter = self.adapters.get(event.source.platform) + # self.adapters is the DEFAULT profile's adapter map; a multiplex + # secondary profile must consult its own adapter's policy (8a9bc38c). + adapter = self._adapter_for_source(event.source) chat_id = event.source.chat_id # Raw get — ``None`` (chat never set a voice mode) is distinct from an # explicit ``"off"``: the ``voice.auto_tts`` fallback below (and in the diff --git a/tests/gateway/test_reply_delivery_policy.py b/tests/gateway/test_reply_delivery_policy.py index 65b738e1c0674..8080dc5fb0b3b 100644 --- a/tests/gateway/test_reply_delivery_policy.py +++ b/tests/gateway/test_reply_delivery_policy.py @@ -11,11 +11,13 @@ def __init__(self, policy): self.policy = policy self.observed = [] self.sent_voice = [] + self.policy_calls = [] def observe_inbound_message(self, event): self.observed.append(event) def reply_delivery_policy(self, event, response, *, voice_mode, already_sent): + self.policy_calls.append(event) return self.policy async def send_voice(self, **kwargs): @@ -34,7 +36,7 @@ def _runner(adapter): return runner -def _event(message_type=MessageType.TEXT): +def _event(message_type=MessageType.TEXT, profile=None): return MessageEvent( text="hello", message_type=message_type, @@ -43,6 +45,7 @@ def _event(message_type=MessageType.TEXT): chat_id="chat-1", chat_type="dm", user_id="user-1", + profile=profile, ), ) @@ -63,6 +66,33 @@ def test_runner_ignores_non_policy_adapter_return_and_preserves_legacy_gate(): assert runner._should_send_voice_reply(event, "hi", [], already_sent=False) is False +def test_multiplex_source_resolves_secondary_profile_adapter_policy(): + """Regression: a secondary-profile source must consult its own adapter's + policy, not the default profile's (see 8a9bc38c).""" + default_adapter = _PolicyAdapter(ReplyDeliveryPolicy(send_voice_reply=False)) + secondary_adapter = _PolicyAdapter(ReplyDeliveryPolicy(send_voice_reply=True)) + runner = _runner(default_adapter) + runner._profile_adapters = {"lars": {Platform.TELEGRAM: secondary_adapter}} + event = _event(MessageType.TEXT, profile="lars") + + assert runner._should_send_voice_reply(event, "hi", [], already_sent=False) is True + assert secondary_adapter.policy_calls == [event] + assert default_adapter.policy_calls == [] + + +def test_multiplex_source_observes_secondary_profile_adapter(): + default_adapter = _PolicyAdapter(ReplyDeliveryPolicy()) + secondary_adapter = _PolicyAdapter(ReplyDeliveryPolicy()) + runner = _runner(default_adapter) + runner._profile_adapters = {"lars": {Platform.TELEGRAM: secondary_adapter}} + event = _event(MessageType.TEXT, profile="lars") + + runner._observe_inbound_message(event) + + assert secondary_adapter.observed == [event] + assert default_adapter.observed == [] + + def test_runner_observes_inbound_message_before_dispatch(): adapter = _PolicyAdapter(ReplyDeliveryPolicy()) runner = _runner(adapter) From 795998e1f5c242aabf1e41825407dc87ef865aa6 Mon Sep 17 00:00:00 2001 From: Jetha Chan Date: Sun, 19 Jul 2026 10:02:58 +0900 Subject: [PATCH 6/6] fix(gateway): isolate reply-policy callback failures Catch exceptions from the adapter reply_delivery_policy callback and fall back to the legacy voice-mode gate so a buggy adapter policy can never disrupt final reply delivery. Mirrors the isolation pattern used by _observe_inbound_message. Co-Authored-By: Claude Fable 5 --- gateway/run.py | 27 ++++++++++++++------ tests/gateway/test_reply_delivery_policy.py | 28 +++++++++++++++++++++ 2 files changed, 47 insertions(+), 8 deletions(-) diff --git a/gateway/run.py b/gateway/run.py index bc259b94bfe8f..1fa3f5ab0eb59 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -19929,14 +19929,25 @@ def _reply_delivery_policy( return ReplyDeliveryPolicy() if adapter and hasattr(adapter, "reply_delivery_policy"): - policy = adapter.reply_delivery_policy( - event, - response, - voice_mode=voice_mode, - already_sent=already_sent, - ) - if isinstance(policy, ReplyDeliveryPolicy): - return policy + # Isolate adapter policy failures: a buggy callback must never + # disrupt final reply delivery (same pattern as + # _observe_inbound_message). Fall back to the legacy path below. + try: + policy = adapter.reply_delivery_policy( + event, + response, + voice_mode=voice_mode, + already_sent=already_sent, + ) + except Exception: + logger.warning( + "Adapter reply_delivery_policy failed for %s; falling back to legacy delivery", + getattr(event.source.platform, "value", event.source.platform), + exc_info=True, + ) + else: + if isinstance(policy, ReplyDeliveryPolicy): + return policy # Legacy fallback for adapters without a policy hook. Mirrors the # pre-policy inline logic, including the ``voice.auto_tts`` term diff --git a/tests/gateway/test_reply_delivery_policy.py b/tests/gateway/test_reply_delivery_policy.py index 8080dc5fb0b3b..323477d98f9a6 100644 --- a/tests/gateway/test_reply_delivery_policy.py +++ b/tests/gateway/test_reply_delivery_policy.py @@ -93,6 +93,34 @@ def test_multiplex_source_observes_secondary_profile_adapter(): assert default_adapter.observed == [] +def test_policy_callback_failure_falls_back_to_legacy_delivery(caplog): + """A raising policy callback must not disrupt the final reply: the legacy + voice-mode gate is used instead and the error is logged.""" + + class _RaisingPolicyAdapter(_PolicyAdapter): + def reply_delivery_policy(self, event, response, *, voice_mode, already_sent): + raise RuntimeError("policy exploded") + + adapter = _RaisingPolicyAdapter(None) + runner = _runner(adapter) + event = _event(MessageType.TEXT) + + with caplog.at_level("WARNING", logger="gateway.run"): + # Legacy gate with voice_mode off: no voice reply, no exception. + assert runner._should_send_voice_reply(event, "hi", [], already_sent=False) is False + # Text delivery is never suppressed when the policy callback fails. + assert ( + runner._should_suppress_text_after_voice_reply(event, "hi", True, already_sent=False) + is False + ) + + assert any("reply_delivery_policy failed" in r.getMessage() for r in caplog.records) + + # Legacy gate still honors an explicit /voice all opt-in. + runner._voice_mode = {runner._voice_key(Platform.TELEGRAM, "chat-1"): "all"} + assert runner._should_send_voice_reply(event, "hi", [], already_sent=False) is True + + def test_runner_observes_inbound_message_before_dispatch(): adapter = _PolicyAdapter(ReplyDeliveryPolicy()) runner = _runner(adapter)