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 logging
import sys
import threading
import weakref
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock, 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"


@pytest.mark.parametrize("collection", [list, tuple, set, frozenset])
def test_allowed_channels_accepts_all_id_collections(collection):
    channel = DiscordChannel(bus=MessageBus(), config={"allowed_channels": collection([" 456 ", "789", ""])})

    assert channel._allowed_channels == frozenset({"456", "789"})


@pytest.mark.parametrize("value", [None, "", " \t ", [], (), set(), frozenset()])
def test_empty_allowed_channels_gives_no_mention_exemption(value):
    channel = DiscordChannel(bus=MessageBus(), config={"allowed_channels": value})

    assert channel._allowed_channels == frozenset()


@pytest.mark.asyncio
async def test_start_without_discord_module_points_at_the_extra(caplog) -> None:
    """discord.py ships in the optional ``discord`` extra, so the missing-dependency
    hint must send the operator to ``uv sync --extra discord`` — the same command
    ``scripts/detect_uv_extras.py`` and the Docker builds already use — rather than
    ``uv add``, which would rewrite pyproject.toml and diverge from ``uv sync --locked``."""
    channel = DiscordChannel(bus=MessageBus(), config={"bot_token": "token"})

    with caplog.at_level("ERROR", logger="app.channels.discord"), patch.dict(sys.modules, {"discord": None}):
        await channel.start()

    assert any("uv sync --extra discord" in record.message for record in caplog.records)
    assert not any("uv add" in record.message for record in caplog.records)
    assert channel._running is False


# ---------------------------------------------------------------------------
# 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"}


@pytest.mark.asyncio
async def test_discord_stop_joins_client_thread_off_the_event_loop() -> None:
    """stop() must not join the client thread on the shared Gateway loop.

    The thread usually exits right after the cross-loop client close, but a
    timed-out close or a slow ``_run_client()`` drain keeps it alive for up to
    the 10s join timeout, and the Gateway loop must keep running meanwhile.
    """
    channel = DiscordChannel(bus=MessageBus(), config={"bot_token": "token"})
    release = threading.Event()
    # Bounded so a join run on the event loop fails the assertion below
    # instead of hanging the test.
    client_thread = threading.Thread(target=release.wait, args=(2,), daemon=True)
    client_thread.start()
    channel._discord_loop = None
    channel._client = None
    channel._thread = client_thread
    channel._cancel_ephemeral_tasks = AsyncMock()

    stop_task = asyncio.create_task(channel.stop())
    await asyncio.sleep(0.05)
    assert not stop_task.done()

    release.set()
    await stop_task
    assert channel._thread is None


class _TypingTarget:
    """Messageable stand-in shaped like discord.py 2.x.

    discord.py 2.0 removed ``Messageable.trigger_typing()``: ``typing()``
    returns a ``Typing`` object, and awaiting it sends one indicator. This
    fake deliberately has no ``trigger_typing``, so code still calling the
    1.x API fails here the way it fails against the real library.
    """

    def __init__(self, error: Exception | None = None) -> None:
        self.sent = asyncio.Event()
        self.calls = 0
        self._error = error

    def typing(self):
        return self._send()

    async def _send(self) -> None:
        self.calls += 1
        self.sent.set()
        if self._error is not None:
            raise self._error


@pytest.mark.asyncio
async def test_start_typing_sends_indicator_through_discord_typing_api() -> None:
    channel = DiscordChannel(bus=MessageBus(), config={"bot_token": "token"})
    channel._running = True
    target = _TypingTarget()

    await channel._start_typing(target, "chat-1")
    try:
        await asyncio.wait_for(target.sent.wait(), timeout=1)
    finally:
        await channel._stop_typing("chat-1")

    assert target.calls == 1


async def _wait_for_typing_calls(target: _TypingTarget, count: int, sleep) -> None:
    for _ in range(1000):
        if target.calls >= count:
            return
        await sleep(0)
    raise AssertionError(f"typing loop made {target.calls} calls, expected {count}")


