hermes-chatto-plugin/test_adapter.py

"""Unit tests for the Chatto platform adapter.

Covers:
  - Emoji shortcode conversion
  - Adapter instantiation and properties
  - Registration and requirements
  - Basic functionality with chattolib
  - Message sending and reactions
  - User lookup (with caching)
  - Presence and custom status
  - DM room management (/join, /leave)

All network calls are mocked — no real HTTP or WebSocket connections.
"""

import asyncio
import os
import sys
from datetime import UTC, datetime, timedelta
from unittest.mock import AsyncMock, MagicMock, patch

import pytest
import pytest_asyncio

# Import chattolib types for tests - using vendored chattolib from adapter

# -- Path setup --
# The Hermes agent itself is not a dependency of this plugin; point HERMES_ROOT
# at a checkout to run these tests outside a deployed agent.
PLUGIN_ROOT = os.path.abspath(os.path.dirname(__file__))
sys.path.insert(0, PLUGIN_ROOT)
sys.path.insert(0, os.environ.get("HERMES_ROOT", "/opt/hermes"))

# Importing ``adapter`` puts the vendored dependencies on sys.path as a side
# effect, but import sorting may legally move that import after the chattolib
# and gateway ones — so bootstrap the vendor paths explicitly instead.
from vendor_path import setup_vendor_path

setup_vendor_path()

from chattolib.client import ChattoClient
from chattolib.exceptions import ChattoError
from chattolib.types import (
    Asset,
    AssetUpload,
    AssetUrl,
    DirectoryMember,
    Message,
    MessageAttachment,
    Page,
    PresenceStatus,
    Room,
    RoomKind,
    RoomViewerState,
    RoomWithViewerState,
    User,
)
from gateway.config import PlatformConfig
from gateway.platforms.base import (
    BasePlatformAdapter,
    CachedMedia,
    MessageType,
    ProcessingOutcome,
    SendResult,
    get_inbound_media_max_bytes,
)

from adapter import (
    ChattoAdapter,
    HermesChatType,
    RoomPolicy,
    _capabilities,
    chat_type_for_room_kind,
    hermes_env_enablement_fn,
    hermes_standalone_sender_fn,
    register,
)
from adapter import (
    hermes_check_fn as check_requirements,
)
from adapter import (
    hermes_validate_config as validate_config,
)
from platform_config import ChattoConstants

_EMOJI_TO_SHORTCODE = ChattoConstants.EMOJI_TO_SHORTCODE
_MAX_MESSAGE_LENGTH = ChattoConstants.MAX_MESSAGE_LENGTH
_SEEN_CAP = ChattoConstants.SEEN_CAP


# -- Helpers --


class _MockPluginContext:
    """Minimal mock for the plugin registration context."""

    def __init__(self):
        self.registered_names = []
        self.registered_kwargs = None

    def register_platform(self, **kwargs):
        from gateway.platform_registry import PlatformEntry, platform_registry

        entry = PlatformEntry(
            name=kwargs["name"],
            label=kwargs.get("label", kwargs["name"]),
            adapter_factory=kwargs.get("adapter_factory"),
            check_fn=kwargs.get("check_fn"),
            validate_config=kwargs.get("validate_config"),
            is_connected=kwargs.get("is_connected"),
            required_env=kwargs.get("required_env", []),
            source="plugin",
        )
        platform_registry.register(entry)
        self.registered_names.append(kwargs["name"])
        self.registered_kwargs = kwargs


def _ensure_chatto_registered():
    """Register the platform so Platform(PLATFORM_NAME) resolves."""
    from gateway.platform_registry import platform_registry

    if not platform_registry.is_registered(ChattoConstants.PLATFORM_NAME):
        ctx = _MockPluginContext()
        register(ctx)


_CHATTO_ENV_KEYS = [
    "CHATTO_BASE_URL",
    "CHATTO_TOKEN",
    "CHATTO_HOME_CHANNEL",
    "CHATTO_REQUIRE_MENTION_ROOMS",
    "CHATTO_OPTIONAL_MENTION_ROOMS",
    "CHATTO_ALLOWED_USERS",
    "CHATTO_ALLOW_ALL_USERS",
    "CHATTO_AUTO_THREAD",
    "CHATTO_REACTIONS",
]


def _clear_chatto_env(monkeypatch=None):
    """Remove all CHATTO_* env vars so tests start from a clean slate."""
    for key in _CHATTO_ENV_KEYS:
        if monkeypatch is not None:
            monkeypatch.delenv(key, raising=False)
        else:
            os.environ.pop(key, None)


def _make_config(**extra_overrides):
    """Create a minimal PlatformConfig for testing."""
    _ensure_chatto_registered()
    extra = {"base_url": "https://chat.example.com"}
    extra.update(extra_overrides)
    return PlatformConfig(enabled=True, extra=extra)


def _make_room(room_id, name, kind):
    """Build a real chattolib Room, as the client would return."""
    return Room(
        id=room_id,
        name=name,
        kind=kind,
        description="",
        archived=False,
        group_id="",
        universal=kind != RoomKind.DM,
    )


def _make_user(user_id, login):
    """Build a real chattolib User, as the member directory would return."""
    return User(id=user_id, login=login, display_name=login.replace("_", " ").title())


def _make_admin_member(user):
    """Dict shape that admin_get_member() returns (pb_to_dict(GetMemberResponse))."""
    return {"member": {"user": {"id": user.id, "login": user.login, "display_name": user.display_name}}}


def _make_attachment(filename, content_type, url="https://cdn.example.com/a"):
    """Build a real MessageAttachment carrying a (pre-signed) asset URL."""
    return MessageAttachment(
        id="asset-" + filename,
        filename=filename,
        content_type=content_type,
        asset_url=AssetUrl(url=url),
    )


def _make_message(
    body="hi",
    attachments=None,
    message_id="msg-1",
    room_id="room-1",
    created_at=None,
    updated_at=None,
):
    """Build a real chattolib Message, as fetch_message() would return."""
    return Message(
        id=message_id,
        room_id=room_id,
        created_at=created_at,
        actor_id="user-1",
        body=body,
        attachments=list(attachments or []),
        updated_at=updated_at,
    )


def _make_posted_payload(room_id="room-1", message_event_id="msg-1"):
    """A message_posted payload whose get_message() the caller stubs."""
    payload = MagicMock()
    payload.room_id = room_id
    payload.message_event_id = message_event_id
    payload.thread_root_event_id = None
    return payload


def _with_get_message(adapter, payload, return_value):
    """Mock client.get_message to return the given Message for a payload.

    Replaces the old payload.fetch_message() pattern — chattolib 0.5 moved
    message-fetch onto ChattoClient.get_message(room_id=..., event_id=...).
    """
    adapter._chatto_client.get_message = AsyncMock(return_value=return_value)
    return payload


def _cached(path, media_type, kind):
    """The CachedMedia that cache_media_bytes() would return for an attachment."""
    return CachedMedia(path=path, media_type=media_type, kind=kind, display_name="f")


def _make_adapter(**extra_overrides):
    """Create a ChattoAdapter with mocked config."""
    _clear_chatto_env()
    cfg = _make_config(**extra_overrides)
    adapter = ChattoAdapter(cfg)
    adapter._chatto_client = MagicMock()
    return adapter


# -- Emoji shortcode conversion --


class TestEmojiShortcode:
    """Test emoji to shortcode mapping."""

    def test_emoji_to_shortcode_exists(self):
        assert isinstance(_EMOJI_TO_SHORTCODE, dict)
        assert len(_EMOJI_TO_SHORTCODE) > 0

    def test_emoji_to_shortcode_common_emojis(self):
        assert _EMOJI_TO_SHORTCODE.get("👍") == "thumbsup"
        assert _EMOJI_TO_SHORTCODE.get("👎") == "thumbsdown"
        assert _EMOJI_TO_SHORTCODE.get("❤️") == "heart"
        assert _EMOJI_TO_SHORTCODE.get("❤") == "heart"
        assert _EMOJI_TO_SHORTCODE.get("✅") == "white_check_mark"
        assert _EMOJI_TO_SHORTCODE.get("❌") == "x"
        # Sent when a message addresses someone else — without the mapping the
        # raw emoji would go out as a shortcode and the server would reject it.
        assert _EMOJI_TO_SHORTCODE.get("🫥") == "dotted_line_face"


# -- Adapter instantiation and properties --


class TestAdapterInstantiation:
    """Test ChattoAdapter creation and basic properties."""

    def test_adapter_creation(self):
        cfg = _make_config()
        adapter = ChattoAdapter(cfg)
        assert adapter is not None
        # Platform members created dynamically from a plugin name carry the
        # name upper-cased; the registered identity is the value.
        assert adapter.platform.value == ChattoConstants.PLATFORM_NAME

    def test_adapter_max_message_length(self):
        """The framework chunks via max_message_length_for_chat(), which reads
        the adapter-scalar MAX_MESSAGE_LENGTH and silently falls back to 4096
        when it is missing."""
        cfg = _make_config()
        adapter = ChattoAdapter(cfg)
        assert adapter.MAX_MESSAGE_LENGTH == _MAX_MESSAGE_LENGTH
        assert adapter.max_message_length_for_chat("room-1") == _MAX_MESSAGE_LENGTH

    def test_adapter_splits_long_messages(self):
        cfg = _make_config()
        adapter = ChattoAdapter(cfg)
        assert adapter.splits_long_messages is True

    def test_adapter_threads_enabled_by_default(self):
        """There is no capability flag for threads — Chatto threading is driven
        by the auto_thread setting, which defaults to on."""
        cfg = _make_config()
        adapter = ChattoAdapter(cfg)
        assert adapter.chatto_config.auto_thread.value is True


# -- Registration and requirements --


class TestRegistration:
    """Test plugin registration."""

    def test_register_called(self):
        ctx = _MockPluginContext()
        register(ctx)
        assert ChattoConstants.PLATFORM_NAME in ctx.registered_names
        assert ctx.registered_kwargs["name"] == ChattoConstants.PLATFORM_NAME
        assert ctx.registered_kwargs["label"] == ChattoConstants.PLATFORM_LABEL
        assert ctx.registered_kwargs["max_message_length"] == _MAX_MESSAGE_LENGTH

    def test_platform_hint_advertises_media_sending(self):
        """This hint is the only thing telling the model the channel can carry
        files — without it the agent has no idea it can deliver an image."""
        ctx = _MockPluginContext()
        register(ctx)
        assert "MEDIA:/absolute/path/to/file" in ctx.registered_kwargs["platform_hint"]

    def test_platform_hint_rules_out_markdown_for_local_files(self):
        """gateway's extract_images only matches https?:// — markdown pointing at
        a local file is never extracted and lands in the chat as literal text."""
        ctx = _MockPluginContext()
        register(ctx)
        hint = ctx.registered_kwargs["platform_hint"]
        assert "Do NOT use markdown image syntax for local files" in hint

    def test_startup_logs_the_capabilities(self, caplog):
        with caplog.at_level("INFO"):
            register(_MockPluginContext())
        logged = "\n".join(caplog.messages)
        assert "images" in logged
        assert "reactions" in logged

    def test_capabilities_skips_methods_we_do_not_override(self):
        """An inherited base fallback is not a capability — claiming it would
        promise the user something the adapter cannot actually do."""
        assert "video" in _capabilities()
        with patch.object(ChattoAdapter, "send_video", BasePlatformAdapter.send_video):
            assert "video" not in _capabilities()

    def test_check_requirements(self):
        assert check_requirements() is True

    def test_check_requirements_missing(self):
        with patch("builtins.__import__", side_effect=ImportError("no chattolib")):
            assert check_requirements() is False

    def test_validate_config(self):
        _clear_chatto_env()
        os.environ["CHATTO_TOKEN"] = "cht_BK_test"
        cfg = PlatformConfig(enabled=True, extra={"base_url": "https://chat.test"})
        assert validate_config(cfg) is True
        _clear_chatto_env()

    def test_validate_config_requires_token(self):
        _clear_chatto_env()
        cfg = PlatformConfig(enabled=True, extra={"base_url": "https://chat.test"})
        assert validate_config(cfg) is False
        _clear_chatto_env()


# -- Config validation: mention-list conflicts and unknown keys --


