deer-flow/backend/app/channels/discord.py

"""Discord channel integration using discord.py."""

from __future__ import annotations

import asyncio
import io
import json
import logging
import threading
from concurrent.futures import Future
from pathlib import Path
from typing import Any

from app.channels.base import Channel
from app.channels.commands import is_known_channel_command
from app.channels.connection_identity import attach_connection_identity
from app.channels.message_bus import InboundMessage, InboundMessageType, InboundReservation, MessageBus, OutboundMessage, ResolvedAttachment

logger = logging.getLogger(__name__)

_DISCORD_MAX_MESSAGE_LEN = 2000

# Bound for outbound work scheduled onto the Discord client's event loop.
# Discord API calls normally return in well under a second; anything still
# pending after this is a dead or wedged client, not a slow response.
DISCORD_OUTBOUND_TIMEOUT_SECONDS = 30.0

# File uploads carry an unbounded-size payload with no per-channel size cap
# (unlike Feishu/Telegram's send_file limits), so they get their own, larger
# bound: a 50 MB artifact over a ~5 Mbps uplink takes ~80 s to push, and a
# large 429 retry-after inside discord.py can extend that further. Cancelling
# a healthy upload mid-transfer would report it as failed, so the bound here
# only exists to convert a wedged client into a logged failure.
DISCORD_UPLOAD_TIMEOUT_SECONDS = 120.0