@pytest.mark.asyncio
async def test_start_typing_warns_on_first_failure_then_logs_at_debug(caplog, monkeypatch) -> None:
    """A loop's first failed indicator logs at WARNING so a permanently dead
    indicator is visible at the default level; later ticks of the same loop
    log at DEBUG, and the loop keeps retrying. A new loop warns again."""
    import app.channels.discord as discord_module

    real_sleep = asyncio.sleep

    async def fast_sleep(_delay):
        await real_sleep(0)

    # Collapse the 10s tick so several ticks run inside the test.
    monkeypatch.setattr(discord_module.asyncio, "sleep", fast_sleep)
    channel = DiscordChannel(bus=MessageBus(), config={"bot_token": "token"})
    channel._running = True
    target = _TypingTarget(error=RuntimeError("missing permissions"))

    def failure_records():
        return [record for record in caplog.records if "failed to send typing indicator" in record.getMessage()]

    with caplog.at_level("DEBUG", logger="app.channels.discord"):
        await channel._start_typing(target, "chat-1")
        typing_task = channel._typing_tasks["chat-1"]
        try:
            await _wait_for_typing_calls(target, 3, real_sleep)
            assert not typing_task.done()
        finally:
            await channel._stop_typing("chat-1")

        first_loop = failure_records()
        assert [record.levelno for record in first_loop[:3]] == [logging.WARNING, logging.DEBUG, logging.DEBUG]
        assert [record.levelno for record in first_loop].count(logging.WARNING) == 1
        assert first_loop[0].exc_info is not None

        caplog.clear()
        target.calls = 0
        await channel._start_typing(target, "chat-1")
        try:
            await _wait_for_typing_calls(target, 1, real_sleep)
        finally:
            await channel._stop_typing("chat-1")

    assert failure_records()[0].levelno == logging.WARNING


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"))
    # The commit runs as a task on the Gateway loop (identity lookup included).
    inbound = await asyncio.wait_for(bus.get_inbound(), timeout=5)
    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"))
    # The commit runs as a task on the Gateway loop (identity lookup included).
    inbound = await asyncio.wait_for(bus.get_inbound(), timeout=5)
    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


# ---------------------------------------------------------------------------
# allowed_guilds / allowed_channels parsing
# ---------------------------------------------------------------------------


class TestDiscordAllowedGuilds:
    """A restriction the operator configured must never silently open the bot."""

    @staticmethod
    def _channel(config_extra: dict) -> DiscordChannel:
        return DiscordChannel(bus=MessageBus(), config={"bot_token": "token", **config_extra})

    @pytest.mark.parametrize("config_extra", [{}, {"allowed_guilds": None}, {"allowed_guilds": []}, {"allowed_guilds": " "}])
    def test_unset_or_empty_allowlist_allows_every_guild_without_warning(self, config_extra, caplog):
        with caplog.at_level(logging.WARNING, logger="app.channels.discord"):
            channel = self._channel(config_extra)

        assert channel._check_guild(SimpleNamespace(id=321))
        assert channel._check_guild(None)
        assert caplog.records == []

    def test_listed_guild_ids_are_allowed_and_others_denied(self):
        channel = self._channel({"allowed_guilds": [321, "654"]})

        assert channel._check_guild(SimpleNamespace(id=321))
        assert channel._check_guild(SimpleNamespace(id=654))
        assert not channel._check_guild(SimpleNamespace(id=1))
        assert not channel._check_guild(None)

    @pytest.mark.parametrize("allowed_guilds", ["321", 321], ids=["str", "int"])
    def test_scalar_guild_id_is_one_entry_not_its_digits(self, allowed_guilds):
        # config.example.yaml sells this shorthand ("can also be a single guild
        # ID"). Iterating the string "321" admitted guilds 1, 2 and 3 while
        # blocking 321; an unquoted YAML int raised TypeError in the constructor.
        channel = self._channel({"allowed_guilds": allowed_guilds})

        assert channel._check_guild(SimpleNamespace(id=321))
        assert not channel._check_guild(SimpleNamespace(id=3))

    @pytest.mark.parametrize("bad_entry", ["@acme", "321,654", 0, -5, None, 4.2, True])
    def test_unparseable_entry_is_dropped_with_warning(self, bad_entry, caplog):
        with caplog.at_level(logging.WARNING, logger="app.channels.discord"):
            channel = self._channel({"allowed_guilds": [321, bad_entry]})

        assert channel._check_guild(SimpleNamespace(id=321))
        assert not channel._check_guild(SimpleNamespace(id=1))
        assert repr(bad_entry) in caplog.text
        # 0, -5 and True are numeric, so the hint has to say what they lack.
        assert "positive numeric guild ID" in caplog.text

    @pytest.mark.parametrize(
        "allowed_guilds",
        # "321,654" is what a $ENV reference to a comma-separated value resolves
        # to; the dict reaches the constructor from the runtime channel config.
        [["@acme", "server"], "@acme", "321,654", [True], [4.2], {"id": 321}],
    )
    def test_allowlist_without_a_parseable_entry_denies_every_guild(self, allowed_guilds, caplog):
        with caplog.at_level(logging.WARNING, logger="app.channels.discord"):
            channel = self._channel({"allowed_guilds": allowed_guilds})

        assert not channel._check_guild(SimpleNamespace(id=321))
        assert not channel._check_guild(SimpleNamespace(id=3))
        assert any(record.levelno == logging.ERROR for record in caplog.records)

    @pytest.mark.asyncio
    @pytest.mark.parametrize(("allowed_guilds", "admitted"), [("321", True), ("322", False)])
    async def test_on_message_applies_the_allowlist(self, allowed_guilds, admitted):
        # _make_discord_message reports guild ID 321 — the documented scalar.
        channel = self._channel({"allowed_guilds": allowed_guilds})
        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()
        # A refused capacity slot ends the hand-off here, so nothing is stubbed
        # beyond it: the assertion is purely "did the gate let the message in".
        channel._reserve_inbound = MagicMock(return_value=None)

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

        assert channel._reserve_inbound.called is admitted