class TestValidateConfigGates:
    """hermes_validate_config rejects contradictory mention lists and warns
    about extra keys that look like misspelled config fields."""

    def _cfg(self, **extra):
        _clear_chatto_env()
        base = {
            "base_url": "https://chat.test",
            "token": "cht_BK_test",
        }
        base.update(extra)
        return PlatformConfig(enabled=True, extra=base)

    def test_room_on_both_mention_lists_is_rejected(self):
        cfg = self._cfg(
            require_mention_rooms=["room-1", "room-2"],
            optional_mention_rooms=["room-2", "room-3"],
        )
        assert validate_config(cfg) is False

    def test_disjoint_mention_lists_are_accepted(self):
        cfg = self._cfg(
            require_mention_rooms=["room-1"],
            optional_mention_rooms=["room-3"],
        )
        assert validate_config(cfg) is True

    @pytest.mark.parametrize("key", ["require_mention_rooms", "optional_mention_rooms"])
    def test_known_keys_do_not_warn(self, key, caplog):
        cfg = self._cfg(**{key: ["room-1"]})
        with caplog.at_level("WARNING"):
            validate_config(cfg)
        assert not [m for m in caplog.messages if "matches no config field" in m]

    def test_a_typo_extra_key_warns_with_a_suggestion(self, caplog):
        """A typo'd key silently resolves to its default otherwise — the
        warning is the only thing telling the user it never reached us."""
        cfg = self._cfg(require_mention_channles=["room-1"])
        with caplog.at_level("WARNING"):
            validate_config(cfg)
        warnings = [m for m in caplog.messages if "matches no config field" in m]
        assert len(warnings) == 1
        assert "require_mention_channles" in warnings[0]
        assert "did you mean 'require_mention_rooms'" in warnings[0]

    def test_a_legacy_renamed_key_suggests_its_successor(self, caplog):
        """Removed fields read as near-misses of their successors."""
        cfg = self._cfg(require_mention=True)
        with caplog.at_level("WARNING"):
            validate_config(cfg)
        warnings = [m for m in caplog.messages if "matches no config field" in m]
        assert len(warnings) == 1
        assert "did you mean 'require_mention_rooms'" in warnings[0]

    @pytest.mark.parametrize(
        "key",
        [
            "_enabled_explicit",
            "group_sessions_per_user",
            "reply_in_thread",
            "gateway_restart_notification",
        ],
    )
    def test_gateway_keys_do_not_warn(self, key, caplog):
        """The gateway places shared keys into every platform's extra — they
        resemble nothing of ours, so near-miss matching leaves them alone."""
        cfg = self._cfg(**{key: True})
        with caplog.at_level("WARNING"):
            validate_config(cfg)
        assert not [m for m in caplog.messages if "matches no config field" in m]


# -- Send functionality --


class TestSend:
    """Test message sending functionality."""

    @pytest_asyncio.fixture
    def adapter(self):
        _clear_chatto_env()
        cfg = _make_config()
        adapter = ChattoAdapter(cfg)
        adapter._chatto_client = MagicMock()
        adapter._chatto_client.post_message = AsyncMock()
        return adapter

    async def test_send_calls_post_message(self, adapter):
        mock_msg = MagicMock()
        mock_msg.id = "msg-123"
        adapter._chatto_client.post_message.return_value = mock_msg
        result = await adapter.send("room-1", "Hello world")
        assert result.success is True
        assert result.message_id == "msg-123"
        adapter._chatto_client.post_message.assert_called_once()

    async def test_send_with_thread(self, adapter):
        mock_msg = MagicMock()
        mock_msg.id = "msg-456"
        adapter._chatto_client.post_message.return_value = mock_msg
        result = await adapter.send("room-1", "Hello", reply_to="thread-123")
        assert result.success is True
        call_kwargs = adapter._chatto_client.post_message.call_args.kwargs
        assert call_kwargs["thread_root_event_id"] == "thread-123"

    async def test_send_raw_response_stays_dict_shaped(self, adapter):
        # The cron scheduler calls .get() on SendResult.raw_response when a job
        # targets a thread; a chattolib Message there crashes delivery
        # bookkeeping after the send and duplicates the message standalone.
        adapter._chatto_client.post_message.return_value = _make_message(
            body="Hello world", message_id="msg-789"
        )
        result = await adapter.send("room-1", "Hello world")
        assert result.success is True
        assert not result.raw_response or isinstance(result.raw_response, dict)

    async def test_send_server_error_is_retryable(self, adapter):
        adapter._chatto_client.post_message.side_effect = ChattoError("server says no")
        result = await adapter.send("room-1", "Hello world")
        assert result.success is False
        assert result.retryable is True

    async def test_send_unexpected_error_is_not_retryable(self, adapter):
        """A bug in our own code must not read as a transient network blip."""
        adapter._chatto_client.post_message.side_effect = RuntimeError("bug")
        result = await adapter.send("room-1", "Hello world")
        assert result.success is False
        assert result.retryable is False


# -- Reactions --


class TestReactions:
    """Test reaction functionality."""

    @pytest_asyncio.fixture
    def adapter(self):
        _clear_chatto_env()
        cfg = _make_config()
        adapter = ChattoAdapter(cfg)
        adapter._chatto_client = MagicMock()
        adapter._chatto_client.add_reaction = AsyncMock()
        adapter._chatto_client.remove_reaction = AsyncMock()
        return adapter

    async def test_send_reaction(self, adapter):
        await adapter.add_reaction("room-1", "msg-1", "👍")
        adapter._chatto_client.add_reaction.assert_called_once()

    async def test_remove_reaction(self, adapter):
        await adapter.remove_reaction("room-1", "msg-1", "👍")
        adapter._chatto_client.remove_reaction.assert_called_once()

    async def test_on_processing_start_adds_eyes_reaction(self, adapter):
        """on_processing_start should call add_reaction with 👀."""
        event = MagicMock()
        event.message_id = "msg-1"
        event.source.chat_id = "room-1"
        await adapter.on_processing_start(event)
        adapter._chatto_client.add_reaction.assert_called_once()
        call_kwargs = adapter._chatto_client.add_reaction.call_args.kwargs
        assert call_kwargs["message_event_id"] == "msg-1"
        assert call_kwargs["room_id"] == "room-1"
        assert call_kwargs["emoji"] == "eyes"

    async def test_on_processing_start_empty_message_id(self, adapter):
        """on_processing_start should skip reaction when message_id is empty."""
        event = MagicMock()
        event.message_id = None
        event.source.chat_id = "room-1"
        await adapter.on_processing_start(event)
        adapter._chatto_client.add_reaction.assert_not_called()

    async def test_on_processing_start_reactions_disabled(self, adapter):
        """on_processing_start should skip when reactions config is False."""
        adapter.chatto_config.reactions.value = False
        event = MagicMock()
        event.message_id = "msg-1"
        event.source.chat_id = "room-1"
        await adapter.on_processing_start(event)
        adapter._chatto_client.add_reaction.assert_not_called()


# -- Edit and Delete Messages --


class TestMessageEditing:
    """Test message editing and deletion."""

    @pytest_asyncio.fixture
    def adapter(self):
        _clear_chatto_env()
        cfg = _make_config()
        adapter = ChattoAdapter(cfg)
        adapter._chatto_client = MagicMock()
        adapter._chatto_client.update_message = AsyncMock()
        adapter._chatto_client.delete_message = AsyncMock(return_value=True)
        return adapter

    async def test_edit_message(self, adapter):
        result = await adapter.edit_message("room-1", "msg-1", "New content")
        assert result.success is True
        adapter._chatto_client.update_message.assert_called_once()
        call_kwargs = adapter._chatto_client.update_message.call_args.kwargs
        assert call_kwargs["room_id"] == "room-1"
        assert call_kwargs["event_id"] == "msg-1"
        assert call_kwargs["body"] == "New content"

    async def test_edit_message_marks_own_edit_seen(self, adapter):
        """The edit echoes back as message_edited — it must not look inbound."""
        mock_msg = MagicMock()
        mock_msg.id = "msg-1"
        adapter._chatto_client.update_message.return_value = mock_msg
        await adapter.edit_message("room-1", "msg-1", "New content")
        assert adapter._is_seen("msg-1") is True

    async def test_edit_message_too_long_refuses(self, adapter):
        """Overlong content must fall back to send() (which splits), not be
        silently truncated into a lossy edit."""
        result = await adapter.edit_message(
            "room-1",
            "msg-1",
            "x" * (_MAX_MESSAGE_LENGTH + 1),
        )
        assert result.success is False
        adapter._chatto_client.update_message.assert_not_called()

    async def test_edit_message_empty_content(self, adapter):
        result = await adapter.edit_message("room-1", "msg-1", "")
        assert result.success is False
        adapter._chatto_client.update_message.assert_not_called()

    async def test_edit_message_error_is_retryable(self, adapter):
        from chattolib.exceptions import ChattoError

        adapter._chatto_client.update_message.side_effect = ChattoError("boom")
        result = await adapter.edit_message("room-1", "msg-1", "New content")
        assert result.success is False
        assert result.retryable is True

    async def test_delete_message(self, adapter):
        result = await adapter.delete_message("room-1", "msg-1")
        assert result is True
        adapter._chatto_client.delete_message.assert_called_once()
        call_kwargs = adapter._chatto_client.delete_message.call_args.kwargs
        assert call_kwargs["room_id"] == "room-1"
        assert call_kwargs["event_id"] == "msg-1"

    async def test_delete_message_missing_ids(self, adapter):
        assert await adapter.delete_message("", "msg-1") is False
        assert await adapter.delete_message("room-1", "") is False
        adapter._chatto_client.delete_message.assert_not_called()

    async def test_delete_message_error_returns_false(self, adapter):
        from chattolib.exceptions import ChattoError

        adapter._chatto_client.delete_message.side_effect = ChattoError("nope")
        assert await adapter.delete_message("room-1", "msg-1") is False


# -- Outgoing text formatting --


class TestFormatMessage:
    """format_message() only fixes what renders wrong in Chatto."""

    def test_normalises_crlf(self):
        adapter = _make_adapter()
        assert adapter.format_message("a\r\nb\rc") == "a\nb\nc"

    def test_collapses_excess_blank_lines(self):
        adapter = _make_adapter()
        assert adapter.format_message("a\n\n\n\n\n\nb") == "a\n\n\nb"

    def test_leaves_markdown_untouched(self):
        adapter = _make_adapter()
        text = "**bold** `code`\n\n```py\nx = 1\n```\n- item"
        assert adapter.format_message(text) == text

    def test_empty_content(self):
        adapter = _make_adapter()
        assert adapter.format_message("") == ""


# -- Handoff threads --


class TestHandoffThread:
    """create_handoff_thread() anchors a handoff on a seed message."""

    @pytest_asyncio.fixture
    def adapter(self):
        adapter = _make_adapter()
        adapter._chatto_client.post_message = AsyncMock()
        adapter._chatto_client.follow_thread = AsyncMock()
        return adapter

    async def test_returns_seed_message_id(self, adapter):
        mock_msg = MagicMock()
        mock_msg.id = "seed-1"
        adapter._chatto_client.post_message.return_value = mock_msg
        adapter._room_kinds["room-1"] = RoomKind.CHANNEL

        result = await adapter.create_handoff_thread("room-1", "Refactor run")

        assert result == "seed-1"
        assert (
            adapter._chatto_client.post_message.call_args.kwargs["room_id"] == "room-1"
        )
        adapter._chatto_client.follow_thread.assert_called_once_with("room-1", "seed-1")
        # Our own seed must not come back in as inbound traffic.
        assert adapter._is_seen("seed-1") is True

    async def test_dm_has_no_threads(self, adapter):
        adapter._room_kinds["dm-1"] = RoomKind.DM
        assert await adapter.create_handoff_thread("dm-1", "x") is None
        adapter._chatto_client.post_message.assert_not_called()

    async def test_seed_post_failure(self, adapter):
        adapter._chatto_client.post_message.side_effect = RuntimeError("down")
        adapter._room_kinds["room-1"] = RoomKind.CHANNEL
        assert await adapter.create_handoff_thread("room-1", "x") is None


# -- Native file / video / audio delivery --


class TestUploadAsset:
    """Drives the real _upload_asset against real chattolib result types.

    The send_* tests stub _upload_asset out, so a wrong field name on the
    chattolib response was invisible to them until it hit a live server.
    """

    @pytest_asyncio.fixture
    def adapter(self, tmp_path):
        adapter = _make_adapter()
        self.path = tmp_path / "horse.jpg"
        self.path.write_bytes(b"\xff\xd8\xff" + b"x" * 100)
        upload = AssetUpload(upload_id="up-1", room_id="room-1")
        adapter._chatto_client.create_upload = AsyncMock(return_value=upload)
        adapter._chatto_client.upload_chunk = AsyncMock(return_value=upload)
        adapter._chatto_client.complete_upload = AsyncMock(
            return_value=(
                upload,
                Asset(
                    id="asset-9",
                    filename="horse.jpg",
                    content_type="image/jpeg",
                    size=103,
                ),
            )
        )
        return adapter

    async def test_returns_the_asset_id(self, adapter):
        assert await adapter._upload_asset("room-1", str(self.path)) == "asset-9"

    async def test_chunks_go_to_the_upload_id_from_create_upload(self, adapter):
        """AssetUpload calls it upload_id, not id — reading the wrong field made
        every upload fail with 'CreateUpload returned no upload ID'."""
        await adapter._upload_asset("room-1", str(self.path))
        assert (
            adapter._chatto_client.upload_chunk.await_args.kwargs["upload_id"] == "up-1"
        )

    async def test_missing_upload_id_is_reported(self, adapter):
        adapter._chatto_client.create_upload = AsyncMock(
            return_value=AssetUpload(upload_id="", room_id="room-1")
        )
        assert await adapter._upload_asset("room-1", str(self.path)) is None


