deer-flow/backend/tests/test_discord_channel.py

"""Tests for Discord channel integration wiring."""

from __future__ import annotations

import asyncio
import builtins
import gc
import json
import threading
import weakref
from types import SimpleNamespace
from unittest.mock import AsyncMock, patch

import pytest

from app.channels.discord import DiscordChannel
from app.channels.manager import CHANNEL_CAPABILITIES
from app.channels.message_bus import InboundMessage, InboundMessageType, MessageBus, OutboundMessage, ResolvedAttachment
from app.channels.service import _CHANNEL_REGISTRY


def test_discord_channel_registered() -> None:
    assert "discord" in _CHANNEL_REGISTRY


def test_discord_channel_capabilities() -> None:
    assert "discord" in CHANNEL_CAPABILITIES


def test_discord_channel_init() -> None:
    bus = MessageBus()
    channel = DiscordChannel(bus=bus, config={"bot_token": "token"})

    assert channel.name == "discord"


# ---------------------------------------------------------------------------
# thread-mapping persistence across restart (#2897)
# ---------------------------------------------------------------------------


def test_discord_thread_mapping_persists_across_restart(tmp_path) -> None:
    """A channel->thread mapping written before shutdown is restored on a
    subsequent start, so conversations are not lost across restarts (#2897)."""
    store_path = tmp_path / "discord_threads.json"

    # First process lifetime: record and persist a mapping.
    first = DiscordChannel(bus=MessageBus(), config={"bot_token": "token"})
    first._thread_store_path = store_path
    first._record_thread_mapping("chan-1", "thread-1")
    first._persist_thread_mappings()
    assert store_path.exists()
    assert json.loads(store_path.read_text()) == {"chan-1": "thread-1"}

    # Restart: a brand-new channel instance reads the same file.
    second = DiscordChannel(bus=MessageBus(), config={"bot_token": "token"})
    second._thread_store_path = store_path
    second._active_threads.clear()
    second._active_thread_ids.clear()
    second._load_active_threads()

    assert second._active_threads == {"chan-1": "thread-1"}
    assert "thread-1" in second._active_thread_ids


@pytest.mark.asyncio
async def test_discord_stop_flushes_thread_mappings(tmp_path) -> None:
    """stop() best-effort flushes in-memory thread mappings to disk so the
    most recent mapping survives a hard shutdown."""
    channel = DiscordChannel(bus=MessageBus(), config={"bot_token": "token"})
    channel._thread_store_path = tmp_path / "discord_threads.json"
    # stop() only flushes once the load has marked the in-memory map
    # authoritative (see DiscordChannel._thread_store_loaded); simulate the
    # normal post-start() state so this exercises the flush, not the guard.
    channel._thread_store_loaded = True
    # Minimal shutdown context: no live client/loop/thread to tear down.
    channel._discord_loop = None
    channel._client = None
    channel._thread = None
    channel._cancel_ephemeral_tasks = AsyncMock()

    channel._record_thread_mapping("chan-2", "thread-2")
    await channel.stop()

    assert channel._thread_store_path.exists()
    assert json.loads(channel._thread_store_path.read_text()) == {"chan-2": "thread-2"}


@pytest.mark.asyncio
async def test_discord_stop_does_not_clobber_store_before_load(tmp_path) -> None:
    """stop() before the initial load must not overwrite the persisted file.

    ``ChannelService`` deliberately stops a channel whose ``start()`` bailed
    before ``_load_active_threads()`` ran (missing bot_token / discord import
    error), when the in-memory map is still empty. An ungated flush would write
    ``{}`` over the persisted mappings — the #2897 data loss this PR exists to
    prevent.
    """
    store_path = tmp_path / "discord_threads.json"
    store_path.write_text(json.dumps({"chan-9": "thread-9"}))

    channel = DiscordChannel(bus=MessageBus(), config={"bot_token": "token"})
    channel._thread_store_path = store_path
    # start() bailed before the load, so the flag is still False.
    assert channel._thread_store_loaded is False
    # Minimal shutdown context: no live client/loop/thread to tear down.
    channel._discord_loop = None
    channel._client = None
    channel._thread = None
    channel._cancel_ephemeral_tasks = AsyncMock()

    await channel.stop()

    # The pre-existing mapping survives intact: nothing was flushed over it.
    assert json.loads(store_path.read_text()) == {"chan-9": "thread-9"}