class DiscordChannel(Channel):
    """Discord bot channel.

    Configuration keys (in ``config.yaml`` under ``channels.discord``):
        - ``bot_token``: Discord Bot token.
        - ``allowed_guilds``: (optional) List of allowed Discord guild IDs. Empty = allow all.
        - ``mention_only``: (optional) If true, only respond when the bot is mentioned.
        - ``allowed_channels``: (optional) List of channel IDs where messages are always accepted
          (even when mention_only is true). Use for channels where you want the bot to respond
          without mentions. Empty = mention_only applies everywhere.
        - ``thread_mode``: (optional) If true, group a channel conversation into a thread.
          Default: same as ``mention_only``.
    """

    def __init__(self, bus: MessageBus, config: dict[str, Any]) -> None:
        super().__init__(name="discord", bus=bus, config=config)
        self._bot_token = str(config.get("bot_token", "")).strip()
        self._allowed_guilds: set[int] = set()
        for guild_id in config.get("allowed_guilds", []):
            try:
                self._allowed_guilds.add(int(guild_id))
            except (TypeError, ValueError):
                continue
        self._mention_only: bool = bool(config.get("mention_only", False))
        self._thread_mode: bool = config.get("thread_mode", self._mention_only)
        self._allowed_channels: set[str] = set()
        for channel_id in config.get("allowed_channels", []):
            self._allowed_channels.add(str(channel_id))

        # Session tracking: channel_id -> Discord thread_id (in-memory, persisted to JSON).
        # Uses a dedicated JSON file separate from ChannelStore, which maps IM
        # conversations to DeerFlow thread IDs — a different concern.
        self._active_threads: dict[str, str] = {}
        # Reverse-lookup set for O(1) thread ID checks (avoids O(n) scan of _active_threads.values()).
        self._active_thread_ids: set[str] = set()
        # Lock protecting _active_threads and the JSON file from concurrent access.
        # _run_client (Discord loop thread) and the main thread both read/write.
        self._thread_store_lock = threading.Lock()
        # Set once _load_active_threads() has run: the in-memory map is then a
        # faithful view of the file, so stop() may flush it. It stays False when
        # start() bails before the load (missing bot_token, discord import
        # error) — ChannelService then stops the instance, and an ungated flush
        # would overwrite the persisted mappings with an empty snapshot (#2897).
        self._thread_store_loaded = False
        store = config.get("channel_store")
        if store is not None:
            self._thread_store_path = store._path.parent / "discord_threads.json"
        else:
            self._thread_store_path = Path.home() / ".deer-flow" / "channels" / "discord_threads.json"

        # Typing indicator management
        self._typing_tasks: dict[str, asyncio.Task] = {}
        # Messages whose hand-off is pending or committed, per typing task. A
        # failed hand-off may stop the indicator only when no other message
        # still relies on it (see ``_acknowledge_after_handoff``).
        self._typing_dependents: dict[asyncio.Task, int] = {}

        # Strong references for this channel's in-flight ack-reaction tasks.
        # The event loop keeps only weak references to scheduled tasks, so a
        # bare ``asyncio.create_task`` could be garbage-collected mid-flight
        # and silently drop the acknowledgment reaction (same retention
        # pattern as the deferred subagent cleanup in #4928). Instance-level
        # on purpose, matching ``_typing_tasks``: one channel's shutdown must
        # not cancel another instance's in-flight reactions.
        self._ack_reaction_tasks: set[asyncio.Task[None]] = set()

        self._client = None
        self._thread: threading.Thread | None = None
        self._discord_loop: asyncio.AbstractEventLoop | None = None
        self._main_loop: asyncio.AbstractEventLoop | None = None
        self._discord_module = None

    async def start(self) -> None:
        if self._running:
            return

        try:
            import discord
        except ImportError:
            logger.error("discord.py is not installed. The Discord channel needs the 'discord' extra: run `cd backend && uv sync --extra discord`.")
            return

        if not self._bot_token:
            logger.error("Discord channel requires bot_token")
            return

        intents = discord.Intents.default()
        intents.messages = True
        intents.guilds = True
        intents.message_content = True

        client = discord.Client(
            intents=intents,
            allowed_mentions=discord.AllowedMentions.none(),
        )
        self._client = client
        self._discord_module = discord
        self._main_loop = asyncio.get_event_loop()
        self._open_threadsafe_future_intake()

        @client.event
        async def on_message(message) -> None:
            await self._on_message(message)

        self._running = True
        self.bus.subscribe_outbound(self._on_outbound)

        self._thread = threading.Thread(target=self._run_client, daemon=True)
        self._thread.start()
        await asyncio.to_thread(self._load_active_threads)
        logger.info("Discord channel started")

    def _load_active_threads(self) -> None:
        """Restore Discord thread mappings from the dedicated JSON file on startup."""
        with self._thread_store_lock:
            try:
                if not self._thread_store_path.exists():
                    logger.debug("[Discord] no thread mappings file at %s", self._thread_store_path)
                else:
                    data = json.loads(self._thread_store_path.read_text())
                    self._active_threads.clear()
                    self._active_thread_ids.clear()
                    for channel_id, thread_id in data.items():
                        self._active_threads[channel_id] = thread_id
                        self._active_thread_ids.add(thread_id)
                    if self._active_threads:
                        logger.info("[Discord] restored %d thread mappings from %s", len(self._active_threads), self._thread_store_path)
            except Exception:
                logger.exception("[Discord] failed to load thread mappings")
                return
            # The in-memory map is now a superset of the on-disk store (empty
            # when the file is absent), so stop() may safely flush it. Until
            # this point — a start() that bailed on a missing bot_token or a
            # discord import error, which ChannelService then stops — the map is
            # empty and flushing would clobber the file with {}.
            self._thread_store_loaded = True

    def _record_thread_mapping(self, channel_id: str, thread_id: str) -> None:
        """Synchronously update the in-memory channel->thread mapping and its reverse-lookup set.

        Runs on the event loop (no IO, no await) so a follow-up message in the
        newly created thread is recognized immediately, before the offloaded
        persistence write completes. Deferring this update into the worker
        thread opened a window where ``_active_thread_ids`` had not yet been
        updated and an inbound message was misclassified as orphaned (see the
        #3927 review). Persistence is handled separately by
        ``_persist_thread_mappings``.
        """
        old_id = self._active_threads.get(channel_id)
        self._active_threads[channel_id] = thread_id
        if old_id:
            self._active_thread_ids.discard(old_id)
        self._active_thread_ids.add(thread_id)

    def _persist_thread_mappings(self) -> None:
        """Flush the current in-memory thread mappings to disk.

        Intended for ``asyncio.to_thread``: this is pure filesystem IO. The
        in-memory state is updated synchronously by ``_record_thread_mapping``,
        so persistence latency never delays visibility of a new mapping to
        inbound-message handling. The mapping is snapshotted under the store
        lock so a concurrent record cannot mutate the dict mid-serialization.
        """
        with self._thread_store_lock:
            try:
                snapshot = dict(self._active_threads)
                self._thread_store_path.parent.mkdir(parents=True, exist_ok=True)
                self._thread_store_path.write_text(json.dumps(snapshot, indent=2))
            except Exception:
                logger.exception("[Discord] failed to persist thread mappings")

    @staticmethod
    def _read_attachment_bytes(path: str) -> bytes:
        """Read an attachment file synchronously (intended for ``asyncio.to_thread``)."""
        with open(path, "rb") as fp:
            return fp.read()

    async def stop(self) -> None:
        self._running = False
        self.bus.unsubscribe_outbound(self._on_outbound)
        await self._close_and_drain_threadsafe_futures()

        # Best-effort durability: flush in-memory thread mappings so the most
        # recent channel->thread mapping survives a hard shutdown (process
        # killed between a thread creation and its background persistence
        # write). The create path already persists off the event loop after
        # each new thread, so this is a safety net, not the primary write path.
        # Gated on _thread_store_loaded: a stop() taken before the initial load
        # has an empty in-memory map, and flushing it would overwrite the file
        # with {} — the #2897 data loss this PR exists to prevent.
        if self._thread_store_loaded:
            try:
                await asyncio.to_thread(self._persist_thread_mappings)
            except Exception:
                logger.warning("[Discord] failed to flush thread mappings during shutdown")

        discord_loop = self._discord_loop
        current_loop = asyncio.get_running_loop()
        if discord_loop is None or discord_loop is current_loop:
            await self._cancel_ephemeral_tasks()
        elif discord_loop.is_running():
            # Serialize cleanup with _start_typing()/_schedule_ack_reaction()
            # on the owning loop.  The
            # loop may exit between is_running() and scheduling, so neither
            # scheduling nor waiting is allowed to block the rest of stop().
            cleanup_coro = self._cancel_ephemeral_tasks()
            try:
                cleanup_future = asyncio.run_coroutine_threadsafe(cleanup_coro, discord_loop)
            except RuntimeError:
                cleanup_coro.close()
                logger.warning("[Discord] event loop stopped before ephemeral-task cleanup could be scheduled")
            else:
                try:
                    await asyncio.wait_for(asyncio.wrap_future(cleanup_future), timeout=10)
                except TimeoutError:
                    cleanup_future.cancel()
                    logger.warning("[Discord] ephemeral-task cleanup timed out after 10s")
                except Exception:
                    logger.exception("[Discord] error while cleaning up ephemeral tasks")

        if self._client and discord_loop and discord_loop.is_running():
            close_coro = self._client.close()
            try:
                close_future = asyncio.run_coroutine_threadsafe(close_coro, discord_loop)
            except RuntimeError:
                close_coro.close()
                logger.warning("[Discord] event loop stopped before client close could be scheduled")
            else:
                try:
                    await asyncio.wait_for(asyncio.wrap_future(close_future), timeout=10)
                except TimeoutError:
                    close_future.cancel()
                    logger.warning("[Discord] client close timed out after 10s")
                except Exception:
                    logger.exception("[Discord] error while closing client")

        if self._thread:
            # The client thread normally exits right after the close above, but
            # a timed-out close or a slow ``_run_client()`` drain can keep it
            # alive for the whole join timeout; keep that wait off the loop.
            await asyncio.to_thread(self._thread.join, timeout=10)
            self._thread = None

        # _run_client() normally drains these tasks in its finally block.  If
        # the owning loop was stopped externally before that cleanup ran, only
        # discard the stale references here; awaiting them from this loop would
        # raise a cross-loop RuntimeError.
        if discord_loop and discord_loop is not current_loop and not discord_loop.is_running():
            self._discard_ephemeral_tasks()

        self._client = None
        self._discord_loop = None
        self._discord_module = None
        logger.info("Discord channel stopped")

    @property
    def is_running(self) -> bool:
        """Running means the client thread is still alive, not just started.

        ``_run_client`` exits when discord.py gives up for good (invalidated
        token, unrecoverable close) while ``_running`` stays True, so the base
        flag alone would keep reporting a healthy channel forever. Mirrors
        ``FeishuChannel.is_running`` so ``ChannelService.ensure_channel_ready``
        can restart the channel after its client thread dies.
        """
        if not self._running:
            return False
        return self._thread is not None and self._thread.is_alive()

    async def _run_on_discord_loop(self, coro, *, timeout: float = DISCORD_OUTBOUND_TIMEOUT_SECONDS):
        """Schedule *coro* on the Discord loop and await it with a bound.

        The Discord client runs on a dedicated thread whose loop is stopped but
        not closed when the client dies, so ``call_soon_threadsafe`` keeps
        queueing callbacks that never run and an unbounded ``wrap_future``
        await would hang a ChannelManager worker forever. ``stop()`` already
        bounds its identical cross-loop awaits with ``wait_for``; this extends
        that pattern to the outbound path. Failing fast when the loop is
        missing or not running turns a dead client into a logged send failure
        instead of a wedged worker.
        """
        loop = self._discord_loop
        if loop is None or not loop.is_running():
            coro.close()
            raise RuntimeError("Discord client event loop is not running")
        future = asyncio.run_coroutine_threadsafe(coro, loop)
        try:
            return await asyncio.wait_for(asyncio.wrap_future(future), timeout=timeout)
        except TimeoutError:
            future.cancel()
            raise

    async def send(self, msg: OutboundMessage) -> None:
        # Stop typing indicator once we're sending the response
        await self._run_on_discord_loop(self._stop_typing(msg.chat_id, msg.thread_ts))

        target = await self._resolve_target(msg)
        if target is None:
            logger.error("[Discord] target not found for chat_id=%s thread_ts=%s", msg.chat_id, msg.thread_ts)
            return

        text = msg.text or ""
        for chunk in self._split_text(text):
            await self._run_on_discord_loop(target.send(chunk))

    async def send_file(self, msg: OutboundMessage, attachment: ResolvedAttachment) -> bool:
        await self._run_on_discord_loop(self._stop_typing(msg.chat_id, msg.thread_ts))

        target = await self._resolve_target(msg)
        if target is None:
            logger.error("[Discord] target not found for file upload chat_id=%s thread_ts=%s", msg.chat_id, msg.thread_ts)
            return False

        if self._discord_module is None:
            return False

        try:
            # Read the attachment off the event loop (open + read are blocking IO),
            # then hand discord.py an in-memory buffer. The bytes are consumed while
            # ``target.send`` runs on ``_discord_loop``; once that future resolves the
            # buffer can be reclaimed, so this avoids leaking a file handle on both the
            # success and failure paths.
            data = await asyncio.to_thread(self._read_attachment_bytes, str(attachment.actual_path))
            file = self._discord_module.File(io.BytesIO(data), filename=attachment.filename)
            await self._run_on_discord_loop(target.send(file=file), timeout=DISCORD_UPLOAD_TIMEOUT_SECONDS)
            logger.info("[Discord] file uploaded: %s", attachment.filename)
            return True
        except Exception:
            logger.exception("[Discord] failed to upload file: %s", attachment.filename)
            return False

    async def _start_typing(self, channel, chat_id: str, thread_ts: str | None = None) -> asyncio.Task[None] | None:
        """Starts a loop to send periodic typing indicators.

        Returns the loop's task when this call started it, ``None`` when the
        channel is stopping or the target already shows typing.
        """
        if not self._running:
            return None
        target_id = thread_ts or chat_id
        if target_id in self._typing_tasks:
            return None  # Already typing for this target

        async def _typing_loop():
            # The loop's first failure logs at WARNING, so an indicator that
            # never works (missing permission, sustained 429s) shows at the
            # default level; later ticks drop to DEBUG so retries cannot flood.
            failure_level = logging.WARNING
            try:
                while True:
                    try:
                        # discord.py 2.x removed ``trigger_typing()``; awaiting
                        # ``typing()`` sends one indicator (~10s on screen).
                        await channel.typing()
                    except Exception:
                        logger.log(failure_level, "[Discord] failed to send typing indicator to %s", target_id, exc_info=True)
                        failure_level = logging.DEBUG
                    await asyncio.sleep(10)
            except asyncio.CancelledError:
                pass

        task = asyncio.create_task(_typing_loop())
        self._typing_tasks[target_id] = task
        task.add_done_callback(lambda done: self._typing_dependents.pop(done, None))
        return task

    async def _cancel_typing_tasks(self) -> None:
        """Cancel and await every typing task on their owning event loop."""
        typing_tasks = list(self._typing_tasks.items())
        for target_id, task in typing_tasks:
            if not task.done():
                task.cancel()
            logger.debug("[Discord] cancelled typing task for target %s", target_id)
        if typing_tasks:
            await asyncio.gather(*(task for _, task in typing_tasks), return_exceptions=True)
        self._typing_tasks.clear()
        self._typing_dependents.clear()

    def _discard_typing_tasks(self) -> None:
        """Forget stale typing tasks after their owning event loop has stopped."""
        for target_id, task in self._typing_tasks.items():
            if not task.done():
                logger.warning(
                    "[Discord] discarding pending typing task for stopped-loop target %s",
                    target_id,
                )
        self._typing_tasks.clear()
        self._typing_dependents.clear()

    async def _cancel_ack_reaction_tasks(self) -> None:
        """Cancel in-flight ack reactions so stop() does not strand them.

        A task interrupted mid-HTTP-call would otherwise stay in the
        retention set forever, pinning this channel instance and the discord
        Message object graph after shutdown.
        """
        pending = [task for task in self._ack_reaction_tasks if not task.done()]
        for task in pending:
            task.cancel()
        if pending:
            await asyncio.gather(*pending, return_exceptions=True)

    def _discard_ack_reaction_tasks(self) -> None:
        """Forget stale ack-reaction tasks after the owning loop stopped."""
        for task in list(self._ack_reaction_tasks):
            if not task.done():
                logger.warning("[Discord] discarding pending ack-reaction task for stopped loop")
        self._ack_reaction_tasks.clear()

    async def _cancel_ephemeral_tasks(self) -> None:
        """Cancel typing indicators and in-flight ack reactions together."""
        await self._cancel_typing_tasks()
        await self._cancel_ack_reaction_tasks()

    def _discard_ephemeral_tasks(self) -> None:
        """Forget stale typing and ack-reaction tasks (stopped-loop fallback)."""
        self._discard_typing_tasks()
        self._discard_ack_reaction_tasks()

    async def _stop_typing(self, chat_id: str, thread_ts: str | None = None) -> None:
        """Stops the typing loop for a specific target."""
        target_id = thread_ts or chat_id
        task = self._typing_tasks.pop(target_id, None)
        if task and not task.done():
            task.cancel()
            logger.debug("[Discord] stopped typing indicator for target %s", target_id)

    async def _add_reaction(self, message) -> None:
        """Add a checkmark reaction to acknowledge the message was received."""
        try:
            await message.add_reaction("✅")
        except Exception:
            logger.debug("[Discord] failed to add reaction to message %s", message.id, exc_info=True)

    def _on_ack_reaction_task_done(self, task: asyncio.Task[None]) -> None:
        self._ack_reaction_tasks.discard(task)
        if task.cancelled():
            return
        exc = task.exception()
        if exc is not None:
            logger.error("[Discord] ack reaction task failed: %s", exc)

    def _schedule_ack_reaction(self, message) -> asyncio.Task[None]:
        """Schedule the ack reaction with a strong task reference.

        The delayed HTTP call must survive until completion; see
        ``_ack_reaction_tasks`` for why a bare ``create_task`` is unsafe here.
        """
        task = asyncio.create_task(self._add_reaction(message))
        self._ack_reaction_tasks.add(task)
        task.add_done_callback(self._on_ack_reaction_task_done)
        return task

    async def _on_message(self, message) -> None:
        if not self._running or not self._client:
            return

        if message.author.bot:
            return

        if self._client.user and message.author.id == self._client.user.id:
            return

        guild = message.guild
        if self._allowed_guilds:
            if guild is None or guild.id not in self._allowed_guilds:
                return

        text = (message.content or "").strip()
        if not text:
            return

        if self._discord_module is None:
            return

        # Determine whether the bot is mentioned in this message
        user = self._client.user if self._client else None
        if user:
            bot_mention = user.mention  # <@ID>
            alt_mention = f"<@!{user.id}>"  # <@!ID> (ping variant)
            standard_mention = f"<@{user.id}>"
        else:
            bot_mention = None
            alt_mention = None
            standard_mention = ""
        has_mention = (bot_mention and bot_mention in message.content) or (alt_mention and alt_mention in message.content) or (standard_mention and standard_mention in message.content)

        # Strip mention from text for processing
        if has_mention:
            text = text.replace(bot_mention or "", "").replace(alt_mention or "", "").replace(standard_mention or "", "").strip()
            # Don't return early if text is empty — still process the mention (e.g., create thread)

        connect_code = self._pending_connect_code(text)
        if connect_code and await self._bind_connection_from_connect_code(message, connect_code):
            return

        # /connect is a local control-plane command and is consumed above. For
        # ordinary messages, reserve the shared intake slot before any Discord
        # API call, thread mapping write, or connection-identity query. The
        # provisional routing fields are used only for overload logging; the
        # final message is built after thread routing completes.
        msg_type = InboundMessageType.COMMAND if is_known_channel_command(text) else InboundMessageType.CHAT
        provisional = self._make_inbound(
            chat_id=str(message.channel.id),
            user_id=str(message.author.id),
            text=text,
            msg_type=msg_type,
            metadata={
                "guild_id": str(guild.id) if guild else None,
                "channel_id": str(message.channel.id),
                "message_id": str(message.id),
            },
        )
        reservation = self._reserve_inbound(provisional)
        if reservation is None:
            return

        await self._handle_admitted_message(
            message,
            text=text,
            msg_type=msg_type,
            guild=guild,
            has_mention=bool(has_mention),
            reservation=reservation,
        )

    async def _handle_admitted_message(
        self,
        message,
        *,
        text: str,
        msg_type: InboundMessageType,
        guild,
        has_mention: bool,
        reservation: InboundReservation,
    ) -> None:
        """Finish Discord routing while owning one bounded intake slot."""
        reservation_transferred = False
        try:
            # --- Determine thread/channel routing and typing target ---
            thread_id = None
            chat_id = None
            typing_target = None  # The Discord object to type into

            if isinstance(message.channel, self._discord_module.Thread):
                # --- Message already inside a thread ---
                thread_obj = message.channel
                thread_id = str(thread_obj.id)
                chat_id = str(thread_obj.parent_id or thread_obj.id)
                typing_target = thread_obj

                # If this is a known active thread, process normally
                if thread_id in self._active_thread_ids:
                    inbound = self._make_inbound(
                        chat_id=chat_id,
                        user_id=str(message.author.id),
                        text=text,
                        msg_type=msg_type,
                        thread_ts=thread_id,
                        metadata={
                            "guild_id": str(guild.id) if guild else None,
                            "channel_id": str(message.channel.id),
                            "message_id": str(message.id),
                        },
                    )
                    inbound.topic_id = thread_id
                    handoff = self._publish_reserved(inbound, reservation, guild_id=str(guild.id) if guild else None)
                    if handoff is None:
                        return
                    reservation_transferred = True
                    await self._acknowledge_after_handoff(message, handoff, typing_target=typing_target, chat_id=chat_id, thread_id=thread_id)
                    return

                # Thread not tracked (orphaned) — create new thread and handle below
                logger.debug("[Discord] message in orphaned thread %s, will create new thread", thread_id)
                thread_id = None
                typing_target = None

            # At this point we're guaranteed to be in a channel, not a thread
            # (the Thread case is handled above). Apply mention_only for all
            # non-thread messages — no special case needed.
            channel_id = str(message.channel.id)

            # Check if there's an active thread for this channel
            if channel_id in self._active_threads:
                # respect mention_only: if enabled, only process messages that mention the bot
                # (unless the channel is in allowed_channels)
                # Messages within a thread are always allowed through (continuation).
                # At this code point we know the message is in a channel, not a thread
                # (Thread case handled above), so always apply the check.
                if self._mention_only and not has_mention and channel_id not in self._allowed_channels:
                    logger.debug("[Discord] skipping no-@ message in channel %s (not in thread)", channel_id)
                    return
                # mention_only + fresh @ → create new thread instead of routing to existing one
                if self._mention_only and has_mention:
                    thread_obj = await self._create_thread(message)
                    if thread_obj is not None:
                        target_thread_id = str(thread_obj.id)
                        self._record_thread_mapping(channel_id, target_thread_id)
                        await asyncio.to_thread(self._persist_thread_mappings)
                        thread_id = target_thread_id
                        chat_id = channel_id
                        typing_target = thread_obj
                        logger.info("[Discord] created new thread %s in channel %s on mention (replacing existing thread)", target_thread_id, channel_id)
                    else:
                        logger.info("[Discord] thread creation failed in channel %s, falling back to channel replies", channel_id)
                        thread_id = channel_id
                        chat_id = channel_id
                        typing_target = message.channel
                else:
                    # Existing session → route to the existing thread
                    target_thread_id = self._active_threads[channel_id]
                    logger.debug("[Discord] routing message in channel %s to existing thread %s", channel_id, target_thread_id)
                    thread_id = target_thread_id
                    chat_id = channel_id
                    typing_target = await self._get_channel_or_thread(target_thread_id)
            elif self._mention_only and not has_mention and channel_id not in self._allowed_channels:
                # Not mentioned and not in an allowed channel → skip
                logger.debug("[Discord] skipping message without mention in channel %s", channel_id)
                return
            elif self._mention_only and has_mention:
                # First mention in this channel → create thread
                thread_obj = await self._create_thread(message)
                if thread_obj is not None:
                    target_thread_id = str(thread_obj.id)
                    self._record_thread_mapping(channel_id, target_thread_id)
                    await asyncio.to_thread(self._persist_thread_mappings)
                    thread_id = target_thread_id
                    chat_id = channel_id
                    typing_target = thread_obj  # Type into the new thread
                    logger.info("[Discord] created thread %s in channel %s for user %s", target_thread_id, channel_id, message.author.display_name)
                else:
                    # Fallback: thread creation failed (disabled/permissions), reply in channel
                    logger.info("[Discord] thread creation failed in channel %s, falling back to channel replies", channel_id)
                    thread_id = channel_id
                    chat_id = channel_id
                    typing_target = message.channel  # Type into the channel
            elif self._thread_mode:
                # thread_mode but mention_only is False → create thread anyway for conversation grouping
                thread_obj = await self._create_thread(message)
                if thread_obj is None:
                    # Thread creation failed (disabled/permissions), fall back to channel replies
                    logger.info("[Discord] thread creation failed in channel %s, falling back to channel replies", channel_id)
                    thread_id = channel_id
                    chat_id = channel_id
                    typing_target = message.channel  # Type into the channel
                else:
                    target_thread_id = str(thread_obj.id)
                    self._record_thread_mapping(channel_id, target_thread_id)
                    await asyncio.to_thread(self._persist_thread_mappings)
                    thread_id = target_thread_id
                    chat_id = channel_id
                    typing_target = thread_obj  # Type into the new thread
            else:
                # No threading — reply directly in channel
                thread_id = channel_id
                chat_id = channel_id
                typing_target = message.channel  # Type into the channel

            inbound = self._make_inbound(
                chat_id=chat_id,
                user_id=str(message.author.id),
                text=text,
                msg_type=msg_type,
                thread_ts=thread_id,
                metadata={
                    "guild_id": str(guild.id) if guild else None,
                    "channel_id": str(message.channel.id),
                    "message_id": str(message.id),
                },
            )
            inbound.topic_id = thread_id

            handoff = self._publish_reserved(inbound, reservation, guild_id=str(guild.id) if guild else None)
            if handoff is None:
                return
            reservation_transferred = True
            await self._acknowledge_after_handoff(message, handoff, typing_target=typing_target, chat_id=chat_id, thread_id=thread_id)
        finally:
            if not reservation_transferred:
                reservation.release()

    def _publish_reserved(self, inbound: InboundMessage, reservation: InboundReservation, *, guild_id: str | None) -> Future[bool] | None:
        """Transfer an already-reserved message to the Gateway loop.

        Called from discord.py's private event-loop thread. The connection
        identity is resolved on the Gateway loop as well, because the
        repository's SQLAlchemy engine and pool belong to that loop: awaiting
        them from the Discord loop fails under asyncpg ("attached to a
        different loop") and binds the pool's wait queue to the wrong loop
        under aiosqlite, breaking the Gateway's own queries.

        Returns the hand-off's completion future (``True`` once committed), or
        ``None`` when the Gateway loop refused it.
        """
        handoff = self._submit_threadsafe_coroutine_future(
            self._commit_reserved_inbound_with_identity(inbound, reservation, guild_id=guild_id),
            self._main_loop,
            name="commit_reserved_inbound",
            msg_id=inbound.metadata.get("message_id"),
            reservation=reservation,
        )
        if handoff is None:
            logger.info("[Discord] main loop stopped before reserved inbound could be scheduled")
        return handoff

    async def _acknowledge_after_handoff(
        self,
        message,
        handoff: Future[bool],
        *,
        typing_target,
        chat_id: str,
        thread_id: str | None,
    ) -> None:
        """Show progress for a handed-off message, undoing it if the hand-off fails.

        Typing starts before the hand-off resolves, as it did before the
        identity lookup moved to the Gateway loop: ``_start_typing`` registers
        without yielding, so a fast reply's ``_stop_typing`` cannot run first
        and leave an indicator behind. The ack reaction waits for the commit.
        If identity resolution or the commit fails, or ``stop()`` drains the
        hand-off, the message was dropped and gets no ack. Another message to
        the same target may have reused the indicator meanwhile, pending or
        already committed, so a failed hand-off stops it only when this message
        started it and no other message still depends on it.
        """
        target_id = thread_id or chat_id
        started_task = await self._start_typing(typing_target, chat_id, thread_id) if typing_target else None
        typing_task = self._typing_tasks.get(target_id) if typing_target else None
        if typing_task is not None:
            self._typing_dependents[typing_task] = self._typing_dependents.get(typing_task, 0) + 1
        try:
            committed = await asyncio.shield(asyncio.wrap_future(handoff))
        except asyncio.CancelledError:
            if not handoff.cancelled():
                raise
            committed = False  # drained by stop()
        except Exception:
            committed = False  # the submission finalizer already logged it
        if committed:
            # Keep this message's hold until a reply stops the indicator.
            self._schedule_ack_reaction(message)
            return
        if typing_task is None:
            return
        remaining = self._typing_dependents.get(typing_task, 1) - 1
        if remaining > 0:
            self._typing_dependents[typing_task] = remaining
            return
        self._typing_dependents.pop(typing_task, None)
        if typing_task is started_task and self._typing_tasks.get(target_id) is typing_task:
            await self._stop_typing(chat_id, thread_id)

    async def _commit_reserved_inbound_with_identity(
        self,
        inbound: InboundMessage,
        reservation: InboundReservation,
        *,
        guild_id: str | None,
    ) -> bool:
        """Attach connection identity and commit the reservation on the Gateway loop."""
        inbound = await self._attach_connection_identity(inbound, guild_id=guild_id)
        return self._commit_reserved_inbound(reservation, inbound)

    async def _attach_connection_identity(self, inbound: InboundMessage, guild_id: str | None = None) -> InboundMessage:
        return await attach_connection_identity(
            inbound,
            repo=self._connection_repo,
            provider="discord",
            workspace_id=guild_id,
            fallback_without_workspace=True,
        )

    async def _bind_connection_from_connect_code(self, message, code: str) -> bool:
        """Dispatch the bind flow to the Gateway loop, where the repository's engine lives."""
        if self._connection_repo is None or not code:
            return False

        scheduled = self._submit_threadsafe_coroutine(
            self._bind_connection_from_connect_code_on_main(message, code),
            self._main_loop,
            name="bind_connection",
            msg_id=getattr(message, "id", None),
        )
        if not scheduled:
            logger.warning("[Discord] main loop not running, cannot bind channel connection")
        # Handled either way so a bind attempt never reaches the agent as a chat
        # turn; an unscheduled bind leaves the code unconsumed, so the user can
        # retry it after the Gateway restarts.
        return True

    async def _bind_connection_from_connect_code_on_main(self, message, code: str) -> bool:
        """Run the bind flow on the Gateway loop; replies go back through the Discord loop."""
        state = await self._connection_repo.consume_oauth_state(provider="discord", state=code)
        if state is None:
            await self._run_on_discord_loop(self._send_connection_reply(message, "Discord connection code is invalid or expired."))
            return True

        guild = getattr(message, "guild", None)
        channel = getattr(message, "channel", None)
        author = getattr(message, "author", None)
        user_id = str(getattr(author, "id", "") or "")
        if not user_id:
            await self._run_on_discord_loop(self._send_connection_reply(message, "Discord connection could not be completed from this message."))
            return True

        guild_id = str(getattr(guild, "id", "") or "") or None
        await self._connection_repo.upsert_connection(
            owner_user_id=state["owner_user_id"],
            provider="discord",
            external_account_id=user_id,
            external_account_name=getattr(author, "display_name", None) or getattr(author, "name", None),
            workspace_id=guild_id,
            workspace_name=getattr(guild, "name", None) if guild is not None else None,
            metadata={
                "guild_id": guild_id,
                "channel_id": str(getattr(channel, "id", "") or ""),
            },
            status="connected",
        )
        await self._run_on_discord_loop(self._send_connection_reply(message, "Discord connected to DeerFlow."))
        return True

    @staticmethod
    async def _send_connection_reply(message, text: str) -> None:
        channel = getattr(message, "channel", None)
        send = getattr(channel, "send", None)
        if send is None:
            return
        try:
            await send(text)
        except Exception:
            logger.exception("[Discord] failed to send connection reply")

    def _run_client(self) -> None:
        self._discord_loop = asyncio.new_event_loop()
        asyncio.set_event_loop(self._discord_loop)
        try:
            self._discord_loop.run_until_complete(self._client.start(self._bot_token))
        except Exception:
            if self._running:
                logger.exception("Discord client error")
        finally:
            try:
                self._discord_loop.run_until_complete(self._cancel_ephemeral_tasks())
            except Exception:
                logger.exception("Error while cleaning up Discord ephemeral tasks")
            try:
                if self._client and not self._client.is_closed():
                    self._discord_loop.run_until_complete(self._client.close())
            except Exception:
                logger.exception("Error during Discord shutdown")

    async def _create_thread(self, message):
        try:
            if self._discord_module is None:
                return None

            # Only TextChannel (type 0) and NewsChannel (type 10) support threads
            channel_type = message.channel.type
            if channel_type not in (
                self._discord_module.ChannelType.text,
                self._discord_module.ChannelType.news,
            ):
                logger.info(
                    "[Discord] channel type %s (%s) does not support threads",
                    channel_type.value,
                    channel_type.name,
                )
                return None

            thread_name = f"deerflow-{message.author.display_name}-{message.id}"[:100]
            return await message.create_thread(name=thread_name)
        except self._discord_module.errors.HTTPException as exc:
            if exc.code == 50024:
                logger.info(
                    "[Discord] cannot create thread in channel %s (error code 50024): %s",
                    message.channel.id,
                    channel_type.name if (channel_type := message.channel.type) else "unknown",
                )
            else:
                logger.exception(
                    "[Discord] failed to create thread for message=%s (HTTPException %s)",
                    message.id,
                    exc.code,
                )
            return None
        except Exception:
            logger.exception("[Discord] failed to create thread for message=%s (threads may be disabled or missing permissions)", message.id)
            return None

    async def _resolve_target(self, msg: OutboundMessage):
        if not self._client or not self._discord_loop:
            return None

        target_ids: list[str] = []
        if msg.thread_ts:
            target_ids.append(msg.thread_ts)
        if msg.chat_id and msg.chat_id not in target_ids:
            target_ids.append(msg.chat_id)

        for raw_id in target_ids:
            target = await self._get_channel_or_thread(raw_id)
            if target is not None:
                return target
        return None

    async def _get_channel_or_thread(self, raw_id: str):
        if not self._client or not self._discord_loop:
            return None

        try:
            target_id = int(raw_id)
        except (TypeError, ValueError):
            return None

        try:
            return await self._run_on_discord_loop(self._fetch_channel(target_id))
        except Exception:
            logger.exception("[Discord] failed to resolve target id=%s", raw_id)
            return None

    async def _fetch_channel(self, target_id: int):
        if not self._client:
            return None

        channel = self._client.get_channel(target_id)
        if channel is not None:
            return channel

        try:
            return await self._client.fetch_channel(target_id)
        except Exception:
            return None

    @staticmethod
    def _split_text(text: str) -> list[str]:
        if not text:
            return [""]

        chunks: list[str] = []
        remaining = text
        while len(remaining) > _DISCORD_MAX_MESSAGE_LEN:
            split_at = remaining.rfind("\n", 0, _DISCORD_MAX_MESSAGE_LEN)
            if split_at <= 0:
                split_at = _DISCORD_MAX_MESSAGE_LEN
            else:
                # Keep the delimiter on this chunk's tail so consecutive
                # Discord messages reconstruct the original text exactly.
                split_at += 1
            chunks.append(remaining[:split_at])
            remaining = remaining[split_at:]

        if remaining:
            chunks.append(remaining)

        return chunks