class TestNativeSends:
    """send_document/_video/_voice upload instead of apologising in text."""

    @pytest_asyncio.fixture
    def adapter(self):
        adapter = _make_adapter()
        adapter._chatto_client.post_message = AsyncMock()
        adapter._upload_asset = AsyncMock(return_value="asset-1")
        adapter.validate_media_delivery_path = MagicMock(side_effect=lambda p: p)
        mock_msg = MagicMock()
        mock_msg.id = "msg-1"
        adapter._chatto_client.post_message.return_value = mock_msg
        return adapter

    @pytest.mark.parametrize(
        "method,arg_name",
        [
            ("send_document", "file_path"),
            ("send_video", "video_path"),
            ("send_voice", "audio_path"),
            ("send_image_file", "image_path"),
        ],
    )
    async def test_uploads_and_attaches(self, adapter, method, arg_name):
        result = await getattr(adapter, method)(
            "room-1",
            **{arg_name: "/tmp/thing.bin"},
            caption="here you go",
        )
        assert result.success is True
        adapter._upload_asset.assert_called_once_with("room-1", "/tmp/thing.bin")
        call_kwargs = adapter._chatto_client.post_message.call_args.kwargs
        assert call_kwargs["attachment_asset_ids"] == ["asset-1"]
        assert call_kwargs["body"] == "here you go"

    async def test_unsafe_path_falls_back_to_notice(self, adapter):
        adapter.validate_media_delivery_path = MagicMock(return_value=None)
        adapter.send = AsyncMock(return_value=SendResult(success=True))
        await adapter.send_document("room-1", "/etc/shadow")
        adapter._upload_asset.assert_not_called()
        # Never echo the host path into chat.
        sent_text = adapter.send.call_args.args[1]
        assert "/etc/shadow" not in sent_text

    async def test_upload_failure_falls_back_to_notice(self, adapter):
        adapter._upload_asset = AsyncMock(return_value=None)
        adapter.send = AsyncMock(return_value=SendResult(success=True))
        await adapter.send_video("room-1", "/tmp/clip.mp4", caption="a clip")
        sent_text = adapter.send.call_args.args[1]
        assert sent_text.startswith("a clip\n")
        assert "/tmp/clip.mp4" not in sent_text

    async def test_attachment_reply_threads_only_with_auto_thread(self, adapter):
        """auto_thread=false must keep attachment replies in the room, exactly
        like text replies — no quiet thread under the incoming message."""
        adapter.chatto_config.auto_thread.value = False
        await adapter.send_document("room-1", "/tmp/thing.bin", reply_to="incoming-1")
        call_kwargs = adapter._chatto_client.post_message.call_args.kwargs
        assert call_kwargs["thread_root_event_id"] == ""

    async def test_attachment_reply_threads_under_the_incoming_message(self, adapter):
        await adapter.send_document("room-1", "/tmp/thing.bin", reply_to="incoming-1")
        call_kwargs = adapter._chatto_client.post_message.call_args.kwargs
        assert call_kwargs["thread_root_event_id"] == "incoming-1"

    async def test_attachment_metadata_thread_id_wins_over_reply_to(self, adapter):
        await adapter.send_document(
            "room-1",
            "/tmp/thing.bin",
            reply_to="incoming-1",
            metadata={"thread_id": "root-9"},
        )
        call_kwargs = adapter._chatto_client.post_message.call_args.kwargs
        assert call_kwargs["thread_root_event_id"] == "root-9"

    async def test_attachment_replies_to_dms_never_thread(self, adapter):
        """DMs don't support threads — not even with auto_thread enabled."""
        adapter._room_kinds["dm-1"] = RoomKind.DM
        await adapter.send_document("dm-1", "/tmp/thing.bin", reply_to="incoming-1")
        call_kwargs = adapter._chatto_client.post_message.call_args.kwargs
        assert call_kwargs["thread_root_event_id"] == ""


# -- Batched image delivery --


class TestSendMultipleImages:
    """A batch of images belongs in ONE Chatto message."""

    @pytest_asyncio.fixture
    def adapter(self):
        adapter = _make_adapter()
        adapter._chatto_client.post_message = AsyncMock()
        mock_msg = MagicMock()
        mock_msg.id = "msg-1"
        adapter._chatto_client.post_message.return_value = mock_msg
        adapter._upload_asset = AsyncMock(side_effect=["asset-1", "asset-2"])
        adapter.validate_media_delivery_path = MagicMock(side_effect=lambda p: p)
        return adapter

    async def test_bundles_into_single_message(self, adapter):
        await adapter.send_multiple_images(
            "room-1",
            [("/tmp/a.png", "first"), ("/tmp/b.png", "second")],
        )
        adapter._chatto_client.post_message.assert_called_once()
        call_kwargs = adapter._chatto_client.post_message.call_args.kwargs
        assert call_kwargs["attachment_asset_ids"] == ["asset-1", "asset-2"]
        assert call_kwargs["body"] == "first\nsecond"

    async def test_single_image_uses_base_path(self, adapter):
        """One image is not a batch — leave it to the base implementation.

        Also pins the send_image_file signature: the base class calls it with
        ``image_path=`` as a keyword, so a renamed parameter degrades every
        native image send to a text notice.
        """
        adapter.send_image_file = AsyncMock(return_value=SendResult(success=True))
        await adapter.send_multiple_images("room-1", [("file:///tmp/a.png", "only")])
        adapter._upload_asset.assert_not_called()
        adapter.send_image_file.assert_called_once()
        assert adapter.send_image_file.call_args.kwargs["image_path"] == "/tmp/a.png"

    async def test_partial_upload_failure_still_sends_the_rest(self, adapter):
        adapter._upload_asset = AsyncMock(side_effect=["asset-1", None])
        await adapter.send_multiple_images(
            "room-1",
            [("/tmp/a.png", "first"), ("/tmp/b.png", "second")],
        )
        call_kwargs = adapter._chatto_client.post_message.call_args.kwargs
        assert call_kwargs["attachment_asset_ids"] == ["asset-1"]

    async def test_file_uri_is_unquoted(self, adapter):
        await adapter.send_multiple_images(
            "room-1",
            [("file:///tmp/a%20b.png", ""), ("/tmp/c.png", "")],
        )
        first_path = adapter._upload_asset.call_args_list[0].args[1]
        assert first_path == "/tmp/a b.png"


# -- Reaction event forwarding --


class TestReactionForwarding:
    """Human reactions reach the gateway's reaction hook surface."""

    @pytest_asyncio.fixture
    def adapter(self):
        adapter = _make_adapter()
        adapter.me = _make_user("bot-user-id", "hermes_bot")
        return adapter

    def _event(self, kind, actor_id="human-1"):
        event = MagicMock()
        event.id = "evt-1"
        event.kind = kind
        event.actor_id = actor_id
        event.payload = MagicMock(
            room_id="room-1",
            message_event_id="msg-1",
            emoji="thumbsup",
        )
        return event

    async def test_forwards_added_reaction(self, adapter):
        handler = AsyncMock()
        adapter.set_reaction_handler(handler)

        await adapter._handle_realtime_event(self._event("reaction_added"))

        handler.assert_called_once()
        payload = handler.call_args.args[0]
        assert payload["event_name"] == "reaction:added"
        assert payload["reaction"] == "thumbsup"
        assert payload["channel_id"] == "room-1"
        assert payload["message_ts"] == "msg-1"
        assert payload["user_id"] == "human-1"
        assert payload["item_type"] == "message"

    async def test_forwards_removed_reaction(self, adapter):
        handler = AsyncMock()
        adapter.set_reaction_handler(handler)
        await adapter._handle_realtime_event(self._event("reaction_removed"))
        assert handler.call_args.args[0]["event_name"] == "reaction:removed"

    async def test_ignores_own_lifecycle_reactions(self, adapter):
        """👀/✅/❌ are ours — forwarding them would feed the agent its own markers."""
        handler = AsyncMock()
        adapter.set_reaction_handler(handler)
        await adapter._handle_realtime_event(
            self._event("reaction_added", actor_id="bot-user-id"),
        )
        handler.assert_not_called()

    async def test_no_handler_registered_is_harmless(self, adapter):
        await adapter._handle_realtime_event(self._event("reaction_added"))

    async def test_handler_exception_does_not_propagate(self, adapter):
        adapter.set_reaction_handler(AsyncMock(side_effect=RuntimeError("hook boom")))
        await adapter._handle_realtime_event(self._event("reaction_added"))


# -- chat_type mapping --


class TestChatTypeMapping:
    """RoomKind -> the gateway's chat_type vocabulary."""

    def test_maps_known_kinds(self):
        assert chat_type_for_room_kind(RoomKind.DM) is HermesChatType.DM
        assert chat_type_for_room_kind(RoomKind.CHANNEL) is HermesChatType.CHANNEL

    def test_unknown_kind_is_group_never_dm(self):
        """'dm' drives session isolation — never guess it for an unknown kind."""
        assert chat_type_for_room_kind(RoomKind.UNSPECIFIED) is HermesChatType.GROUP
        assert chat_type_for_room_kind(None) is HermesChatType.GROUP

    def test_values_match_the_gateway_vocabulary(self):
        """session.py:161 declares exactly these strings; SessionSource.description
        and the PII-redacted context prompt branch on them."""
        assert [t.value for t in HermesChatType] == ["dm", "group", "channel", "thread"]

    def test_is_a_plain_str_at_call_sites(self):
        assert HermesChatType.CHANNEL == "channel"
        assert f"{HermesChatType.DM}" == "dm"

    async def test_get_chat_info_reports_channel(self):
        adapter = _make_adapter()
        adapter._room_names["room-1"] = "Team"
        adapter._room_kinds["room-1"] = RoomKind.CHANNEL
        info = await adapter.get_chat_info("room-1")
        assert info == {"name": "Team", "type": "channel"}

    async def test_get_chat_info_reports_dm(self):
        adapter = _make_adapter()
        adapter._room_kinds["dm-1"] = RoomKind.DM
        assert (await adapter.get_chat_info("dm-1"))["type"] == "dm"

    async def test_dispatch_stamps_the_mapped_chat_type(self):
        """The value reaching build_source decides how the agent is told where
        it is — a raw RoomKind lands in SessionSource.description's else-branch."""
        adapter = _make_adapter()
        adapter.chatto_config.allow_all_users.value = True
        adapter.chatto_config.optional_mention_rooms.value = ["room-1"]
        adapter.me = _make_user("bot-user-id", "hermes_bot")
        adapter._room_kinds["room-1"] = RoomKind.CHANNEL
        adapter._user_cache["user-1"] = _make_user("user-1", "alice")
        adapter.handle_message = AsyncMock()
        payload = _make_posted_payload()
        _with_get_message(adapter, payload, _make_message(body="hi"))

        await adapter._dispatch_message_posted(payload, payload.message_event_id)

        event = adapter.handle_message.call_args.args[0]
        assert event.source.chat_type == "channel"


# -- Presence --


class TestPresence:
    """Presence is a server-side TTL: stop re-announcing and the bot goes offline."""

    def _adapter(self):
        adapter = _make_adapter()
        adapter._chatto_client.set_presence = AsyncMock()
        return adapter

    async def test_refresh_loop_keeps_reannouncing_online(self):
        """The bug: a single announce at connect lapses and never comes back."""
        adapter = self._adapter()
        adapter._closing = False
        with patch.object(ChattoConstants, "PRESENCE_REFRESH_INTERVAL", 0.01):
            task = asyncio.create_task(adapter._presence_refresh_loop())
            for _ in range(200):
                if adapter._chatto_client.set_presence.await_count >= 3:
                    break
                await asyncio.sleep(0.01)
            adapter._closing = True
            task.cancel()
            try:
                await task
            except asyncio.CancelledError:
                pass

        assert adapter._chatto_client.set_presence.await_count >= 3
        for call in adapter._chatto_client.set_presence.await_args_list:
            assert call.kwargs["status"] == PresenceStatus.ONLINE

    async def test_refresh_survives_a_failing_call(self):
        """One bad tick must not kill the loop and strand the bot offline."""
        adapter = self._adapter()
        adapter._closing = False
        adapter._chatto_client.set_presence = AsyncMock(
            side_effect=[RuntimeError("boom"), None, None]
        )
        with patch.object(ChattoConstants, "PRESENCE_REFRESH_INTERVAL", 0.01):
            task = asyncio.create_task(adapter._presence_refresh_loop())
            for _ in range(200):
                if adapter._chatto_client.set_presence.await_count >= 3:
                    break
                await asyncio.sleep(0.01)
            adapter._closing = True
            task.cancel()
            try:
                await task
            except asyncio.CancelledError:
                pass

        assert adapter._chatto_client.set_presence.await_count >= 3

    async def test_announce_online_reports_failure(self):
        adapter = self._adapter()
        adapter._chatto_client.set_presence = AsyncMock(
            side_effect=RuntimeError("nope")
        )
        assert await adapter._announce_online() is False

    async def test_disconnect_does_not_broadcast_offline(self):
        """chattolib raises ValueError on OFFLINE — going offline means stopping."""
        adapter = self._adapter()
        adapter._chatto_client.close = AsyncMock()
        client = adapter._chatto_client  # disconnect() drops the reference
        await adapter.disconnect()
        client.set_presence.assert_not_called()


# -- Mentions of other people --