def _make_discord_message(text: str):
    return SimpleNamespace(
        id=111,
        content=text,
        author=SimpleNamespace(id=123, bot=False, display_name="alice"),
        guild=SimpleNamespace(id=321),
        channel=SimpleNamespace(id=456),
        add_reaction=lambda _emoji: None,
    )


@pytest.mark.asyncio
async def test_discord_bot_mention_slash_skill_routes_as_chat() -> None:
    bus = MessageBus()
    channel = DiscordChannel(bus=bus, config={"bot_token": "token"})
    channel._running = True
    channel._client = SimpleNamespace(user=SimpleNamespace(id=999, mention="<@999>"))
    channel._discord_module = SimpleNamespace(Thread=type("FakeThread", (), {}))
    channel._main_loop = asyncio.get_running_loop()

    async def noop(*_args, **_kwargs):
        return None

    channel._start_typing = noop
    channel._add_reaction = noop

    await channel._on_message(_make_discord_message("<@999> /data-analysis analyze uploads/foo.csv"))
    await asyncio.sleep(0)

    inbound = bus.get_inbound_nowait()
    bus.inbound_task_done()
    assert inbound.text == "/data-analysis analyze uploads/foo.csv"
    assert inbound.msg_type == InboundMessageType.CHAT
    assert inbound.topic_id == "456"


@pytest.mark.asyncio
async def test_discord_bot_mention_known_command_routes_as_command() -> None:
    bus = MessageBus()
    channel = DiscordChannel(bus=bus, config={"bot_token": "token"})
    channel._running = True
    channel._client = SimpleNamespace(user=SimpleNamespace(id=999, mention="<@999>"))
    channel._discord_module = SimpleNamespace(Thread=type("FakeThread", (), {}))
    channel._main_loop = asyncio.get_running_loop()

    async def noop(*_args, **_kwargs):
        return None

    channel._start_typing = noop
    channel._add_reaction = noop

    await channel._on_message(_make_discord_message("<@999> /help"))
    await asyncio.sleep(0)

    inbound = bus.get_inbound_nowait()
    bus.inbound_task_done()
    assert inbound.text == "/help"
    assert inbound.msg_type == InboundMessageType.COMMAND
    assert inbound.topic_id == "456"


@pytest.mark.asyncio
async def test_discord_full_queue_rejects_before_thread_or_identity_side_effects() -> None:
    bus = MessageBus(inbound_queue_maxsize=1)
    await bus.publish_inbound(
        InboundMessage(
            channel_name="slack",
            chat_id="C1",
            user_id="U1",
            text="already queued",
        )
    )
    channel = DiscordChannel(bus=bus, config={"bot_token": "token", "thread_mode": True})
    channel._running = True
    channel._client = SimpleNamespace(user=SimpleNamespace(id=999, mention="<@999>"))
    channel._discord_module = SimpleNamespace(Thread=type("FakeThread", (), {}))
    channel._main_loop = asyncio.get_running_loop()
    channel._create_thread = AsyncMock()
    channel._attach_connection_identity = AsyncMock()

    await channel._on_message(_make_discord_message("hello"))

    channel._create_thread.assert_not_awaited()
    channel._attach_connection_identity.assert_not_awaited()
    assert channel._active_threads == {}
    assert bus.inbound_queue.qsize() == 1


