fix: ignore typing failures

Make Modmail keep working when typing is disabled/outage

Author: lorenzo132

bot.py

@@ -1403,7 +1403,10 @@ async def on_typing(self, channel, user, _):
             thread = await self.threads.find(recipient=user)
 
             if thread:
-                await thread.channel.typing()
+                try:
+                    await thread.channel.typing()
+                except Exception:
+                    pass
         else:
             if not self.config.get("mod_typing"):
                 return
@@ -1413,7 +1416,10 @@ async def on_typing(self, channel, user, _):
                 for user in thread.recipients:
                     if await self.is_blocked(user):
                         continue
-                    await user.typing()
+                    try:
+                        await user.typing()
+                    except Exception:
+                        pass
 
     async def handle_reaction_events(self, payload):
         user = self.get_user(payload.user_id)
@@ -1720,7 +1726,10 @@ async def on_command_error(
                 return
 
         if isinstance(exception, (commands.BadArgument, commands.BadUnionArgument)):
-            await context.typing()
+            try:
+                await context.typing()
+            except Exception:
+                pass
             await context.send(embed=discord.Embed(color=self.error_color, description=str(exception)))
         elif isinstance(exception, commands.CommandNotFound):
             logger.warning("CommandNotFound: %s", exception)

cogs/modmail.py

@@ -1210,7 +1210,8 @@ async def logs(self, ctx, *, user: User = None):
         `user` may be a user ID, mention, or name.
         """
 
-        await ctx.typing()
+        async with safe_typing(ctx):
+            pass
 
         if not user:
             thread = ctx.thread
@@ -1342,7 +1343,8 @@ async def logs_search(self, ctx, limit: Optional[int] = None, *, query):
         Provide a `limit` to specify the maximum number of logs the bot should find.
         """
 
-        await ctx.typing()
+        async with safe_typing(ctx):
+            pass
 
         entries = await self.bot.api.search_by_text(query, limit)
 
@@ -1371,7 +1373,7 @@ async def reply(self, ctx, *, msg: str = ""):
 
         ctx.message.content = msg
 
-        async with ctx.typing():
+        async with safe_typing(ctx):
             await ctx.thread.reply(ctx.message)
 
     @commands.command(aliases=["formatreply"])
@@ -1393,7 +1395,7 @@ async def freply(self, ctx, *, msg: str = ""):
             msg, channel=ctx.channel, recipient=ctx.thread.recipient, author=ctx.message.author
         )
         ctx.message.content = msg
-        async with ctx.typing():
+        async with safe_typing(ctx):
             await ctx.thread.reply(ctx.message)
 
     @commands.command(aliases=["formatanonreply"])
@@ -1415,7 +1417,7 @@ async def fareply(self, ctx, *, msg: str = ""):
             msg, channel=ctx.channel, recipient=ctx.thread.recipient, author=ctx.message.author
         )
         ctx.message.content = msg
-        async with ctx.typing():
+        async with safe_typing(ctx):
             await ctx.thread.reply(ctx.message, anonymous=True)
 
     @commands.command(aliases=["formatplainreply"])
@@ -1437,7 +1439,7 @@ async def fpreply(self, ctx, *, msg: str = ""):
             msg, channel=ctx.channel, recipient=ctx.thread.recipient, author=ctx.message.author
         )
         ctx.message.content = msg
-        async with ctx.typing():
+        async with safe_typing(ctx):
             await ctx.thread.reply(ctx.message, plain=True)
 
     @commands.command(aliases=["formatplainanonreply"])
@@ -1459,7 +1461,7 @@ async def fpareply(self, ctx, *, msg: str = ""):
             msg, channel=ctx.channel, recipient=ctx.thread.recipient, author=ctx.message.author
         )
         ctx.message.content = msg
-        async with ctx.typing():
+        async with safe_typing(ctx):
             await ctx.thread.reply(ctx.message, anonymous=True, plain=True)
 
     @commands.command(aliases=["anonreply", "anonymousreply"])
@@ -1476,7 +1478,7 @@ async def areply(self, ctx, *, msg: str = ""):
         and `anon_tag` config variables to do so.
         """
         ctx.message.content = msg
-        async with ctx.typing():
+        async with safe_typing(ctx):
             await ctx.thread.reply(ctx.message, anonymous=True)
 
     @commands.command(aliases=["plainreply"])
@@ -1490,7 +1492,7 @@ async def preply(self, ctx, *, msg: str = ""):
         automatically embedding image URLs.
         """
         ctx.message.content = msg