class TestForeignMention:
    """In an open channel (CHATTO_OPTIONAL_MENTION_ROOMS) the bot reads
    everything, so a message aimed at a named colleague would otherwise get an
    unsolicited answer. Acknowledge it with 🫥 and stay out of the conversation."""

    def _adapter(self, **overrides):
        adapter = _make_adapter()
        adapter.chatto_config.allow_all_users.value = True
        adapter.chatto_config.optional_mention_rooms.value = ["room-1"]
        adapter.chatto_config.reactions.value = True
        for key, value in overrides.items():
            getattr(adapter.chatto_config, key).value = value
        adapter.me = _make_user("bot-user-id", "hermes_bot")
        room_members = [
            DirectoryMember(user=adapter.me),
            DirectoryMember(user=_make_user("user-1", "alice")),
            DirectoryMember(user=_make_user("user-2", "bob")),
        ]
        for member in room_members:
            adapter._user_cache[member.user.id] = member.user
        adapter.handle_message = AsyncMock()
        adapter.add_reaction = AsyncMock(return_value=True)
        adapter._room_kinds["room-1"] = RoomKind.CHANNEL
        adapter._room_kinds["dm-1"] = RoomKind.DM
        # The directory knows bob and nobody else. Resolution is
        # case-insensitive, as Chatto's single mention namespace guarantees
        # (FDR-006) — the fake models that at its own boundary.
        adapter._chatto_client.admin_get_member = AsyncMock(
            side_effect=lambda **kw: (
                _make_admin_member(_make_user("user-2", "bob"))
                if str(kw.get("login", "")).lower() == "bob"
                else {}
            )
        )
        room_members = [
            DirectoryMember(user=_make_user("user-1", "alice")),
            DirectoryMember(user=_make_user("user-2", "bob")),
        ]
        room_ids = [m.user.id for m in room_members]
        adapter._chatto_client.list_room_members = AsyncMock(
            return_value=(room_ids, Page())
        )
        adapter._chatto_client.batch_get_room_members = AsyncMock(
            return_value=room_members
        )
        return adapter

    async def _dispatch(self, adapter, body, room_id="room-1"):
        payload = _make_posted_payload(room_id=room_id)
        _with_get_message(adapter, payload, _make_message(body=body, room_id=room_id))
        await adapter._dispatch_message_posted(payload, payload.message_event_id)

    async def test_message_for_someone_else_is_only_acknowledged(self):
        adapter = self._adapter()
        await self._dispatch(adapter, "@bob can you take a look?")
        adapter.handle_message.assert_not_called()
        adapter.add_reaction.assert_awaited_once()
        assert adapter.add_reaction.await_args.args[2] == "🫥"

    async def test_being_mentioned_alongside_someone_else_still_answers(self):
        adapter = self._adapter()
        await self._dispatch(adapter, "@bob and @hermes_bot, thoughts?")
        adapter.handle_message.assert_called_once()
        adapter.add_reaction.assert_not_awaited()

    async def test_several_people_addressed_and_none_of_them_us(self):
        adapter = self._adapter()
        adapter._chatto_client.admin_get_member = AsyncMock(
            side_effect=lambda **kw: (
                _make_admin_member(_make_user("u", kw["login"]))
                if kw.get("login") in {"bob", "carol"}
                else {}
            )
        )
        await self._dispatch(adapter, "@bob @carol schaut mal drüber")
        adapter.handle_message.assert_not_called()
        adapter.add_reaction.assert_awaited_once()

    async def test_a_real_handle_after_an_unknown_one_still_counts(self):
        """The scan must not stop at the first token it cannot resolve."""
        adapter = self._adapter()
        await self._dispatch(adapter, "@nonexistent @bob schaut mal drüber")
        adapter.handle_message.assert_not_called()
        adapter.add_reaction.assert_awaited_once()

    async def test_being_named_among_several_others_still_answers(self):
        adapter = self._adapter()
        await self._dispatch(adapter, "@bob @hermes_bot @carol — was meint ihr?")
        adapter.handle_message.assert_called_once()
        adapter.add_reaction.assert_not_awaited()

    async def test_broadcast_alongside_a_named_colleague_still_answers(self):
        """@here keeps the bot in the audience; naming bob as well does not
        remove it."""
        adapter = self._adapter()
        await self._dispatch(adapter, "@here @bob schaut mal drüber")
        adapter.handle_message.assert_called_once()
        adapter.add_reaction.assert_not_awaited()

    async def test_broadcast_mentions_address_the_bot_too(self):
        adapter = self._adapter()
        # @here/@all are Chatto's virtual handles (FDR-006) — the bot is in
        # their audience. @channel/@everyone do not exist as handles; they are
        # unknown candidates, and an unknown handle is not someone else either,
        # so the message still gets answered rather than acknowledged away.
        for body in ("@here standup in 5", "@all heads up", "@everyone hi"):
            adapter.handle_message.reset_mock()
            await self._dispatch(adapter, body)
            adapter.handle_message.assert_called_once()

    async def test_broadcast_mention_alone_is_not_someone_else(self):
        """A pure @here must not be read as a message aimed at another person."""
        adapter = self._adapter()
        await self._dispatch(adapter, "@here standup in 5")
        adapter.handle_message.assert_called_once()
        adapter.add_reaction.assert_not_awaited()

    async def test_trailing_punctuation_still_resolves_the_handle(self):
        """'frag @bob.' must resolve bob — a sentence-ending dot is not part of
        the handle, mirroring the web frontend's extraction."""
        adapter = self._adapter()
        await self._dispatch(adapter, "frag mal @bob.")
        adapter.handle_message.assert_not_called()
        adapter.add_reaction.assert_awaited_once()

    async def test_foreign_mention_matching_is_case_insensitive(self):
        adapter = self._adapter()
        await self._dispatch(adapter, "@BOB can you take a look?")
        adapter.handle_message.assert_not_called()
        adapter.add_reaction.assert_awaited_once()

    async def test_own_mention_matching_is_case_insensitive(self):
        adapter = self._adapter()
        await self._dispatch(adapter, "@HERMES_BOT ping")
        adapter.handle_message.assert_called_once()

    async def test_mentions_inside_a_fenced_block_are_ignored(self):
        """Upstream does not resolve mentions in code blocks; quoting '@bob'
        there must not silence us."""
        adapter = self._adapter()
        await self._dispatch(adapter, "so sieht ein ping aus:\n```\n@bob\n```")
        adapter.handle_message.assert_called_once()

    async def test_mentions_inside_an_inline_span_are_ignored(self):
        adapter = self._adapter()
        await self._dispatch(adapter, "schreib `@bob` mal direkt an")
        adapter.handle_message.assert_called_once()

    async def test_an_unterminated_fence_shields_the_rest(self):
        """An unclosed fence runs to the end of the text, like upstream."""
        adapter = self._adapter()
        await self._dispatch(adapter, "beispiel:\n``` bash\n@bob schaut")
        adapter.handle_message.assert_called_once()

    async def test_broadcast_handles_are_shared_across_cases_in_the_cache(self):
        """The directory cache keys on the lowercased handle: one lookup
        serves '@bob' and '@BOB' alike."""
        adapter = self._adapter()
        await self._dispatch(adapter, "@BOB ping")
        await self._dispatch(adapter, "@bob again")
        assert adapter._chatto_client.admin_get_member.await_count == 1

    async def test_talking_about_mentions_is_not_a_mention(self):
        """Verbatim from the field: the instruction to send a mention later must
        not read as a mention now. '@-mention' is not a handle anyone holds."""
        adapter = self._adapter()
        await self._dispatch(
            adapter,
            "Bitte schreibe um 8 Uhr Europe/Berlin per @-mention den "
            'Chatto-Nutzer "nickk" an und sage: Guten Morgen.',
        )
        adapter.handle_message.assert_called_once()
        adapter.add_reaction.assert_not_awaited()

    async def test_handle_nobody_holds_is_not_a_mention(self):
        """A plausible-looking @token that resolves to no user is not someone
        else — answering a false positive beats falling silent on one."""
        adapter = self._adapter()
        adapter._chatto_client.admin_get_member = AsyncMock(return_value={})
        await self._dispatch(adapter, "gilt das auch für @nonexistent_person?")
        adapter.handle_message.assert_called_once()
        adapter.add_reaction.assert_not_awaited()

    async def test_a_resolvable_handle_is_looked_up_once(self):
        adapter = self._adapter()
        await self._dispatch(adapter, "@bob ping")
        await self._dispatch(adapter, "@bob again")
        assert adapter._chatto_client.admin_get_member.await_count == 1
        assert adapter.handle_message.await_count == 0

    async def test_an_email_address_is_not_a_mention(self):
        adapter = self._adapter()
        await self._dispatch(adapter, "schreib an bob@example.com")
        adapter.handle_message.assert_called_once()
        adapter._chatto_client.admin_get_member.assert_not_awaited()

    async def test_plain_message_is_unaffected(self):
        adapter = self._adapter()
        await self._dispatch(adapter, "how do I reset the cache?")
        adapter.handle_message.assert_called_once()

    async def test_dms_are_answered_even_when_they_name_someone_else(self):
        adapter = self._adapter()
        await self._dispatch(adapter, "@bob said the build is red", room_id="dm-1")
        adapter.handle_message.assert_called_once()

    async def test_require_list_keeps_discarding_without_a_reaction(self):
        """The require gate wins: it drops the message before we get here, and
        it deliberately says nothing at all."""
        adapter = self._adapter(
            optional_mention_rooms=[],
            require_mention_rooms=["room-1"],
        )
        await self._dispatch(adapter, "@bob can you take a look?")
        adapter.handle_message.assert_not_called()
        adapter.add_reaction.assert_not_awaited()

    async def test_silence_holds_when_reactions_are_disabled(self):
        adapter = self._adapter(reactions=False)
        await self._dispatch(adapter, "@bob can you take a look?")
        adapter.handle_message.assert_not_called()
        adapter.add_reaction.assert_not_awaited()


# -- Room mention policies --


class TestRoomPolicies:
    """Every non-DM room is opt-in via the two mention lists — Chatto has no
    group rooms, and an unknown kind counts as a channel too. A DM is already
    addressed at the bot."""

    def _adapter(self):
        adapter = _make_adapter()
        adapter.chatto_config.allow_all_users.value = True
        adapter.me = _make_user("bot-user-id", "hermes_bot")
        adapter._user_cache["user-1"] = _make_user("user-1", "alice")
        adapter.handle_message = AsyncMock()
        return adapter

    async def _dispatch(self, adapter, room_id, body):
        payload = _make_posted_payload(room_id=room_id)
        _with_get_message(adapter, payload, _make_message(body=body, room_id=room_id))
        await adapter._dispatch_message_posted(payload, payload.message_event_id)

    async def test_unlisted_room_is_silent(self):
        """Opt-in by default: an unlisted room is not dispatched at all."""
        adapter = self._adapter()
        adapter._room_kinds["room-1"] = RoomKind.CHANNEL
        await self._dispatch(adapter, "room-1", "@hermes_bot hi there")
        adapter.handle_message.assert_not_called()

    async def test_unlisted_room_is_dropped_before_any_api_call(self):
        """Silent rooms must not even fetch the message."""
        adapter = self._adapter()
        adapter._room_kinds["room-1"] = RoomKind.CHANNEL
        payload = _make_posted_payload(room_id="room-1")
        adapter._chatto_client.get_message = AsyncMock()

        await adapter._dispatch_message_posted(payload, payload.message_event_id)

        adapter._chatto_client.get_message.assert_not_awaited()
        adapter.handle_message.assert_not_called()

    async def test_unknown_room_kind_follows_the_mention_lists(self):
        """A server that never sets kind shows up as UNSPECIFIED; Chatto has
        no group rooms, so it is treated as a channel: silent when unlisted,
        gated when listed."""
        adapter = self._adapter()
        adapter._room_kinds["mystery-room"] = RoomKind.UNSPECIFIED
        payload = _make_posted_payload(room_id="mystery-room")
        adapter._chatto_client.get_message = AsyncMock()

        await adapter._dispatch_message_posted(payload, payload.message_event_id)

        adapter._chatto_client.get_message.assert_not_awaited()

        adapter.chatto_config.optional_mention_rooms.value = ["mystery-room"]
        await self._dispatch(adapter, "mystery-room", "hi there")
        adapter.handle_message.assert_called_once()

    async def test_require_list_room_without_mention_is_discarded(self):
        adapter = self._adapter()
        adapter.chatto_config.require_mention_rooms.value = ["room-1"]
        adapter._room_kinds["room-1"] = RoomKind.CHANNEL
        await self._dispatch(adapter, "room-1", "hi there")
        adapter.handle_message.assert_not_called()

    async def test_require_list_room_with_mention_is_answered(self):
        adapter = self._adapter()
        adapter.chatto_config.require_mention_rooms.value = ["room-1"]
        adapter._room_kinds["room-1"] = RoomKind.CHANNEL
        await self._dispatch(adapter, "room-1", "@hermes_bot hi there")
        adapter.handle_message.assert_called_once()

    async def test_broadcast_mention_counts_as_addressed(self):
        """@here/@all address the bot too — one definition of 'addressed'
        serves this gate and the someone-else check alike (FDR-006)."""
        adapter = self._adapter()
        adapter.chatto_config.require_mention_rooms.value = ["room-1"]
        adapter._room_kinds["room-1"] = RoomKind.CHANNEL
        await self._dispatch(adapter, "room-1", "@here standup in 5")
        adapter.handle_message.assert_called_once()

    async def test_optional_list_room_answers_without_a_mention(self):
        adapter = self._adapter()
        adapter.chatto_config.optional_mention_rooms.value = ["room-1"]
        adapter._room_kinds["room-1"] = RoomKind.CHANNEL
        await self._dispatch(adapter, "room-1", "hi there")
        adapter.handle_message.assert_called_once()

    async def test_optional_list_beats_the_require_list_at_runtime(self):
        """validate_config rejects the overlap, but if contradictory config
        reaches a running adapter anyway, answering is safer than silence."""
        adapter = self._adapter()
        adapter.chatto_config.require_mention_rooms.value = ["room-1"]
        adapter.chatto_config.optional_mention_rooms.value = ["room-1"]
        adapter._room_kinds["room-1"] = RoomKind.CHANNEL
        assert adapter._room_policy("room-1") is RoomPolicy.OPEN

    async def test_dm_is_answered_without_a_mention(self):
        """The point of the room_kind check: mention gating must not mute DMs."""
        adapter = self._adapter()
        adapter._room_kinds["dm-1"] = RoomKind.DM
        await self._dispatch(adapter, "dm-1", "hi there")
        adapter.handle_message.assert_called_once()


# -- Room roster on channel_context --