@pytest.mark.asyncio
async def test_discord_releases_early_reservation_when_thread_creation_raises() -> None:
    bus = MessageBus(inbound_queue_maxsize=1)
    channel = DiscordChannel(bus=bus, config={"bot_token": "token", "thread_mode": True})
    channel._running = True
    channel._client = SimpleNamespace(user=SimpleNamespace(id=999, mention="<@999>"))
    channel._discord_module = SimpleNamespace(Thread=type("FakeThread", (), {}))
    channel._main_loop = asyncio.get_running_loop()
    channel._create_thread = AsyncMock(side_effect=RuntimeError("thread create failed"))

    with pytest.raises(RuntimeError, match="thread create failed"):
        await channel._on_message(_make_discord_message("hello"))

    # The exception path must return the capacity slot to the shared bus.
    await bus.publish_inbound(
        InboundMessage(
            channel_name="slack",
            chat_id="C1",
            user_id="U1",
            text="capacity was released",
        )
    )
    assert bus.inbound_queue.qsize() == 1


# ---------------------------------------------------------------------------
# send_file file-handle lifecycle
# ---------------------------------------------------------------------------


def _start_bg_loop() -> tuple[asyncio.AbstractEventLoop, threading.Thread]:
    """Spin up a real background event loop, mirroring ``DiscordChannel._discord_loop``.

    ``send_file`` schedules work onto ``_discord_loop`` via
    ``run_coroutine_threadsafe`` and awaits the result with ``wrap_future``, so a
    real running loop is the most faithful way to exercise that path.
    """
    loop = asyncio.new_event_loop()
    ready = threading.Event()

    def _runner() -> None:
        loop.call_soon(ready.set)
        loop.run_forever()

    thread = threading.Thread(target=_runner, daemon=True)
    thread.start()
    ready.wait()
    return loop, thread


def _stop_bg_loop(loop: asyncio.AbstractEventLoop, thread: threading.Thread) -> None:
    loop.call_soon_threadsafe(loop.stop)
    thread.join(timeout=5)
    loop.close()


def _build_send_file_channel(bg_loop: asyncio.AbstractEventLoop) -> DiscordChannel:
    channel = DiscordChannel(bus=MessageBus(), config={"bot_token": "token"})
    channel._discord_loop = bg_loop
    channel._discord_module = SimpleNamespace(File=lambda fp, filename=None: fp)

    async def _noop(*_args, **_kwargs):
        return None

    channel._stop_typing = _noop
    return channel


def _tracking_open():
    """Wrap ``builtins.open`` to record every handle it returns."""
    handles: list = []
    real_open = builtins.open

    def _open(path, *args, **kwargs):
        handle = real_open(path, *args, **kwargs)
        handles.append(handle)
        return handle

    return handles, _open


async def _noop_coro(*_args, **_kwargs):
    return None


def _resolve_to(target):
    async def _resolve_target(_msg):
        return target

    return _resolve_target


@pytest.mark.asyncio
async def test_send_file_closes_file_handle(tmp_path) -> None:
    """The file handle opened for upload is closed once send_file returns (success path)."""
    bg_loop, bg_thread = _start_bg_loop()
    try:
        channel = _build_send_file_channel(bg_loop)
        target = SimpleNamespace(send=_noop_coro)
        channel._resolve_target = _resolve_to(target)

        path = tmp_path / "upload.txt"
        path.write_bytes(b"hello")
        att = ResolvedAttachment("/mnt/user-data/outputs/upload.txt", path, "upload.txt", "text/plain", 5, False)
        msg = OutboundMessage(channel_name="discord", chat_id="c1", thread_id="t1", text="t")

        handles, tracking_open = _tracking_open()
        with patch("builtins.open", tracking_open):
            result = await channel.send_file(msg, att)

        assert result is True
        assert len(handles) == 1
        assert handles[0].closed is True
    finally:
        _stop_bg_loop(bg_loop, bg_thread)


@pytest.mark.asyncio
async def test_send_file_closes_handle_when_send_fails(tmp_path) -> None:
    """The file handle is still closed when target.send raises (failure path)."""
    bg_loop, bg_thread = _start_bg_loop()
    try:
        channel = _build_send_file_channel(bg_loop)

        async def _failing_send(*, file=None):
            raise RuntimeError("upload failed")

        target = SimpleNamespace(send=_failing_send)
        channel._resolve_target = _resolve_to(target)

        path = tmp_path / "upload.txt"
        path.write_bytes(b"hello")
        att = ResolvedAttachment("/mnt/user-data/outputs/upload.txt", path, "upload.txt", "text/plain", 5, False)
        msg = OutboundMessage(channel_name="discord", chat_id="c1", thread_id="t1", text="t")

        handles, tracking_open = _tracking_open()
        with patch("builtins.open", tracking_open):
            result = await channel.send_file(msg, att)

        assert result is False
        assert len(handles) == 1
        assert handles[0].closed is True
    finally:
        _stop_bg_loop(bg_loop, bg_thread)