class TestDiscordAllowedChannels:
    """``allowed_channels`` is parsed by the same lines, so it breaks the same way."""

    @staticmethod
    def _channel(config_extra: dict) -> DiscordChannel:
        return DiscordChannel(bus=MessageBus(), config={"bot_token": "token", "mention_only": True, "thread_mode": False, **config_extra})

    @pytest.mark.parametrize("allowed_channels", ["456", 456], ids=["str", "int"])
    def test_scalar_channel_id_is_one_entry_not_its_digits(self, allowed_channels):
        channel = self._channel({"allowed_channels": allowed_channels})

        assert "456" in channel._allowed_channels
        assert "4" not in channel._allowed_channels

    @pytest.mark.asyncio
    @pytest.mark.parametrize(("allowed_channels", "admitted"), [("456", True), ("457", False)])
    async def test_exempt_channel_needs_no_mention(self, allowed_channels, admitted):
        channel = self._channel({"allowed_channels": allowed_channels})
        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._reserve_inbound = MagicMock(return_value=MagicMock())
        channel._publish_reserved = MagicMock(return_value=None)

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

        assert channel._publish_reserved.called is admitted


class TestDiscordDedupeClaimBeforeThreadCreation:
    """Every instance's Discord session receives every MESSAGE_CREATE. The manager
    drops the duplicate, but only after both instances created a Discord thread
    for it. The adapter therefore claims the manager's dedupe key first."""

    @staticmethod
    def _wire(tmp_path, *, dedupe=None):
        from app.channels.dedupe_store import MemoryInboundDedupeStore
        from app.channels.manager import ChannelManager
        from app.channels.store import JsonChannelStore

        bus = MessageBus()
        dedupe = dedupe or MemoryInboundDedupeStore()
        ChannelManager(bus=bus, store=JsonChannelStore(tmp_path / "store.json"), inbound_dedupe_store=dedupe)
        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(return_value=SimpleNamespace(id=777))
        dedupe.try_record = AsyncMock(wraps=dedupe.try_record)
        channel._start_typing = AsyncMock(return_value=None)
        channel._add_reaction = AsyncMock(return_value=None)
        return bus, channel, dedupe

    @staticmethod
    def _key():
        from app.channels.manager import ChannelManager

        return ChannelManager._inbound_dedupe_key(InboundMessage(channel_name="discord", chat_id="456", user_id="123", text="", metadata={"guild_id": "321", "channel_id": "456", "message_id": "111"}))

    @pytest.mark.asyncio
    async def test_message_claimed_by_a_peer_creates_no_thread(self, tmp_path) -> None:
        bus, channel, dedupe = self._wire(tmp_path)
        assert await dedupe.try_record(self._key()) is False  # the peer instance won

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

        channel._create_thread.assert_not_awaited()
        assert bus.inbound_queue.qsize() == 0
        assert bus._inbound_reservations == set()

    @pytest.mark.asyncio
    async def test_winning_claim_creates_the_thread_and_marks_the_published_message(self, tmp_path) -> None:
        bus, channel, dedupe = self._wire(tmp_path)

        await channel._on_message(_make_discord_message("hello"))
        inbound = await asyncio.wait_for(bus.get_inbound(), timeout=5)
        bus.inbound_task_done()

        channel._create_thread.assert_awaited_once()
        assert (inbound.chat_id, inbound.topic_id, inbound.dedupe_claimed) == ("456", "777", True)
        assert self._key() in dedupe._store

        # The same MESSAGE_CREATE on a second session sharing the dedupe store.
        _, peer, _ = self._wire(tmp_path, dedupe=dedupe)
        await peer._on_message(_make_discord_message("hello"))
        peer._create_thread.assert_not_awaited()

    @pytest.mark.asyncio
    async def test_refused_hand_off_gives_the_claim_back(self, tmp_path) -> None:
        _, channel, dedupe = self._wire(tmp_path)
        channel._publish_reserved = MagicMock(return_value=None)

        await channel._on_message(_make_discord_message("hello"))
        for _ in range(20):
            if self._key() not in dedupe._store:
                break
            await asyncio.sleep(0.01)

        channel._create_thread.assert_awaited_once()
        dedupe.try_record.assert_awaited_once_with(self._key())
        assert self._key() not in dedupe._store

    @pytest.mark.asyncio
    async def test_failed_thread_creation_gives_the_claim_back(self, tmp_path) -> None:
        _, channel, dedupe = self._wire(tmp_path)
        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"))
        for _ in range(20):
            if self._key() not in dedupe._store:
                break
            await asyncio.sleep(0.01)

        dedupe.try_record.assert_awaited_once_with(self._key())
        assert self._key() not in dedupe._store

    @pytest.mark.asyncio
    async def test_failed_commit_gives_the_claim_back(self, tmp_path) -> None:
        bus, channel, dedupe = self._wire(tmp_path)
        # The intake closed (shutdown) between the reservation and the commit.
        channel._commit_reserved_inbound = MagicMock(return_value=False)

        await channel._on_message(_make_discord_message("hello"))
        for _ in range(20):
            if self._key() not in dedupe._store:
                break
            await asyncio.sleep(0.01)

        dedupe.try_record.assert_awaited_once_with(self._key())
        assert self._key() not in dedupe._store

    @pytest.mark.asyncio
    async def test_stop_during_identity_lookup_gives_the_claim_back(self, tmp_path) -> None:
        """stop() cancels the hand-off while it resolves the connection identity:
        the message was never committed, so a peer must be able to take it."""
        _, channel, dedupe = self._wire(tmp_path)
        lookup_started = asyncio.Event()

        async def lookup_forever(inbound, guild_id=None):
            lookup_started.set()
            await asyncio.Event().wait()

        channel._attach_connection_identity = lookup_forever
        handling = asyncio.create_task(channel._on_message(_make_discord_message("hello")))
        await asyncio.wait_for(lookup_started.wait(), timeout=5)
        assert self._key() in dedupe._store

        await channel.stop()
        await asyncio.wait_for(handling, timeout=5)

        assert self._key() not in dedupe._store
        peer_bus, peer, _ = self._wire(tmp_path, dedupe=dedupe)
        await peer._on_message(_make_discord_message("hello"))
        await asyncio.wait_for(peer_bus.get_inbound(), timeout=5)

    @pytest.mark.asyncio
    async def test_claim_whose_release_is_refused_during_stop_is_given_back_by_stop(self, tmp_path) -> None:
        """Thread creation still in flight when stop() closes cross-loop intake:
        the release cannot be submitted then, so stop() releases it last."""
        _, channel, dedupe = self._wire(tmp_path)
        create_started = asyncio.Event()
        fail_create = asyncio.Event()

        async def create_then_fail(message):
            create_started.set()
            await fail_create.wait()
            raise RuntimeError("thread create failed")

        channel._create_thread = create_then_fail
        handling = asyncio.create_task(channel._on_message(_make_discord_message("hello")))
        await asyncio.wait_for(create_started.wait(), timeout=5)
        await channel._close_and_drain_threadsafe_futures()  # stop() has begun
        fail_create.set()
        with pytest.raises(RuntimeError, match="thread create failed"):
            await handling
        assert self._key() in dedupe._store

        await channel.stop()

        assert self._key() not in dedupe._store