def _roster_user(user_id, login, presence=PresenceStatus.UNSPECIFIED, deleted=False):
    """A directory user — used as input to _members() and _directory_members()."""
    return User(
        id=user_id,
        login=login,
        display_name=login.replace("_", " ").title(),
        presence_status=presence,
        deleted=deleted,
    )


class TestRoomRoster:
    """Every channel turn carries the member roster in channel_context —
    threads hold isolated sessions, so each turn needs its own copy. The
    roster is a room-scoped mini projection: fetched once, presence changes
    patch the cached users in place; membership events and reconnects discard
    a room so its next turn refetches once."""

    def _adapter(self):
        adapter = _make_adapter()
        adapter.chatto_config.allow_all_users.value = True
        adapter.me = _make_user("bot-user-id", "hermes_bot")
        adapter._user_cache["user-1"] = _roster_user("user-1", "alice")
        adapter.handle_message = AsyncMock()
        return adapter

    def _seed_channel(self, adapter, room_id="room-1"):
        adapter.chatto_config.require_mention_rooms.value = [room_id]
        adapter._room_kinds[room_id] = RoomKind.CHANNEL

    @staticmethod
    def _members(*users, total_count=None):
        member_ids = [user.id for user in users]
        page = Page(total_count=total_count or len(member_ids))
        return member_ids, page

    @staticmethod
    def _directory_members(*users):
        """DirectoryMember objects that batch_get_room_members() returns."""
        directory_members = [DirectoryMember(user=user) for user in users]
        return directory_members

    async def _dispatch(
        self,
        adapter,
        body,
        *,
        room_id="room-1",
        message_id="msg-1",
        thread_root=None,
    ):
        payload = _make_posted_payload(room_id=room_id, message_event_id=message_id)
        payload.thread_root_event_id = thread_root
        _with_get_message(
            adapter,
            payload,
            _make_message(body=body, room_id=room_id, message_id=message_id),
        )
        await adapter._dispatch_message_posted(payload, payload.message_event_id)

    async def _edit(self, adapter, body, *, message_id="msg-1"):
        event, payload = _make_edited_event(
            room_id="room-1", message_event_id=message_id
        )
        _with_get_message(
            adapter, payload, _make_message(body=body, message_id=message_id)
        )
        await adapter._handle_realtime_event(event)

    def _dispatched_event(self, adapter):
        return adapter.handle_message.await_args.args[0]

    async def test_new_thread_carries_the_roster(self):
        adapter = self._adapter()
        self._seed_channel(adapter)
        adapter._chatto_client.list_room_members = AsyncMock(
            return_value=self._members(
                _roster_user("user-1", "alice", PresenceStatus.ONLINE),
                _roster_user("user-2", "bob", PresenceStatus.OFFLINE),
                _roster_user("bot-user-id", "hermes_bot"),
                _roster_user("user-3", "carol", deleted=True),
            )
        )
        adapter._chatto_client.batch_get_room_members = AsyncMock(
            return_value=self._directory_members(
                _roster_user("user-1", "alice", PresenceStatus.ONLINE),
                _roster_user("user-2", "bob", PresenceStatus.OFFLINE),
                _roster_user("bot-user-id", "hermes_bot"),
                _roster_user("user-3", "carol", deleted=True),
            )
        )

        await self._dispatch(adapter, "@hermes_bot hi")

        context = self._dispatched_event(adapter).channel_context
        assert "@alice (Alice, online)" in context
        assert "@bob (Bob, offline)" in context
        # Our own account and deleted users are nobody the agent must greet.
        assert "@hermes_bot" not in context
        assert "carol" not in context

    async def test_follow_up_carries_the_roster_too(self):
        """Threads hold isolated sessions, so every channel turn needs its own
        copy of the roster — repetition is the point, deduplication a bug."""
        adapter = self._adapter()
        self._seed_channel(adapter)
        alice = _roster_user("user-1", "alice")
        adapter._chatto_client.list_room_members = AsyncMock(
            return_value=self._members(alice)
        )
        adapter._chatto_client.batch_get_room_members = AsyncMock(
            return_value=self._directory_members(alice)
        )

        await self._dispatch(adapter, "@hermes_bot hi")
        await self._dispatch(
            adapter,
            "@hermes_bot and one more thing",
            message_id="msg-2",
            thread_root="msg-1",
        )

        # Still one directory lookup: the projection spares the API, not the
        # prompt.
        assert adapter._chatto_client.list_room_members.await_count == 1
        context = self._dispatched_event(adapter).channel_context
        assert context is not None and "@alice" in context

    async def test_unspecified_room_kind_still_gets_one(self):
        """A server that never sets kind counts as a channel — roster too."""
        adapter = self._adapter()
        adapter.chatto_config.require_mention_rooms.value = ["room-1"]
        adapter._room_kinds["room-1"] = RoomKind.UNSPECIFIED
        alice = _roster_user("user-1", "alice")
        adapter._chatto_client.list_room_members = AsyncMock(
            return_value=self._members(alice)
        )
        adapter._chatto_client.batch_get_room_members = AsyncMock(
            return_value=self._directory_members(alice)
        )

        await self._dispatch(adapter, "@hermes_bot hi")

        assert "@alice" in self._dispatched_event(adapter).channel_context

    async def test_dm_never_carries_a_roster(self):
        adapter = self._adapter()
        adapter._room_kinds["dm-1"] = RoomKind.DM
        adapter._chatto_client.list_room_members = AsyncMock()
        adapter._chatto_client.batch_get_room_members = AsyncMock()

        await self._dispatch(adapter, "hi there", room_id="dm-1")

        adapter.handle_message.assert_called_once()
        adapter._chatto_client.list_room_members.assert_not_awaited()
        assert self._dispatched_event(adapter).channel_context is None

    async def test_directory_failure_still_dispatches(self):
        adapter = self._adapter()
        self._seed_channel(adapter)
        adapter._chatto_client.list_room_members = AsyncMock(
            side_effect=RuntimeError("directory down")
        )
        adapter._chatto_client.batch_get_room_members = AsyncMock()

        await self._dispatch(adapter, "@hermes_bot hi")

        adapter.handle_message.assert_called_once()
        assert self._dispatched_event(adapter).channel_context is None

    async def test_large_room_notes_the_unfetched_rest(self):
        adapter = self._adapter()
        self._seed_channel(adapter)
        alice = _roster_user("user-1", "alice")
        bob = _roster_user("user-2", "bob")
        adapter._chatto_client.list_room_members = AsyncMock(
            return_value=self._members(alice, bob, total_count=150)
        )
        adapter._chatto_client.batch_get_room_members = AsyncMock(
            return_value=self._directory_members(alice, bob)
        )

        await self._dispatch(adapter, "@hermes_bot hi")

        assert "… and 148 more" in self._dispatched_event(adapter).channel_context

    async def test_late_mention_edit_announces_exactly_once(self):
        """A forgotten mention added by edit opens a fresh turn — with the
        roster; the already-answered lock then keeps further edits out."""
        adapter = self._adapter()
        self._seed_channel(adapter)
        alice = _roster_user("user-1", "alice")
        adapter._chatto_client.list_room_members = AsyncMock(
            return_value=self._members(alice)
        )
        adapter._chatto_client.batch_get_room_members = AsyncMock(
            return_value=self._directory_members(alice)
        )

        await self._edit(adapter, "no mention yet")
        await self._edit(adapter, "@hermes_bot now it counts")
        await self._edit(adapter, "@hermes_bot once more")

        assert adapter.handle_message.await_count == 1
        assert adapter._chatto_client.list_room_members.await_count == 1

    async def test_announcement_guard_is_not_rearmed_by_empty_rosters(self):
        """A failed lookup must not burn the thread's one announcement."""
        adapter = self._adapter()
        self._seed_channel(adapter)
        adapter._chatto_client.list_room_members = AsyncMock(return_value=([], Page()))
        adapter._chatto_client.batch_get_room_members = AsyncMock()

        await self._dispatch(adapter, "@hermes_bot hi")
        assert self._dispatched_event(adapter).channel_context is None

        adapter.handle_message.reset_mock()
        alice = _roster_user("user-1", "alice")
        adapter._chatto_client.list_room_members = AsyncMock(
            return_value=self._members(alice)
        )
        adapter._chatto_client.batch_get_room_members = AsyncMock(
            return_value=self._directory_members(alice)
        )
        payload = _make_posted_payload(room_id="room-1", message_event_id="msg-9")
        _with_get_message(
            adapter,
            payload,
            _make_message(
                body="@hermes_bot retry", room_id="room-1", message_id="msg-9"
            ),
        )
        await adapter._dispatch_message_posted(payload, payload.message_event_id)

        assert adapter.handle_message.await_count == 1
        assert "@alice" in self._dispatched_event(adapter).channel_context

    async def test_presence_change_shows_on_next_turn_from_cache(self):
        """Presence moves patch the cached user; the next turn in the room
        renders the updated line — no second directory lookup."""
        adapter = self._adapter()
        self._seed_channel(adapter)
        # alice starts UNSPECIFIED (no presence label in the rendered entry).
        alice = _roster_user("user-1", "alice")
        adapter._chatto_client.list_room_members = AsyncMock(
            return_value=self._members(alice)
        )
        adapter._chatto_client.batch_get_room_members = AsyncMock(
            return_value=self._directory_members(alice)
        )

        await self._dispatch(adapter, "@hermes_bot hi")
        assert adapter._chatto_client.list_room_members.await_count == 1

        await adapter._handle_realtime_event(
            _make_presence_event(user_id="user-1", status=PresenceStatus.OFFLINE)
        )

        await self._dispatch(
            adapter,
            "@hermes_bot again",
            message_id="msg-2",
            thread_root="msg-1",
        )
        assert adapter._chatto_client.list_room_members.await_count == 1
        context = self._dispatched_event(adapter).channel_context
        assert context is not None and "@alice (Alice, offline)" in context

    async def test_presence_churn_costs_no_extra_lookups(self):
        """Rapid away/offline flapping is absorbed by the cache: the roster
        still rides on every turn, but the directory is never re-asked."""
        adapter = self._adapter()
        self._seed_channel(adapter)
        alice = _roster_user("user-1", "alice")
        adapter._chatto_client.list_room_members = AsyncMock(
            return_value=self._members(alice)
        )
        adapter._chatto_client.batch_get_room_members = AsyncMock(
            return_value=self._directory_members(alice)
        )

        await self._dispatch(adapter, "@hermes_bot hi")

        for _ in range(5):
            await adapter._handle_realtime_event(
                _make_presence_event(user_id="user-1", status=PresenceStatus.AWAY)
            )
        await self._dispatch(
            adapter,
            "@hermes_bot again",
            message_id="msg-2",
            thread_root="msg-1",
        )
        assert adapter._chatto_client.list_room_members.await_count == 1
        context = self._dispatched_event(adapter).channel_context
        assert context is not None and "(Alice, away)" in context

    async def test_non_member_presence_change_does_not_alter_the_line(self):
        """A presence move by someone outside the roster projection leaves the
        rendered line untouched — but the turn still carries it."""
        adapter = self._adapter()
        self._seed_channel(adapter)
        alice = _roster_user("user-1", "alice")
        adapter._chatto_client.list_room_members = AsyncMock(
            return_value=self._members(alice)
        )
        adapter._chatto_client.batch_get_room_members = AsyncMock(
            return_value=self._directory_members(alice)
        )

        await self._dispatch(adapter, "@hermes_bot hi")
        adapter._user_cache["user-9"] = _roster_user("user-9", "mallory")

        await adapter._handle_realtime_event(
            _make_presence_event(user_id="user-9", status=PresenceStatus.OFFLINE)
        )

        await self._dispatch(
            adapter,
            "@hermes_bot again",
            message_id="msg-2",
            thread_root="msg-1",
        )
        context = self._dispatched_event(adapter).channel_context
        assert context is not None and "mallory" not in context

    async def test_own_presence_change_is_ignored(self):
        """Our 60s presence refresh patches nothing — the agent must not see
        the bot in its own audience."""
        adapter = self._adapter()
        self._seed_channel(adapter)
        alice = _roster_user("user-1", "alice")
        adapter._chatto_client.list_room_members = AsyncMock(
            return_value=self._members(alice)
        )
        adapter._chatto_client.batch_get_room_members = AsyncMock(
            return_value=self._directory_members(alice)
        )

        await self._dispatch(adapter, "@hermes_bot hi")
        bot = _roster_user("bot-user-id", "hermes_bot", PresenceStatus.ONLINE)
        adapter._user_cache["bot-user-id"] = bot

        await adapter._handle_realtime_event(
            _make_presence_event(user_id="bot-user-id", status=PresenceStatus.OFFLINE)
        )

        assert bot.presence_status == PresenceStatus.ONLINE

    async def test_presence_change_before_any_announcement_is_noop(self):
        """No room announced yet — a presence event must not raise or fetch,
        and an unknown user stays unknown until a roster fetch names them."""
        adapter = self._adapter()
        self._seed_channel(adapter)
        adapter._chatto_client.list_room_members = AsyncMock()
        adapter._chatto_client.batch_get_room_members = AsyncMock()

        await adapter._handle_realtime_event(_make_presence_event(user_id="user-42"))

        assert adapter._chatto_client.list_room_members.await_count == 0
        assert "user-42" not in adapter._user_cache

    async def test_membership_event_refetches_once(self):
        """A join/leave makes the cached roster wrong: it is discarded and the
        next turn refetches once, delivering the updated line."""
        adapter = self._adapter()
        self._seed_channel(adapter)
        alice = _roster_user("user-1", "alice")
        bob = _roster_user("user-2", "bob")
        adapter._chatto_client.list_room_members = AsyncMock(
            return_value=self._members(alice, bob)
        )
        adapter._chatto_client.batch_get_room_members = AsyncMock(
            return_value=self._directory_members(alice, bob)
        )

        await self._dispatch(adapter, "@hermes_bot hi")
        assert "alice" in self._dispatched_event(adapter).channel_context

        await adapter._handle_realtime_event(
            _make_room_event("user_left_room", room_id="room-1")
        )
        assert "room-1" not in adapter._roster_announced
        assert "room-1" not in adapter._rosters

        # The server directory now reflects who actually remained.
        adapter._chatto_client.list_room_members.return_value = self._members(bob)
        adapter._chatto_client.batch_get_room_members.return_value = (
            self._directory_members(bob)
        )
        await self._dispatch(
            adapter,
            "@hermes_bot again",
            message_id="msg-2",
            thread_root="msg-1",
        )
        assert adapter._chatto_client.list_room_members.await_count == 2
        context = self._dispatched_event(adapter).channel_context
        assert context is not None and "@bob" in context and "alice" not in context

        await self._dispatch(
            adapter,
            "@hermes_bot third",
            message_id="msg-3",
            thread_root="msg-1",
        )
        # Still one refetch only — but the roster keeps riding on every turn.
        assert adapter._chatto_client.list_room_members.await_count == 2
        context = self._dispatched_event(adapter).channel_context
        assert context is not None and "@bob" in context

    async def test_reconnect_discards_cached_rosters_for_a_fresh_snapshot(self):
        """Protocol v1 sends no presence snapshot on subscribe, so every
        (re)connect discards the projections; rooms refetch lazily on their
        next turn instead of paying a burst right away."""
        adapter = self._adapter()
        self._seed_channel(adapter)
        alice = _roster_user("user-1", "alice")
        adapter._chatto_client.list_room_members = AsyncMock(
            return_value=self._members(alice)
        )
        adapter._chatto_client.batch_get_room_members = AsyncMock(
            return_value=self._directory_members(alice)
        )

        await self._dispatch(adapter, "@hermes_bot hi")
        assert adapter._roster_announced

        adapter._chatto_client.list_rooms = AsyncMock(return_value=[])
        await adapter._refresh_rooms()
        assert not adapter._roster_announced
        assert not adapter._rosters

        await self._dispatch(
            adapter,
            "@hermes_bot back online?",
            message_id="msg-2",
            thread_root="msg-1",
        )
        assert adapter._chatto_client.list_room_members.await_count == 2
        assert "@alice" in self._dispatched_event(adapter).channel_context

    async def test_stale_room_eviction_drops_its_roster(self):
        """Rooms we no longer belong to lose their projection along with their
        joined-list entry."""
        adapter = self._adapter()
        self._seed_channel(adapter, room_id="gone-1")
        adapter._joined_room_ids = ["gone-1"]
        alice = _roster_user("user-1", "alice")
        adapter._chatto_client.list_room_members = AsyncMock(
            return_value=self._members(alice)
        )
        adapter._chatto_client.batch_get_room_members = AsyncMock(
            return_value=self._directory_members(alice)
        )

        await self._dispatch(adapter, "@hermes_bot hi", room_id="gone-1")
        assert "gone-1" in adapter._rosters

        kept = _make_room_state(_make_room("kept", "Kept", RoomKind.CHANNEL), True)
        adapter._chatto_client.list_rooms = AsyncMock(return_value=[kept])
        await adapter._refresh_rooms()

        assert "gone-1" not in adapter._roster_announced
        assert "gone-1" not in adapter._rosters