@pytest.mark.asyncio
async def test_ack_reaction_task_retained_under_gc() -> None:
    """The ack-reaction task is retained and runs to completion.

    This pins the retention contract (scheduled → in the set, done →
    discarded): the fake coroutine never suspends on an unrooted future, so
    actual mid-flight GC cannot be reproduced deterministically here — the
    same limitation the #4928 precedent test has.

    A bare ``asyncio.create_task`` holds only a weak loop reference, so the
    ✅ acknowledgment could be garbage-collected mid-flight. The instance-level
    retention set keeps the task strongly referenced until completion (same
    pattern as #4928 / #4931).
    """
    bus = MessageBus()
    channel = DiscordChannel(bus=bus, config={"bot_token": "token"})

    reacted = asyncio.Event()

    async def _add_reaction(_emoji: str) -> None:
        reacted.set()

    message = SimpleNamespace(id=111, add_reaction=_add_reaction)

    task = channel._schedule_ack_reaction(message)
    weak_task = weakref.ref(task)
    del task
    gc.collect()

    retained = weak_task()
    assert retained is not None, "ack reaction task was garbage-collected mid-flight"
    assert retained in channel._ack_reaction_tasks

    await asyncio.wait_for(retained, timeout=1.0)
    assert reacted.is_set()
    # Completed tasks are discarded so the retention set cannot grow unboundedly.
    assert retained not in channel._ack_reaction_tasks


@pytest.mark.asyncio
async def test_ack_reaction_retention_is_isolated_per_channel() -> None:
    """One channel's shutdown must not cancel another instance's in-flight reactions.

    The retention set is instance-level (matching ``_typing_tasks``): a
    module-level set would let ``stop()`` on one channel drain every other
    channel's pending ack tasks.
    """
    first = DiscordChannel(bus=MessageBus(), config={"bot_token": "token"})
    second = DiscordChannel(bus=MessageBus(), config={"bot_token": "token"})

    release = asyncio.Event()

    async def _hang(_emoji: str) -> None:
        await release.wait()

    task = second._schedule_ack_reaction(SimpleNamespace(id=666, add_reaction=_hang))
    await asyncio.sleep(0)  # let the task start and suspend
    assert task in second._ack_reaction_tasks
    assert task not in first._ack_reaction_tasks

    # Full shutdown of the first channel must leave the second instance's
    # in-flight reaction alone (the review asked for independent shutdown).
    await first.stop()

    assert not task.done()
    assert task in second._ack_reaction_tasks

    release.set()
    await asyncio.wait_for(task, timeout=1.0)


@pytest.mark.asyncio
async def test_ack_reaction_task_survives_reaction_failure() -> None:
    """A failing add_reaction completes quietly and is discarded from the set."""

    bus = MessageBus()
    channel = DiscordChannel(bus=bus, config={"bot_token": "token"})

    async def _boom(_emoji: str) -> None:
        raise RuntimeError("discord api down")

    message = SimpleNamespace(id=222, add_reaction=_boom)

    task = channel._schedule_ack_reaction(message)
    await asyncio.wait_for(task, timeout=1.0)
    assert task not in channel._ack_reaction_tasks


@pytest.mark.asyncio
async def test_stop_drains_in_flight_ack_reaction_tasks() -> None:
    """stop()'s cleanup path cancels in-flight ack reactions on the owning loop.

    Without the drain, a task interrupted mid-HTTP-call would sit in the
    module retention set forever, pinning the channel and Message graph.
    """
    release = asyncio.Event()

    async def _hang(_emoji: str) -> None:
        await release.wait()

    channel = DiscordChannel(bus=MessageBus(), config={"bot_token": "token"})
    message = SimpleNamespace(id=333, add_reaction=_hang)

    task = channel._schedule_ack_reaction(message)
    assert task in channel._ack_reaction_tasks
    # Yield once so the task actually starts and suspends inside _hang —
    # cancelling a never-started task would not exercise the mid-flight path.
    await asyncio.sleep(0)
    assert not task.done()

    await channel._cancel_ephemeral_tasks()

    assert task.cancelled()
    assert task not in channel._ack_reaction_tasks


