fix(gateway): align fallback delete with sibling style + add regression tests

Follow-up to HuangYuChuh's #17384 cherry-pick:

- Use defensive getattr+logger.debug for delete_message lookup, mirroring
  the sibling _try_send_fresh_final cleanup pattern at L820+. Platforms
  that don't implement delete_message no longer raise AttributeError; the
  failure path now logs at debug for diagnosability instead of silently
  swallowing.
- Add three regression tests in tests/gateway/test_stream_consumer.py:
  - delete_message awaited on happy-path exit with stale id
  - delete_message NOT awaited when no fallback chunks reached the user
  - no crash on adapters that lack delete_message (spec-restricted mock)

gateway/stream_consumer.py

@@ -689,14 +689,20 @@ class GatewayStreamConsumer:
             self._notify_new_message()
 
         # Remove the frozen partial message so the user only sees the
-        # complete fallback response.  Best-effort — if the delete fails
-        # (e.g. flood control still active, or bot lacks permission), the
-        # partial message remains but at least the full answer was delivered.
+        # complete fallback response.  Best-effort — if the platform doesn't
+        # implement ``delete_message``, the delete fails (flood control still
+        # active, bot lacks permission, message too old to delete), the
+        # partial remains but at least the full answer was delivered.
         if stale_message_id and stale_message_id != last_message_id:
-            try:
-                await self.adapter.delete_message(self.chat_id, stale_message_id)
-            except Exception:
-                pass
+            delete_fn = getattr(self.adapter, "delete_message", None)
+            if delete_fn is not None:
+                try:
+                    await delete_fn(self.chat_id, stale_message_id)
+                except Exception as e:
+                    logger.debug(
+                        "Fallback partial cleanup failed (%s): %s",
+                        stale_message_id, e,
+                    )
 
         self._message_id = last_message_id
         self._already_sent = True

tests/gateway/test_stream_consumer.py

@@ -793,6 +793,79 @@ class TestSegmentBreakOnToolBoundary:
             "_send_fallback_final — the #10807 fix should prevent this"
         )
 
+    @pytest.mark.asyncio
+    async def test_fallback_final_deletes_partial_after_chunks_succeed(self):
+        """After fallback chunks land, the frozen partial must be deleted so
+        the user sees only the complete response (#16668)."""
+        adapter = MagicMock()
+        adapter.send = AsyncMock(
+            return_value=SimpleNamespace(success=True, message_id="msg_new"),
+        )
+        adapter.edit_message = AsyncMock(
+            return_value=SimpleNamespace(success=True),
+        )
+        adapter.delete_message = AsyncMock(return_value=None)
+        adapter.MAX_MESSAGE_LENGTH = 4096
+
+        config = StreamConsumerConfig(edit_interval=0.01, buffer_threshold=5)
+        consumer = GatewayStreamConsumer(adapter, "chat_123", config)
+
+        # Seed the consumer as if it already edited a partial message that
+        # later got stuck (flood control etc.) — _message_id is the stale id.
+        consumer._message_id = "msg_partial"
+        consumer._last_sent_text = "Working on i"
+
+        await consumer._send_fallback_final("Working on it. Done!")
+
+        adapter.delete_message.assert_awaited_once_with("chat_123", "msg_partial")
+        assert consumer._final_response_sent is True
+
+    @pytest.mark.asyncio
+    async def test_fallback_final_does_not_delete_when_no_chunks_reach_user(self):
+        """If every fallback send fails, the partial is the only thing the
+        user has — must NOT be deleted."""
+        adapter = MagicMock()
+        adapter.send = AsyncMock(
+            return_value=SimpleNamespace(success=False, error="network down"),
+        )
+        adapter.edit_message = AsyncMock(
+            return_value=SimpleNamespace(success=True),
+        )
+        adapter.delete_message = AsyncMock(return_value=None)
+        adapter.MAX_MESSAGE_LENGTH = 4096
+
+        config = StreamConsumerConfig(edit_interval=0.01, buffer_threshold=5)
+        consumer = GatewayStreamConsumer(adapter, "chat_123", config)
+
+        consumer._message_id = "msg_partial"
+        consumer._last_sent_text = "Working on i"
+
+        await consumer._send_fallback_final("Working on it. Done!")
+
+        adapter.delete_message.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_fallback_final_skips_delete_when_adapter_lacks_method(self):
+        """Platforms without delete_message must not crash the fallback path."""
+        adapter = MagicMock(spec=["send", "edit_message", "MAX_MESSAGE_LENGTH"])
+        adapter.send = AsyncMock(
+            return_value=SimpleNamespace(success=True, message_id="msg_new"),
+        )
+        adapter.edit_message = AsyncMock(
+            return_value=SimpleNamespace(success=True),
+        )
+        adapter.MAX_MESSAGE_LENGTH = 4096
+
+        config = StreamConsumerConfig(edit_interval=0.01, buffer_threshold=5)
+        consumer = GatewayStreamConsumer(adapter, "chat_123", config)
+
+        consumer._message_id = "msg_partial"
+        consumer._last_sent_text = "Working on i"
+
+        # Should not raise even though the adapter has no delete_message.
+        await consumer._send_fallback_final("Working on it. Done!")
+        assert consumer._final_response_sent is True
+
 
 class TestInterimCommentaryMessages:
     @pytest.mark.asyncio