# -- Inbound edits (edit-dispatch) --


def _make_edited_event(room_id="room-1", message_event_id="msg-1", actor_id="human-1"):
    """A message_edited envelope whose payload the caller stubs."""
    event = MagicMock()
    event.id = "evt-edit-1"
    event.kind = "message_edited"
    event.actor_id = actor_id
    payload = MagicMock()
    payload.room_id = room_id
    payload.message_event_id = message_event_id
    event.payload = payload
    return event, payload


def _make_presence_event(
    user_id="user-1", actor_id=None, status=PresenceStatus.OFFLINE
):
    """A presence_changed envelope — actor is the user whose status moved."""
    event = MagicMock()
    event.id = f"evt-presence-{user_id}"
    event.kind = "presence_changed"
    event.actor_id = actor_id if actor_id is not None else user_id
    payload = MagicMock()
    payload.user_id = user_id
    payload.status = status
    event.payload = payload
    return event


def _make_room_event(kind, room_id="room-1", actor_id="user-1"):
    """A user_joined_room / user_left_room envelope."""
    event = MagicMock()
    event.id = f"evt-{kind}-{room_id}"
    event.kind = kind
    event.actor_id = actor_id
    payload = MagicMock()
    payload.room_id = room_id
    event.payload = payload
    return event


class TestEditDispatch:
    """message_edited events become corrections, late mentions or nothing."""

    def _adapter(self):
        adapter = _make_adapter()
        adapter.chatto_config.allow_all_users.value = True
        adapter.me = _make_user("bot-user-id", "hermes_bot")
        adapter._user_cache["user-1"] = _make_user("user-1", "alice")
        adapter.handle_message = AsyncMock()
        return adapter

    async def _edit(
        self,
        adapter,
        body,
        *,
        message_id="msg-1",
        room_id="room-1",
        created_at=None,
        updated_at=None,
    ):
        event, payload = _make_edited_event(
            room_id=room_id, message_event_id=message_id
        )
        _with_get_message(
            adapter,
            payload,
            _make_message(
                body=body,
                message_id=message_id,
                room_id=room_id,
                created_at=created_at,
                updated_at=updated_at,
            ),
        )
        await adapter._handle_realtime_event(event)
        return payload

    async def _post_original(self, adapter, body, *, message_id, room_id):
        """Dispatch the original posted message so its turn exists."""
        payload = _make_posted_payload(room_id=room_id, message_event_id=message_id)
        _with_get_message(
            adapter,
            payload,
            _make_message(body=body, message_id=message_id, room_id=room_id),
        )
        await adapter._dispatch_message_posted(payload, payload.message_event_id)

    def test_edit_dispatch_defaults_on_with_five_minute_window(self):
        adapter = self._adapter()
        assert adapter.chatto_config.edit_dispatch.value is True
        assert adapter.chatto_config.edit_window.value == 300

    async def test_own_edit_echo_is_ignored(self):
        """Our streaming edits echo back as message_edited — never inbound traffic."""
        adapter = self._adapter()
        event, _payload = _make_edited_event(actor_id="bot-user-id")
        adapter._chatto_client.get_message = AsyncMock()

        await adapter._handle_realtime_event(event)

        adapter._chatto_client.get_message.assert_not_awaited()
        adapter.handle_message.assert_not_called()

    async def test_disabled_feature_drops_edits(self):
        adapter = self._adapter()
        adapter.chatto_config.edit_dispatch.value = False

        await self._edit(adapter, "corrected")

        adapter.handle_message.assert_not_called()

    async def test_read_only_room_costs_no_api_call(self):
        """Silent-room gating sits before fetch_message, like for posts."""
        adapter = self._adapter()

        await self._edit(adapter, "corrected")
        adapter._chatto_client.get_message.assert_not_awaited()  # type: ignore[union-attr]
        adapter.handle_message.assert_not_called()

    async def test_mention_added_by_edit_starts_a_turn(self):
        adapter = self._adapter()
        adapter._room_kinds["room-1"] = RoomKind.CHANNEL
        adapter.chatto_config.require_mention_rooms.value = ["room-1"]

        await self._edit(adapter, "@hermes_bot corrected text")

        adapter.handle_message.assert_called_once()
        assert (
            adapter.handle_message.call_args.args[0].text
            == "@hermes_bot corrected text"
        )
        assert "msg-1" in adapter._dispatched_ids

    async def test_edit_without_the_mention_stays_silent(self):
        """Gates run against the new body — no mention in, no answer out."""
        adapter = self._adapter()
        adapter._room_kinds["room-1"] = RoomKind.CHANNEL
        adapter.chatto_config.require_mention_rooms.value = ["room-1"]

        await self._edit(adapter, "still no mention")

        adapter.handle_message.assert_not_called()

    async def test_already_answered_message_stays_answered(self):
        """The retro-lock: editing an old answered message must not re-animate it."""
        adapter = self._adapter()
        adapter._dispatched_ids.append("msg-1")

        await self._edit(adapter, "late fix")

        adapter.handle_message.assert_not_called()

    async def test_deleted_message_is_ignored(self):
        adapter = self._adapter()
        now = datetime.now(UTC)
        event, payload = _make_edited_event()
        message = _make_message(body="gone", created_at=now, updated_at=now)
        message.deleted_at = now
        _with_get_message(adapter, payload, message)

        await adapter._handle_realtime_event(event)

        adapter.handle_message.assert_not_called()

    async def test_stale_edit_outside_the_window_is_ignored(self):
        adapter = self._adapter()
        now = datetime.now(UTC)

        await self._edit(
            adapter,
            "very late fix",
            created_at=now,
            updated_at=now + timedelta(seconds=400),
        )

        adapter.handle_message.assert_not_called()

    async def test_fresh_edit_within_the_window_passes_gates(self):
        adapter = self._adapter()
        adapter._room_kinds["dm-room"] = RoomKind.DM
        now = datetime.now(UTC)

        await self._edit(
            adapter,
            "typo fixed",
            room_id="dm-room",
            created_at=now,
            updated_at=now + timedelta(seconds=5),
        )

        adapter.handle_message.assert_called_once()

    async def test_midrun_edit_restarts_the_turn(self):
        adapter = self._adapter()
        adapter._room_kinds["dm-room"] = RoomKind.DM
        await self._post_original(
            adapter, "orignal text", message_id="msg-run", room_id="dm-room"
        )
        first_event = adapter.handle_message.call_args.args[0]
        await adapter.on_processing_start(first_event)
        adapter.cancel_session_processing = AsyncMock()

        await self._edit(
            adapter, "original text", message_id="msg-run", room_id="dm-room"
        )

        adapter.cancel_session_processing.assert_awaited_once()
        assert adapter.handle_message.await_count == 2
        assert adapter.handle_message.await_args.args[0].text == "original text"

    async def test_completion_hook_rearms_the_processing_map(self):
        """After any outcome (incl. CANCELLED) the session accepts new turns."""
        adapter = self._adapter()
        adapter._room_kinds["dm-room"] = RoomKind.DM
        await self._post_original(
            adapter, "hello", message_id="msg-run", room_id="dm-room"
        )
        event = adapter.handle_message.call_args.args[0]

        await adapter.on_processing_start(event)
        session_key = adapter._session_key_for(event.source)
        assert adapter._processing.get(session_key) == "msg-run"

        await adapter.on_processing_complete(event, ProcessingOutcome.CANCELLED)
        assert session_key not in adapter._processing

    async def test_queued_followup_gets_corrected_in_place(self):
        """A still-queued message answers later with its edited wording."""
        from types import SimpleNamespace

        adapter = self._adapter()
        adapter._room_kinds["dm-room"] = RoomKind.DM
        await self._post_original(
            adapter, "queue me", message_id="msg-other", room_id="dm-room"
        )
        other_event = adapter.handle_message.call_args.args[0]
        session_key = adapter._session_key_for(other_event.source)
        pending = SimpleNamespace(message_id="msg-run", text="stale wording")
        adapter._pending_messages[session_key] = pending
        adapter._dispatched_ids.extend(["msg-other", "msg-run"])
        adapter.cancel_session_processing = AsyncMock()

        await self._edit(
            adapter, "fixed wording", message_id="msg-run", room_id="dm-room"
        )

        assert pending.text == "fixed wording"
        adapter.cancel_session_processing.assert_not_awaited()
        # No second dispatch: the queued original carries the correction.
        assert adapter.handle_message.await_count == 1

    async def test_dm_command_edit_neither_reruns_nor_dispatches(self):
        """/join ran when posted; its edited copy must not run twice or leak."""
        adapter = self._adapter()
        adapter._room_kinds["dm-room"] = RoomKind.DM
        adapter._handle_dm_command = AsyncMock(return_value=True)

        await self._edit(adapter, "/leave", room_id="dm-room")

        adapter._handle_dm_command.assert_not_called()
        adapter.handle_message.assert_not_called()


# -- Inbound attachments --