@pytest.mark.asyncio
async def test_ack_reaction_task_failure_is_logged_and_discarded(caplog) -> None:
    """An exception escaping _add_reaction is logged at error and discarded."""
    bus = MessageBus()
    channel = DiscordChannel(bus=bus, config={"bot_token": "token"})

    async def _explode(_self, _message) -> None:
        raise RuntimeError("unexpected boom")

    with patch.object(DiscordChannel, "_add_reaction", _explode), caplog.at_level("ERROR", logger="app.channels.discord"):
        task = channel._schedule_ack_reaction(SimpleNamespace(id=444))
        with pytest.raises(RuntimeError, match="unexpected boom"):
            await asyncio.wait_for(task, timeout=1.0)

    assert task not in channel._ack_reaction_tasks
    assert any("ack reaction task failed" in record.message for record in caplog.records)


@pytest.mark.asyncio
async def test_stop_wiring_drains_ack_tasks_across_loops() -> None:
    """The real stop() path (cross-loop branch) drains in-flight ack reactions.

    Guards the wiring itself: reverting stop() to typing-only cleanup must
    fail here, not just at the helper level.
    """
    bg_loop, bg_thread = _start_bg_loop()
    try:
        channel = DiscordChannel(bus=MessageBus(), config={"bot_token": "token"})
        channel._discord_loop = bg_loop

        release = asyncio.Event()

        async def _hang(_emoji: str) -> None:
            await release.wait()

        async def _schedule_on_bg_loop() -> asyncio.Task:
            task = channel._schedule_ack_reaction(SimpleNamespace(id=555, add_reaction=_hang))
            await asyncio.sleep(0.05)  # let the task start and suspend on the event
            return task

        schedule_future = asyncio.run_coroutine_threadsafe(_schedule_on_bg_loop(), bg_loop)
        task = await asyncio.wrap_future(schedule_future)
        assert task in channel._ack_reaction_tasks

        await channel.stop()

        assert task.cancelled()
        assert not channel._ack_reaction_tasks
    finally:
        _stop_bg_loop(bg_loop, bg_thread)


# ---------------------------------------------------------------------------
# Dead-client fail-fast and is_running thread-aliveness
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_send_fails_fast_when_discord_loop_is_not_running() -> None:
    """A stopped (not closed) Discord loop must fail the send, not hang the worker.

    ``_run_client`` leaves the loop stopped-but-unclosed when the client dies,
    which is exactly the state where ``call_soon_threadsafe`` queues callbacks
    that never run — the permanent-hang case this guards against.
    """
    channel = DiscordChannel(bus=MessageBus(), config={"bot_token": "token"})
    channel._discord_loop = asyncio.new_event_loop()  # created, never run
    channel._running = True
    msg = OutboundMessage(channel_name="discord", chat_id="c1", thread_id="t1", text="hello")

    try:
        with pytest.raises(RuntimeError, match="event loop is not running"):
            await channel.send(msg)
    finally:
        channel._discord_loop.close()


@pytest.mark.asyncio
async def test_outbound_loop_call_times_out_when_never_completes() -> None:
    """Even on a live loop, an outbound call that never resolves is bounded by the timeout."""
    bg_loop, bg_thread = _start_bg_loop()
    try:
        channel = DiscordChannel(bus=MessageBus(), config={"bot_token": "token"})
        channel._discord_loop = bg_loop

        async def _never_completes() -> None:
            await asyncio.sleep(3600)

        with pytest.raises(TimeoutError):
            await channel._run_on_discord_loop(_never_completes(), timeout=0.1)

        # The cancelled call leaves its task parked on the bg loop (concurrent
        # cancellation cannot reach a running run_coroutine_threadsafe task);
        # clean it up so stopping the loop has nothing pending.
        async def _cancel_leftovers() -> None:
            current = asyncio.current_task()
            for task in asyncio.all_tasks():
                if task is not current:
                    task.cancel()

        cleanup = asyncio.run_coroutine_threadsafe(_cancel_leftovers(), bg_loop)
        cleanup.result(timeout=5)
    finally:
        _stop_bg_loop(bg_loop, bg_thread)


