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.allowed_users import parse_channel_ids
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
from deerflow.utils.file_io import await_drained

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


def _parse_discord_guild_id(entry: Any) -> int | None:
    """Return a Discord guild (snowflake) ID, or ``None`` if ``entry`` is not one."""
    if isinstance(entry, int) and not isinstance(entry, bool) and entry > 0:
        return entry
    if isinstance(entry, str):
        text = entry.strip()
        if text.isascii() and text.isdigit():
            return int(text) or None
    return None


def _parse_allowed_guilds(allowed_guilds: Any) -> frozenset[int] | None:
    """Parse ``channels.discord.allowed_guilds``; ``None`` means no allowlist.

    A single ID is shorthand for a one-entry list, the form
    ``config.example.yaml`` advertises. Iterating a scalar instead turns
    ``"321"`` into guilds 1, 2 and 3 and raises ``TypeError`` on an unquoted
    YAML int. Entries that are not positive numeric snowflakes are dropped with
    a warning. If none remain, the result is an empty set that denies every
    guild: the operator asked for a restriction, so an unreadable one must not
    open the bot to every server it joined.
    """
    if allowed_guilds is None or (isinstance(allowed_guilds, str) and not allowed_guilds.strip()):
        return None
    entries = list(allowed_guilds) if isinstance(allowed_guilds, list | tuple | set) else [allowed_guilds]
    if not entries:
        return None
    guild_ids: set[int] = set()
    for entry in entries:
        guild_id = _parse_discord_guild_id(entry)
        if guild_id is None:
            logger.warning("[Discord] Ignoring allowed_guilds entry %r: expected a positive numeric guild ID; list several IDs as a YAML list", entry)
        else:
            guild_ids.add(guild_id)
    if not guild_ids:
        logger.error("[Discord] allowed_guilds has no valid numeric guild ID; denying every guild until it is fixed")
    return frozenset(guild_ids)


def _parse_allowed_channels(allowed_channels: Any) -> frozenset[str]:
    """Parse ``channels.discord.allowed_channels``; a single ID is one entry.

    Channel IDs are snowflakes that only ever get compared as strings here, so
    the entries are kept verbatim rather than validated — an unreadable ID just
    never matches, which leaves ``mention_only`` in force (fail-closed).
    """
    return parse_channel_ids(allowed_channels, logger=logger, channel_name="Discord", key_name="allowed_channels")


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, or a
          single ID. Empty = allow all; a non-empty list with no valid ID denies
          every guild.
        - ``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), or a single ID. 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: frozenset[int] | None = _parse_allowed_guilds(config.get("allowed_guilds"))
        self._mention_only: bool = bool(config.get("mention_only", False))
        self._thread_mode: bool = config.get("thread_mode", self._mention_only)
        self._allowed_channels: frozenset[str] = _parse_allowed_channels(config.get("allowed_channels"))

        # Session tracking: channel_id -> Discord thread_id (in-memory, persisted to JSON).
        # Uses a dedicated JSON file next to the channel store's state, separate from
        # the ChannelStore itself, which maps IM conversations to DeerFlow thread IDs.
        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
        channels_dir = getattr(config.get("channel_store"), "channels_dir", None)
        if channels_dir is not None:
            self._thread_store_path = Path(channels_dir) / "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")

    async def _persist_thread_mappings_async(self) -> None:
        """Persist mappings without letting cancellation detach the worker thread."""
        await await_drained(asyncio.to_thread(self._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 self._persist_thread_mappings_async()
            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

    def _check_guild(self, guild) -> bool:
        if self._allowed_guilds is None:
            return True
        return guild is not None and guild.id in self._allowed_guilds

    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 not self._check_guild(guild):
            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 self._persist_thread_mappings_async()
                        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 self._persist_thread_mappings_async()
                    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 self._persist_thread_mappings_async()
                    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