class TestInboundAttachments:
    """Messages carrying files must reach the agent, body or not."""

    @pytest_asyncio.fixture
    def adapter(self):
        adapter = _make_adapter()
        adapter.chatto_config.allow_all_users.value = True
        adapter.me = _make_user("bot-user-id", "hermes_bot")
        adapter._room_kinds["room-1"] = RoomKind.DM
        adapter._user_cache["user-1"] = _make_user("user-1", "alice")
        adapter.handle_message = AsyncMock()
        adapter._download_attachment_bytes = AsyncMock(
            return_value=b"\x89PNG\r\n\x1a\nrest"
        )
        return adapter

    async def test_image_attachment_becomes_media_url(self, adapter):
        payload = _make_posted_payload()
        adapter._chatto_client.get_room = AsyncMock()
        message = _make_message(
            body="look at this",
            attachments=[_make_attachment("shot.png", "image/png")],
        )
        _with_get_message(adapter, payload, message)

        with patch(
            "adapter.cache_media_bytes",
            return_value=_cached("/cache/shot.png", "image/png", "image"),
        ):
            await adapter._dispatch_message_posted(payload, payload.message_event_id)

        event = adapter.handle_message.call_args.args[0]
        assert event.media_urls == ["/cache/shot.png"]
        assert event.media_types == ["image/png"]
        assert event.message_type == MessageType.PHOTO

    async def test_attachment_only_message_is_not_dropped(self, adapter):
        """The empty-body early return is what silently ate file uploads."""
        payload = _make_posted_payload()
        message = _make_message(
            body="",
            attachments=[_make_attachment("report.pdf", "application/pdf")],
        )
        _with_get_message(adapter, payload, message)

        with patch(
            "adapter.cache_media_bytes",
            return_value=_cached("/cache/report.pdf", "application/pdf", "document"),
        ):
            await adapter._dispatch_message_posted(payload, payload.message_event_id)

        adapter.handle_message.assert_called_once()
        event = adapter.handle_message.call_args.args[0]
        assert event.message_type == MessageType.DOCUMENT
        assert event.media_urls == ["/cache/report.pdf"]

    async def test_empty_message_without_attachments_is_dropped(self, adapter):
        payload = _make_posted_payload()
        _with_get_message(adapter, payload, _make_message(body=""))
        await adapter._dispatch_message_posted(payload, payload.message_event_id)
        adapter.handle_message.assert_not_called()

    async def test_download_failure_still_delivers_the_text(self, adapter):
        adapter._download_attachment_bytes = AsyncMock(side_effect=RuntimeError("404"))
        payload = _make_posted_payload()
        _with_get_message(
            adapter,
            payload,
            _make_message(
                body="see attached",
                attachments=[_make_attachment("a.png", "image/png")],
            ),
        )

        await adapter._dispatch_message_posted(payload, payload.message_event_id)

        event = adapter.handle_message.call_args.args[0]
        assert event.text == "see attached"
        assert event.media_urls == []
        assert event.message_type == MessageType.TEXT

    async def test_attachment_without_asset_url_is_skipped(self, adapter):
        """Videos are announced before transcoding finishes."""
        payload = _make_posted_payload()
        att = _make_attachment("clip.mp4", "video/mp4")
        att.asset_url = None
        _with_get_message(
            adapter,
            payload,
            _make_message(
                body="clip",
                attachments=[att],
            ),
        )

        await adapter._dispatch_message_posted(payload, payload.message_event_id)

        event = adapter.handle_message.call_args.args[0]
        assert event.media_urls == []
        adapter._download_attachment_bytes.assert_not_called()

    async def test_document_wins_over_image(self, adapter):
        """Mixed batches classify as DOCUMENT — that gates context injection."""
        assert (
            adapter._message_type_for_media_kinds(["image", "document"])
            is MessageType.DOCUMENT
        )
        assert adapter._message_type_for_media_kinds(["image"]) is MessageType.PHOTO
        assert adapter._message_type_for_media_kinds(["video"]) is MessageType.VIDEO
        assert adapter._message_type_for_media_kinds(["audio"]) is MessageType.AUDIO
        assert adapter._message_type_for_media_kinds([]) is MessageType.TEXT

    async def test_oversized_attachment_is_rejected(self, adapter):
        """The gateway media cap must bound what a hostile upload can buffer."""
        import httpx

        big = get_inbound_media_max_bytes() + 1
        transport = httpx.MockTransport(
            lambda request: httpx.Response(
                200,
                headers={"content-length": str(big)},
                content=b"x",
            )
        )
        real_adapter = _make_adapter()
        real_client_cls = httpx.AsyncClient

        with (
            patch(
                "httpx.AsyncClient", lambda **kw: real_client_cls(transport=transport)
            ),
            pytest.raises(ValueError),
        ):
            await real_adapter._download_attachment_bytes(
                "https://chat.example.com/a.png"
            )


# NOTE: there are deliberately no tests for get_user(), set_presence() or
# set_custom_status() on the adapter. Those are not adapter responsibilities —
# callers use the chattolib client directly, which exposes them (client.get_user,
# client.set_presence, client.update_custom_status). The adapter only touches
# presence in connect()/disconnect().


# -- Room operations --


class TestRoomOperations:
    """Test room creation and DM initiation."""

    @pytest_asyncio.fixture
    def adapter(self):
        _clear_chatto_env()
        cfg = _make_config()
        adapter = ChattoAdapter(cfg)
        adapter._chatto_client = MagicMock()
        # AsyncMock, not MagicMock: the adapter awaits these, and awaiting a
        # plain MagicMock raises TypeError, which create_room()/start_dm()
        # swallow into a None return.
        adapter._chatto_client.create_room = AsyncMock()
        adapter._chatto_client.start_dm = AsyncMock()
        adapter._room_names = {}
        adapter._room_kinds = {}
        return adapter

    async def test_create_room(self, adapter):
        adapter._chatto_client.create_room.return_value = _make_room(
            "room-123",
            "Test Room",
            RoomKind.CHANNEL,
        )
        result = await adapter.create_room("Test Room", "A test room")
        assert result == "room-123"
        adapter._chatto_client.create_room.assert_called_once()
        assert adapter._room_names["room-123"] == "Test Room"

    async def test_start_dm(self, adapter):
        adapter._chatto_client.start_dm.return_value = _make_room(
            "dm-123",
            "DM with user",
            RoomKind.DM,
        )
        result = await adapter.start_dm("user-123")
        assert result == "dm-123"
        adapter._chatto_client.start_dm.assert_called_once()
        assert adapter._room_kinds["dm-123"] == RoomKind.DM


# -- DM room management (/join, /leave) --


def _make_room_state(room, is_member):
    """Build a RoomWithViewerState the way list_rooms()/get_room() return it."""
    return RoomWithViewerState(
        room=room,
        viewer_state=RoomViewerState(is_member=is_member),
    )


class TestDmRoomCommands:
    """/join and /leave arrive over DMs, change server-side membership and
    must never reach the agent pipeline."""

    def _adapter(self):
        adapter = _make_adapter()
        adapter.chatto_config.allow_all_users.value = True
        adapter.me = _make_user("bot-user-id", "hermes_bot")
        adapter._user_cache["user-1"] = _make_user("user-1", "alice")
        client = adapter._chatto_client
        client.list_rooms = AsyncMock(return_value=[])
        client.get_room_events = AsyncMock(return_value=MagicMock(events=[]))
        client.join_room = AsyncMock()
        client.leave_room = AsyncMock(return_value=True)
        client.get_room = AsyncMock()
        adapter._room_kinds["dm-1"] = RoomKind.DM
        adapter.handle_message = AsyncMock()
        adapter.send = AsyncMock()
        return adapter

    async def _dispatch(self, adapter, body, room_id="dm-1"):
        payload = _make_posted_payload(room_id=room_id)
        _with_get_message(adapter, payload, _make_message(body=body, room_id=room_id))
        await adapter._dispatch_message_posted(payload, payload.message_event_id)

    def _reply(self, adapter):
        assert adapter.send.await_count == 1
        return adapter.send.await_args.kwargs["content"]

    async def test_join_silent_channel_names_the_env_lines(self):
        """Users copy room IDs from this reply — an unlisted channel hands
        them both mention-list lines verbatim instead of staying quiet."""
        adapter = self._adapter()
        adapter._seed_room = AsyncMock()
        state = _make_room_state(
            _make_room("room-9", "Deploy", RoomKind.CHANNEL), False
        )
        adapter._chatto_client.list_rooms = AsyncMock(return_value=[state])
        adapter._chatto_client.join_room = AsyncMock(return_value=state.room)

        await self._dispatch(adapter, "/join #deploy")

        adapter._chatto_client.join_room.assert_awaited_once_with("room-9")
        assert adapter._joined_room_ids == ["room-9"]
        reply = self._reply(adapter)
        assert "Joined 'Deploy' (room-9)" in reply
        assert "CHATTO_REQUIRE_MENTION_ROOMS=room-9" in reply
        assert "CHATTO_OPTIONAL_MENTION_ROOMS=room-9" in reply

    async def test_join_silent_channel_does_not_seed(self):
        """Same rule as _refresh_rooms: history nothing will ever answer is
        not pushed into context."""
        adapter = self._adapter()
        adapter._seed_room = AsyncMock()
        state = _make_room_state(
            _make_room("room-9", "Deploy", RoomKind.CHANNEL), False
        )
        adapter._chatto_client.list_rooms = AsyncMock(return_value=[state])
        adapter._chatto_client.join_room = AsyncMock(return_value=state.room)

        await self._dispatch(adapter, "/join #deploy")

        adapter._seed_room.assert_not_awaited()

    async def test_join_listed_channel_reports_policy_and_seeds(self):
        adapter = self._adapter()
        adapter.chatto_config.require_mention_rooms.value = ["room-9"]
        adapter._seed_room = AsyncMock()
        state = _make_room_state(
            _make_room("room-9", "Deploy", RoomKind.CHANNEL), False
        )
        adapter._chatto_client.list_rooms = AsyncMock(return_value=[state])
        adapter._chatto_client.join_room = AsyncMock(return_value=state.room)

        await self._dispatch(adapter, "/join #deploy")

        adapter._seed_room.assert_awaited_once_with("room-9")
        reply = self._reply(adapter)
        assert "answers only @mentions" in reply
        assert "CHATTO_OPTIONAL_MENTION_ROOMS=" not in reply

    async def test_join_room_with_unknown_kind_gets_the_channel_hint(self):
        """A server that never sets kind counts as a channel too — the reply
        carries the mention-list hint and nothing is seeded."""
        adapter = self._adapter()
        adapter._seed_room = AsyncMock()
        state = _make_room_state(
            _make_room("grp-9", "Deploy", RoomKind.UNSPECIFIED), False
        )
        adapter._chatto_client.list_rooms = AsyncMock(return_value=[state])
        adapter._chatto_client.join_room = AsyncMock(return_value=state.room)

        await self._dispatch(adapter, "/join #deploy")

        adapter._chatto_client.join_room.assert_awaited_once_with("grp-9")
        assert adapter._joined_room_ids == ["grp-9"]
        adapter._seed_room.assert_not_awaited()
        reply = self._reply(adapter)
        assert "Joined 'Deploy' (grp-9)" in reply
        assert "CHATTO_REQUIRE_MENTION_ROOMS=grp-9" in reply

    async def test_join_skips_rpc_when_already_a_member(self):
        """Natively invited accounts hold membership already — they only need
        seeding into the joined list (which silent channels skip)."""
        adapter = self._adapter()
        adapter._seed_room = AsyncMock()
        state = _make_room_state(_make_room("room-9", "Deploy", RoomKind.CHANNEL), True)
        adapter._chatto_client.list_rooms = AsyncMock(return_value=[state])

        await self._dispatch(adapter, "/join #deploy")

        adapter._chatto_client.join_room.assert_not_awaited()
        assert adapter._joined_room_ids == ["room-9"]
        assert "Already a member" in self._reply(adapter)

    async def test_join_unknown_name_reports_without_joining(self):
        adapter = self._adapter()
        adapter._chatto_client.list_rooms = AsyncMock(return_value=[])

        await self._dispatch(adapter, "/join #nope")

        adapter._chatto_client.join_room.assert_not_awaited()
        assert "No room named '#nope'" in self._reply(adapter)
        assert adapter._joined_room_ids == []

    async def test_ambiguous_name_offers_the_candidate_ids(self):
        adapter = self._adapter()
        matches = [
            _make_room_state(_make_room(f"r-{i}", "General", RoomKind.CHANNEL), False)
            for i in range(2)
        ]
        adapter._chatto_client.list_rooms = AsyncMock(return_value=matches)
        adapter._chatto_client.join_room = AsyncMock()

        await self._dispatch(adapter, "/join #general")

        adapter._chatto_client.join_room.assert_not_awaited()
        reply = self._reply(adapter)
        assert "r-0" in reply and "r-1" in reply

    async def test_join_by_room_id_verifies_via_get_room(self):
        adapter = self._adapter()
        state = _make_room_state(
            _make_room("room-9", "Deploy", RoomKind.CHANNEL), False
        )
        adapter._chatto_client.get_room = AsyncMock(return_value=state)
        adapter._chatto_client.join_room = AsyncMock(return_value=state.room)

        await self._dispatch(adapter, "/join room-9")

        adapter._chatto_client.get_room.assert_awaited_once_with("room-9")
        adapter._chatto_client.join_room.assert_awaited_once_with("room-9")
        assert adapter._joined_room_ids == ["room-9"]

    async def test_leave_drops_the_room_from_the_joined_list(self):
        adapter = self._adapter()
        adapter._joined_room_ids = ["room-7"]
        adapter._roster_announced.append("room-7")
        adapter._rosters["room-7"] = MagicMock()
        state = _make_room_state(_make_room("room-7", "Deploy", RoomKind.CHANNEL), True)
        adapter._chatto_client.get_room = AsyncMock(return_value=state)

        await self._dispatch(adapter, "/leave room-7")

        adapter._chatto_client.leave_room.assert_awaited_once_with("room-7")
        assert adapter._joined_room_ids == []
        # Leaving the audience means keeping no roster projection for it.
        assert "room-7" not in adapter._roster_announced
        assert "room-7" not in adapter._rosters
        assert "Left 'Deploy' (room-7)" in self._reply(adapter)

    async def test_leave_refuses_direct_messages(self):
        adapter = self._adapter()
        state = _make_room_state(_make_room("dm-2", "", RoomKind.DM), True)
        adapter._chatto_client.get_room = AsyncMock(return_value=state)

        await self._dispatch(adapter, "/leave dm-2")

        adapter._chatto_client.leave_room.assert_not_awaited()
        assert "Direct messages cannot be left" in self._reply(adapter)

    async def test_leave_refuses_home_channel(self):
        """Leaving CHATTO_HOME_CHANNEL would break cron/notification delivery."""
        adapter = self._adapter()
        adapter.chatto_config.home_channel.value = "room-7"
        adapter._joined_room_ids = ["room-7"]
        state = _make_room_state(_make_room("room-7", "Home", RoomKind.CHANNEL), True)
        adapter._chatto_client.get_room = AsyncMock(return_value=state)

        await self._dispatch(adapter, "/leave room-7")

        adapter._chatto_client.leave_room.assert_not_awaited()
        assert "home channel" in self._reply(adapter)

    async def test_commands_outside_dms_are_ignored(self):
        """In a channel the text is just a message — mention gating applies,
        no command runs, nothing is sent."""
        adapter = self._adapter()
        adapter.chatto_config.require_mention_rooms.value = ["chan-1"]
        adapter._room_kinds["chan-1"] = RoomKind.CHANNEL

        await self._dispatch(adapter, "/leave room-7", room_id="chan-1")

        adapter._chatto_client.leave_room.assert_not_awaited()
        adapter.handle_message.assert_not_called()
        adapter.send.assert_not_called()

    async def test_non_command_dm_falls_through_to_pipeline(self):
        adapter = self._adapter()
        await self._dispatch(adapter, "/status all good")
        adapter.handle_message.assert_awaited_once()
        adapter.send.assert_not_called()

    async def test_missing_argument_gets_usage_reply(self):
        adapter = self._adapter()
        for body in ("/join", "/leave"):
            adapter.send.reset_mock()
            await self._dispatch(adapter, body)
            assert self._reply(adapter).startswith("Usage:")