-        async with ctx.typing():
+        async with safe_typing(ctx):
             await ctx.thread.reply(ctx.message, plain=True)
 
     @commands.command(aliases=["plainanonreply", "plainanonymousreply"])
@@ -1504,7 +1506,7 @@ async def pareply(self, ctx, *, msg: str = ""):
         automatically embedding image URLs.
         """
         ctx.message.content = msg
-        async with ctx.typing():
+        async with safe_typing(ctx):
             await ctx.thread.reply(ctx.message, anonymous=True, plain=True)
 
     @commands.group(invoke_without_command=True)
@@ -1517,7 +1519,7 @@ async def note(self, ctx, *, msg: str = ""):
         Useful for noting context.
         """
         ctx.message.content = msg
-        async with ctx.typing():
+        async with safe_typing(ctx):
             msg = await ctx.thread.note(ctx.message)
             await msg.pin()
 
@@ -1529,7 +1531,7 @@ async def note_persistent(self, ctx, *, msg: str = ""):
         Take a persistent note about the current user.
         """
         ctx.message.content = msg
-        async with ctx.typing():
+        async with safe_typing(ctx):
             msg = await ctx.thread.note(ctx.message, persistent=True)
             await msg.pin()
         await self.bot.api.create_note(recipient=ctx.thread.recipient, message=ctx.message, message_id=msg.id)

cogs/plugins.py

@@ -20,7 +20,7 @@
 from core import checks
 from core.models import PermissionLevel, getLogger
 from core.paginator import EmbedPaginatorSession
-from core.utils import trigger_typing, truncate
+from core.utils import trigger_typing, truncate, safe_typing
 
 logger = getLogger(__name__)
 
@@ -484,7 +484,7 @@ async def update_plugin(self, ctx, plugin_name):
             embed = discord.Embed(description="Plugin is not installed.", color=self.bot.error_color)
             return await ctx.send(embed=embed)
 
-        async with ctx.typing():
+        async with safe_typing(ctx):
             embed = discord.Embed(
                 description=f"Successfully updated {plugin.name}.", color=self.bot.main_color
             )

cogs/utility.py

@@ -1,3 +1,4 @@
+from core.utils import trigger_typing, truncate, safe_typing
 import asyncio
 import inspect
 import os

core/thread.py

@@ -1457,11 +1457,14 @@ def lottie_to_png(data):
         ):
             logger.info("Sending a message to %s when DM disabled is set.", self.recipient)
 
+        # Best-effort typing: never block message delivery if typing fails
         try:
             await destination.typing()
         except discord.NotFound:
             logger.warning("Channel not found.")
             raise
+        except (discord.Forbidden, discord.HTTPException, Exception) as e:
+            logger.warning("Unable to send typing to %s: %s. Continuing without typing.", destination, e)
 
         if not from_mod and not note:
             mentions = await self.get_notifications()

core/utils.py

@@ -1,5 +1,6 @@
 import base64
 import functools
+import contextlib
 import re
 import typing
 from datetime import datetime, timezone
@@ -34,6 +35,7 @@
     "normalize_alias",
     "format_description",
     "trigger_typing",
+    "safe_typing",
     "escape_code_block",
     "tryint",
     "get_top_role",
@@ -425,10 +427,42 @@ def format_description(i, names):
     )
 
 
+class _SafeTyping:
+    """Best-effort typing context manager.
+
+    Suppresses errors from Discord's typing endpoint so core flows continue
+    when typing is disabled or experiencing outages.
+    """
+
+    def __init__(self, target):
+        # target can be a Context or any Messageable (channel/DM/user)
+        self._target = target
+        self._cm = None
+
+    async def __aenter__(self):
+        try:
+            self._cm = self._target.typing()
+            return await self._cm.__aenter__()
+        except Exception:
+            # typing is best-effort; ignore any failure
+            self._cm = None
+
+    async def __aexit__(self, exc_type, exc, tb):
+        if self._cm is not None:
+            with contextlib.suppress(Exception):
+                return await self._cm.__aexit__(exc_type, exc, tb)
+
+
+def safe_typing(target):
+    return _SafeTyping(target)
+
+
 def trigger_typing(func):
     @functools.wraps(func)
     async def wrapper(self, ctx: commands.Context, *args, **kwargs):
-        await ctx.typing()
+        # Fire and forget typing; do not block on failures
+        async with safe_typing(ctx):
+            pass
         return await func(self, ctx, *args, **kwargs)
 
     return wrapper