def test_is_running_tracks_thread_aliveness() -> None:
    """``is_running`` reflects the client thread, so readiness can restart a dead channel."""
    channel = DiscordChannel(bus=MessageBus(), config={"bot_token": "token"})
    assert channel.is_running is False  # never started

    channel._running = True
    assert channel.is_running is False  # started but the thread object is gone

    finished = threading.Thread(target=lambda: None)
    finished.start()
    finished.join()
    channel._thread = finished
    assert channel.is_running is False  # dead client thread (fatal exit)

    channel._thread = threading.current_thread()
    assert channel.is_running is True  # live thread -> healthy


def _close_unawaited_mock_coroutines(run_mock) -> None:
    """Close the coroutines handed to an AsyncMock that never awaited them.

    ``_run_on_discord_loop`` receives already-created coroutine objects; an
    AsyncMock stand-in records them without awaiting, so they must be closed
    explicitly or GC warns about never-awaited coroutines.
    """
    for call in run_mock.await_args_list:
        call.args[0].close()


@pytest.mark.asyncio
async def test_send_file_upload_call_uses_the_dedicated_upload_timeout(tmp_path) -> None:
    """Pin the upload call site to DISCORD_UPLOAD_TIMEOUT_SECONDS.

    Regressing ``send_file``'s upload call to the 30 s control-plane default
    (the exact bug round 1 of this review caught) keeps every helper-level
    test green; only the call site's ``timeout=`` kwarg can guard it.
    """
    from app.channels.discord import DISCORD_UPLOAD_TIMEOUT_SECONDS

    channel = DiscordChannel(bus=MessageBus(), config={"bot_token": "token"})
    channel._discord_module = SimpleNamespace(File=lambda fp, filename=None: fp)
    run_mock = AsyncMock(return_value=None)
    channel._run_on_discord_loop = run_mock  # type: ignore[method-assign]
    channel._resolve_target = _resolve_to(SimpleNamespace(send=_noop_coro))

    path = tmp_path / "upload.txt"
    path.write_bytes(b"hello")
    att = ResolvedAttachment("/mnt/user-data/outputs/upload.txt", path, "upload.txt", "text/plain", 5, False)
    msg = OutboundMessage(channel_name="discord", chat_id="c1", thread_id="t1", text="t")

    try:
        assert await channel.send_file(msg, att) is True
    finally:
        _close_unawaited_mock_coroutines(run_mock)

    assert len(run_mock.await_args_list) == 2  # stop_typing, then the upload
    stop_call, upload_call = run_mock.await_args_list
    assert "timeout" not in stop_call.kwargs  # control-plane call keeps the 30 s default
    assert upload_call.kwargs.get("timeout") == DISCORD_UPLOAD_TIMEOUT_SECONDS


@pytest.mark.asyncio
async def test_send_control_calls_keep_the_default_outbound_bound() -> None:
    """``send``'s typing-stop and message sends rely on the 30 s default, not an override."""
    channel = DiscordChannel(bus=MessageBus(), config={"bot_token": "token"})
    run_mock = AsyncMock(return_value=None)
    channel._run_on_discord_loop = run_mock  # type: ignore[method-assign]
    channel._resolve_target = _resolve_to(SimpleNamespace(send=_noop_coro))

    msg = OutboundMessage(channel_name="discord", chat_id="c1", thread_id="t1", text="hello")

    try:
        await channel.send(msg)
    finally:
        _close_unawaited_mock_coroutines(run_mock)

    assert len(run_mock.await_args_list) == 2  # stop_typing + one text chunk
    for call in run_mock.await_args_list:
        assert "timeout" not in call.kwargs