class TestJoinedRoomsRefresh:
    """_refresh_rooms mirrors the joined list against server-side membership."""

    def _adapter(self):
        adapter = _make_adapter()
        client = adapter._chatto_client
        client.list_rooms = AsyncMock(return_value=[])
        return adapter

    async def test_drops_rooms_no_longer_joined(self):
        adapter = self._adapter()
        adapter._joined_room_ids = ["gone-1", "kept"]
        kept = _make_room_state(_make_room("kept", "Kept", RoomKind.CHANNEL), True)
        adapter._chatto_client.list_rooms = AsyncMock(return_value=[kept])

        await adapter._refresh_rooms()

        assert adapter._joined_room_ids == ["kept"]

    async def test_warns_once_when_home_channel_is_not_joined(self):
        adapter = self._adapter()
        adapter.chatto_config.home_channel.value = "home-x"
        other = _make_room_state(_make_room("other", "Other", RoomKind.CHANNEL), True)
        adapter._chatto_client.list_rooms = AsyncMock(return_value=[other])

        await adapter._refresh_rooms()
        assert adapter._home_warning_logged

        await adapter._refresh_rooms()
        assert adapter._home_warning_logged

    async def test_no_warning_while_home_channel_is_member(self):
        adapter = self._adapter()
        adapter.chatto_config.home_channel.value = "home-x"
        home = _make_room_state(_make_room("home-x", "Home", RoomKind.CHANNEL), True)
        adapter._chatto_client.list_rooms = AsyncMock(return_value=[home])

        await adapter._refresh_rooms()
        assert not adapter._home_warning_logged


class TestSilentRoomRefresh:
    """_refresh_rooms keeps silent rooms joined but skips seeding them."""

    def _adapter(self, optional_mention_rooms):
        adapter = _make_adapter()
        adapter.chatto_config.optional_mention_rooms.value = optional_mention_rooms
        client = adapter._chatto_client
        client.list_rooms = AsyncMock(return_value=[])
        client.join_room = AsyncMock()
        adapter._seed_room = AsyncMock()
        return adapter

    async def test_silent_rooms_are_joined_but_not_seeded(self):
        adapter = self._adapter(["team-1"])
        news = _make_room_state(_make_room("news-1", "News", RoomKind.CHANNEL), True)
        team = _make_room_state(_make_room("team-1", "Team", RoomKind.CHANNEL), True)
        adapter._chatto_client.list_rooms = AsyncMock(return_value=[news, team])

        await adapter._refresh_rooms()

        assert sorted(adapter._joined_room_ids) == ["news-1", "team-1"]
        adapter._seed_room.assert_awaited_once_with("team-1")

    async def test_join_log_tags_policies_kinds_and_universal(self, caplog):
        adapter = self._adapter(["team-1"])
        adapter.chatto_config.require_mention_rooms.value = ["dep-1"]
        news = Room(id="news-1", name="News", kind=RoomKind.CHANNEL, universal=True)
        states = [
            _make_room_state(news, True),
            _make_room_state(_make_room("team-1", "Team", RoomKind.CHANNEL), True),
            _make_room_state(_make_room("dep-1", "Deploy", RoomKind.CHANNEL), True),
            _make_room_state(_make_room("dm-1", "Paula", RoomKind.DM), True),
        ]
        adapter._chatto_client.list_rooms = AsyncMock(return_value=states)

        with caplog.at_level("INFO"):
            await adapter._refresh_rooms()

        joined = [m for m in caplog.messages if "currently joined" in m]
        assert joined, "expected the joined-rooms summary log line"
        summary = joined[-1]
        assert '"News" (news-1) [read-only] [universal]' in summary
        assert '"Team" (team-1)' in summary and "[every-message]" in summary
        assert '"Deploy" (dep-1) [on-mention]' in summary
        assert '"Paula" (dm-1) [dm]' in summary

    async def test_join_log_shows_dm_partner_username(self, caplog):
        adapter = self._adapter([])
        adapter.me = User(id="bot-1", login="hermes", display_name="Hermes")
        dm_users = [
            DirectoryMember(user=adapter.me),
            DirectoryMember(user=_make_user("user-9", "paula")),
        ]

        async def list_members(room_id, **_kwargs):
            members = dm_users if room_id == "dm-1" else []
            return [m.user.id for m in members], Page()

        async def batch_members(room_id, user_ids):
            return [dm for dm in dm_users if dm.user.id in user_ids]

        adapter._chatto_client.list_room_members = AsyncMock(side_effect=list_members)
        adapter._chatto_client.batch_get_room_members = AsyncMock(
            side_effect=batch_members
        )
        dm = _make_room_state(_make_room("dm-1", "", RoomKind.DM), True)
        adapter._chatto_client.list_rooms = AsyncMock(return_value=[dm])

        with caplog.at_level("INFO"):
            await adapter._refresh_rooms()

        joined = [m for m in caplog.messages if "currently joined" in m]
        assert joined, "expected the joined-rooms summary log line"
        assert '"paula" (dm-1) [dm]' in joined[-1]

    async def test_join_log_appears_on_every_refresh(self, caplog):
        adapter = self._adapter(["team-1"])
        team = _make_room_state(_make_room("team-1", "Team", RoomKind.CHANNEL), True)
        adapter._chatto_client.list_rooms = AsyncMock(return_value=[team])

        with caplog.at_level("INFO"):
            await adapter._refresh_rooms()
            await adapter._refresh_rooms()

        joined = [m for m in caplog.messages if "currently joined" in m]
        assert len(joined) == 2


# -- Constants --


class TestConstants:
    """Test that constants are properly defined."""

    def test_max_message_length(self):
        assert _MAX_MESSAGE_LENGTH == 10000

    def test_seen_cap(self):
        assert _SEEN_CAP == 500


# -- Client creation credentials --


class TestEnvEnablement:
    """hermes_env_enablement_fn seeds extra verbatim from set env vars only.

    The gateway commits this dict onto the platform's extra unconditionally,
    so a fabricated default would overwrite what the user set in config.yaml.
    """

    def _seed(self, monkeypatch, **env):
        for key in _CHATTO_ENV_KEYS:
            monkeypatch.delenv(key, raising=False)
        for key, value in env.items():
            monkeypatch.setenv(key, value)
        return hermes_env_enablement_fn()

    def test_no_env_vars_yield_an_empty_seed(self, monkeypatch):
        assert self._seed(monkeypatch) == {}

    def test_set_base_url_is_seeded(self, monkeypatch):
        seed = self._seed(monkeypatch, CHATTO_BASE_URL="  https://chat.example.com  ")
        assert seed == {"base_url": "https://chat.example.com"}

    def test_unset_base_url_is_not_fabricated(self, monkeypatch):
        """The ChattoHQ default is the ConfigField's business, not a seed
        entry that would clobber a YAML-configured base_url."""
        assert "base_url" not in self._seed(monkeypatch)

    def test_values_are_seeded_under_their_config_keys(self, monkeypatch):
        seed = self._seed(
            monkeypatch,
            CHATTO_HOME_CHANNEL="room-9",
            CHATTO_REQUIRE_MENTION_ROOMS="a,b",
            CHATTO_AUTO_THREAD="false",
        )
        assert seed == {
            "home_channel": "room-9",
            "require_mention_rooms": "a,b",
            "auto_thread": "false",
        }

    def test_token_is_seeded(self, monkeypatch):
        seed = self._seed(monkeypatch, CHATTO_TOKEN="cht_BK_test")
        assert seed == {"token": "cht_BK_test"}


class TestOpenClientToken:
    """Token-only auth: no token means a normal failure (ValueError), not a crash."""

    def _adapter(self):
        return _make_adapter()

    async def test_token_creates_client(self):
        """The CHATTO_TOKEN path must produce a ChattoClient."""
        adapter = self._adapter()
        client = await adapter._open_client(
            base_url="https://chat.example.com", token="cht_BK_test"
        )
        assert isinstance(client, ChattoClient)

    async def test_no_token_raises_value_error(self):
        """Misconfiguration reads as a logged creation failure, not a crash."""
        adapter = self._adapter()
        with pytest.raises(ValueError, match="CHATTO_TOKEN not configured"):
            await adapter._open_client(base_url="https://chat.example.com", token=None)


# -- Cron / standalone delivery --


class TestStandaloneSender:
    """hermes_standalone_sender_fn normalises and splits like send() does —
    cron output must not die on the server's per-message limit."""

    def _posted(self, count):
        """A client whose posts succeed; returns (client, sent_ids)."""
        sent_ids = []

        async def post(chat_id, body, **kwargs):
            msg = MagicMock()
            msg.id = f"msg-{len(sent_ids)}"
            sent_ids.append(msg.id)
            return msg

        client = MagicMock()
        client.post_message = AsyncMock(side_effect=post)
        client.close = AsyncMock()
        return client, sent_ids

    async def _send(self, message, **kwargs):
        cfg = _make_config(token="test-token")
        client = MagicMock()
        client.post_message = AsyncMock(return_value=MagicMock(id="msg-only"))
        client.close = AsyncMock()
        with patch("adapter.ChattoClient") as client_cls:
            client_cls.return_value = client
            result = await hermes_standalone_sender_fn(cfg, "room-1", message, **kwargs)
        return result, client

    async def test_short_message_posts_once(self):
        result, client = await self._send("hello")
        assert result.success is True
        assert result.message_id == "msg-only"
        client.post_message.assert_awaited_once()

    async def test_long_message_is_split(self):
        """A 12k-char cron job becomes several posts, first ID reported."""
        cfg = _make_config(token="test-token")
        client, sent_ids = self._posted(5)
        with patch("adapter.ChattoClient") as client_cls:
            client_cls.return_value = client
            result = await hermes_standalone_sender_fn(cfg, "room-1", "x" * 12000)

        assert result.success is True
        assert result.message_id == "msg-0"
        assert len(sent_ids) >= 2
        for call in client.post_message.await_args_list:
            assert len(call.args[1]) <= ChattoConstants.MAX_MESSAGE_LENGTH

    async def test_failure_in_a_later_chunk_reports_failure(self):
        cfg = _make_config(token="test-token")
        client = MagicMock()
        first = MagicMock()
        first.id = "msg-1"
        client.post_message = AsyncMock(side_effect=[first, RuntimeError("boom")])
        client.close = AsyncMock()
        with patch("adapter.ChattoClient") as client_cls:
            client_cls.return_value = client
            result = await hermes_standalone_sender_fn(cfg, "room-1", "y" * 12000)

        assert result.success is False
        assert "boom" in (result.error or "")
        # The short-lived client is closed even on failure.
        client.close.assert_awaited_once()

    @pytest.mark.parametrize(
        ("error", "expected_retryable"),
        [
            (ChattoError("boom"), True),
            (RuntimeError("boom"), False),
        ],
    )
    async def test_failure_classification_matches_the_send_paths(
        self, error, expected_retryable
    ):
        """Server-side failures retry; our own bugs must not loop."""
        cfg = _make_config(token="test-token")
        client = MagicMock()
        client.post_message = AsyncMock(side_effect=error)
        client.close = AsyncMock()
        with patch("adapter.ChattoClient") as client_cls:
            client_cls.return_value = client
            result = await hermes_standalone_sender_fn(cfg, "room-1", "hello")

        assert result.success is False
        assert result.retryable is expected_retryable

    async def test_thread_id_reaches_every_chunk(self):
        cfg = _make_config(token="test-token")
        client, _ids = self._posted(5)
        with patch("adapter.ChattoClient") as client_cls:
            client_cls.return_value = client
            result = await hermes_standalone_sender_fn(
                cfg, "room-1", "z" * 12000, thread_id="root-9"
            )

        assert result.success is True
        for call in client.post_message.await_args_list:
            assert call.kwargs["thread_root_event_id"] == "root-9"