refactor(gateway): add adapter reply delivery policy

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,

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,94 @@ 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."""
+        # 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:
+            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."""
+        # 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
-        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))
+
+        if not response or response.startswith("Error:"):
+            return ReplyDeliveryPolicy()
 
-        adapter = self.adapters.get(event.source.platform)
+        if adapter and hasattr(adapter, "reply_delivery_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
+        # (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 +19999,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 +20028,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 +20044,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 +20058,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 +20088,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 +20111,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:

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)