"""Tests for the native matrix_message tool."""

from __future__ import annotations

import asyncio
import json
import tempfile
from pathlib import Path
from typing import TYPE_CHECKING, ClassVar
from unittest.mock import ANY, AsyncMock, MagicMock, Mock, call, patch

import nio
import pytest

import mindroom.tools  # noqa: F401
from mindroom import interactive
from mindroom.attachments import register_local_attachment
from mindroom.config.agent import AgentConfig
from mindroom.config.main import Config
from mindroom.config.matrix import MindRoomUserConfig
from mindroom.constants import ORIGINAL_SENDER_KEY, SKIP_MENTIONS_KEY, SOURCE_KIND_KEY
from mindroom.custom_tools.attachments import AttachmentTools
from mindroom.custom_tools.matrix_message import MatrixMessageTools
from mindroom.dispatch_source import TRUSTED_INTERNAL_RELAY_SOURCE_KIND
from mindroom.interactive import parse_and_format_interactive
from mindroom.matrix.client import DeliveredMatrixEvent, RoomThreadsPageError
from mindroom.matrix.message_extras import MINDROOM_MESSAGE_EXTRAS_KEY
from mindroom.matrix.state import MatrixState, _load_matrix_state_file_cached
from mindroom.message_target import MessageTarget
from mindroom.session_ids import create_session_id
from mindroom.tool_system.metadata import TOOL_METADATA, get_tool_by_name
from mindroom.tool_system.runtime_context import ToolRuntimeContext, tool_runtime_context
from tests.conftest import (
    bind_runtime_paths,
    delivered_matrix_event,
    delivered_matrix_side_effect,
    make_conversation_cache_mock,
    make_event_cache_mock,
    make_matrix_client_mock,
    make_visible_message,
    runtime_paths_for,
    test_runtime_paths,
)

if TYPE_CHECKING:
    from collections.abc import AsyncIterator


_DEFAULT_RESOLVED_THREAD_ID = object()
_DEFAULT_EVENT_CACHE = object()


@pytest.fixture(autouse=True)
def _reset_matrix_message_rate_limit() -> None:
    MatrixMessageTools._recent_actions.clear()


def _empty_async_iterator() -> AsyncIterator[object]:
    async def iterator() -> AsyncIterator[object]:
        if False:
            yield None

    return iterator()


@pytest.fixture(autouse=True)
def _reset_interactive_state() -> None:
    interactive._active_questions.clear()
    interactive._persistence_file = None


def _make_context(
    *,
    room_id: str = "!room:localhost",
    thread_id: str | None = "$thread:localhost",
    requester_id: str = "@user:localhost",
    bot_accounts: list[str] | None = None,
    mindroom_user: MindRoomUserConfig | None = None,
    resolved_thread_id: object = _DEFAULT_RESOLVED_THREAD_ID,
    reply_to_event_id: str | None = "$reply:localhost",
    storage_path: Path | None = None,
    attachment_ids: tuple[str, ...] = (),
    agent_thread_mode: str = "thread",
    event_cache: object = _DEFAULT_EVENT_CACHE,
) -> ToolRuntimeContext:
    async def _latest_thread_event_id(
        _room_id: str,
        thread_id: str | None,
        *_args: object,
        **_kwargs: object,
    ) -> str | None:
        return thread_id

    runtime_root = storage_path or Path(tempfile.mkdtemp())
    config = bind_runtime_paths(
        Config(
            agents={
                "general": AgentConfig(
                    display_name="General Agent",
                    thread_mode=agent_thread_mode,
                ),
            },
            bot_accounts=bot_accounts or [],
            mindroom_user=mindroom_user,
        ),
        test_runtime_paths(runtime_root),
    )
    client = make_matrix_client_mock(user_id="@mindroom_general:localhost")
    client.room_send = AsyncMock()
    client.room_messages = AsyncMock()
    client.room_get_event_relations = MagicMock(
        side_effect=lambda *_args, **_kwargs: _empty_async_iterator(),
    )
    conversation_cache = make_conversation_cache_mock()
    conversation_cache.get_latest_thread_event_id_if_needed.side_effect = _latest_thread_event_id
    conversation_cache.notify_outbound_message = Mock()
    conversation_cache.notify_outbound_redaction = Mock()
    return ToolRuntimeContext(
        agent_name="general",
        target=MessageTarget(
            room_id=room_id,
            source_thread_id=thread_id,
            resolved_thread_id=thread_id if resolved_thread_id is _DEFAULT_RESOLVED_THREAD_ID else resolved_thread_id,
            reply_to_event_id=reply_to_event_id,
            session_id=create_session_id(
                room_id,
                thread_id if resolved_thread_id is _DEFAULT_RESOLVED_THREAD_ID else resolved_thread_id,
            ),
        ),
        requester_id=requester_id,
        client=client,
        config=config,
        runtime_paths=runtime_paths_for(config),
        conversation_cache=conversation_cache,
        event_cache=make_event_cache_mock() if event_cache is _DEFAULT_EVENT_CACHE else event_cache,
        room=None,
        storage_path=storage_path,
        attachment_ids=attachment_ids,
    )


def _make_room_thread_root(
    *,
    event_id: str,
    sender: str,
    timestamp: int,
    body: str | None = None,
    reply_count: int | None = None,
    encrypted: bool = False,
) -> MagicMock:
    event = MagicMock(spec=nio.MegolmEvent if encrypted else nio.RoomMessageText)
    event.event_id = event_id
    event.sender = sender
    event.server_timestamp = timestamp
    if body is not None:
        event.body = body

    source: dict[str, object] = {
        "event_id": event_id,
        "sender": sender,
        "origin_server_ts": timestamp,
    }
    if encrypted:
        source["type"] = "m.room.encrypted"
        source["content"] = {
            "algorithm": "m.megolm.v1.aes-sha2",
            "ciphertext": "ciphertext",
            "device_id": "DEVICE",
            "sender_key": "sender_key",
            "session_id": "session_id",
        }
    else:
        source["type"] = "m.room.message"
        source["content"] = {"msgtype": "m.text", "body": body or ""}
    if reply_count is not None:
        source["unsigned"] = {"m.relations": {"m.thread": {"count": reply_count}}}
    event.source = source
    return event


def _make_bundled_replacement(
    *,
    event_id: str,
    body: str,
    msgtype: str,
    bundle_key: str | None = None,
    sender: str = "@editor:localhost",
    visible_body: str | None = None,
    long_text: dict[str, object] | None = None,
    url: str | None = None,
) -> dict[str, object]:
    new_content: dict[str, object] = {
        "body": body,
        "msgtype": msgtype,
    }
    if visible_body is not None:
        new_content["io.mindroom.visible_body"] = visible_body
    if long_text is not None:
        new_content["io.mindroom.long_text"] = long_text
    if url is not None:
        new_content["url"] = url

    replacement_event = {
        "type": "m.room.message",
        "event_id": f"{event_id}-edit",
        "sender": sender,
        "origin_server_ts": 9999,
        "content": {
            "body": f"* {body}",
            "msgtype": "m.text",
            "m.new_content": new_content,
            "m.relates_to": {
                "rel_type": "m.replace",
                "event_id": event_id,
            },
        },
    }
    if bundle_key is None:
        return replacement_event
    return {bundle_key: replacement_event}


def test_matrix_message_tool_registered_and_instantiates() -> None:
    """Matrix message tool should be available from metadata registry."""
    config = bind_runtime_paths(
        Config(agents={"general": AgentConfig(display_name="General Agent")}),
        test_runtime_paths(Path(tempfile.mkdtemp())),
    )
    assert "matrix_message" in TOOL_METADATA
    assert isinstance(
        get_tool_by_name("matrix_message", runtime_paths_for(config), worker_target=None),
        MatrixMessageTools,
    )


@pytest.mark.asyncio
async def test_matrix_message_requires_runtime_context() -> None:
    """Tool should fail clearly when called without Matrix runtime context."""
    payload = json.loads(await MatrixMessageTools().matrix_message(action="send", message="hello"))
    assert payload["status"] == "error"
    assert payload["tool"] == "matrix_message"
    assert "context" in payload["message"]


@pytest.mark.asyncio
async def test_matrix_message_send_defaults_to_room_level() -> None:
    """Send action should stay room-level unless a thread is explicitly passed."""
    tool = MatrixMessageTools()
    ctx = _make_context(thread_id="$ctx-thread:localhost")

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.send_message_result",
            new=AsyncMock(side_effect=delivered_matrix_side_effect("$evt")),
        ) as mock_send,
        tool_runtime_context(ctx),
    ):
        payload = json.loads(await tool.matrix_message(action="send", message="hello"))

    assert payload["status"] == "ok"
    assert payload["action"] == "send"
    assert payload["thread_id"] is None
    assert payload["attachment_thread_id"] is None
    sent_content = mock_send.await_args.args[2]
    assert sent_content["body"] == "hello"
    assert "m.relates_to" not in sent_content
    assert sent_content[SKIP_MENTIONS_KEY] is True
    assert ORIGINAL_SENDER_KEY not in sent_content
    assert SOURCE_KIND_KEY not in sent_content


@pytest.mark.asyncio
async def test_matrix_message_active_mentions_mark_trusted_human_relay() -> None:
    """Intentional mention dispatch should preserve a trusted human requester."""
    tool = MatrixMessageTools()
    ctx = _make_context(thread_id=None)

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.send_message_result",
            new=AsyncMock(side_effect=delivered_matrix_side_effect("$evt")),
        ) as mock_send,
        tool_runtime_context(ctx),
    ):
        payload = json.loads(
            await tool.matrix_message(
                action="send",
                message="@general continue work",
                ignore_mentions=False,
            ),
        )

    assert payload["status"] == "ok"
    sent_content = mock_send.await_args.args[2]
    assert sent_content["m.mentions"] == {"user_ids": [ctx.client.user_id]}
    assert SKIP_MENTIONS_KEY not in sent_content
    assert sent_content[ORIGINAL_SENDER_KEY] == ctx.requester_id
    assert sent_content[SOURCE_KIND_KEY] == TRUSTED_INTERNAL_RELAY_SOURCE_KIND


@pytest.mark.asyncio
async def test_matrix_message_active_mentions_do_not_promote_managed_requester() -> None:
    """Intentional mention dispatch should not classify managed requesters as humans."""
    tool = MatrixMessageTools()
    ctx = _make_context(
        thread_id=None,
        requester_id="@mindroom_router:localhost",
    )

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.send_message_result",
            new=AsyncMock(side_effect=delivered_matrix_side_effect("$evt")),
        ) as mock_send,
        tool_runtime_context(ctx),
    ):
        payload = json.loads(
            await tool.matrix_message(
                action="send",
                message="@general continue work",
                ignore_mentions=False,
            ),
        )

    assert payload["status"] == "ok"
    sent_content = mock_send.await_args.args[2]
    assert sent_content["m.mentions"] == {"user_ids": [ctx.client.user_id]}
    assert SKIP_MENTIONS_KEY not in sent_content
    assert ORIGINAL_SENDER_KEY not in sent_content
    assert SOURCE_KIND_KEY not in sent_content


@pytest.mark.parametrize(
    ("requester_id", "bot_accounts", "mindroom_user"),
    [
        ("@bridge_bot:localhost", ["@bridge_bot:localhost"], None),
        ("@mindroom_user:localhost", [], MindRoomUserConfig()),
    ],
)
@pytest.mark.asyncio
async def test_matrix_message_active_mentions_do_not_promote_non_human_requester(
    requester_id: str,
    bot_accounts: list[str],
    mindroom_user: MindRoomUserConfig | None,
) -> None:
    """Trusted relay provenance should require a human requester."""
    tool = MatrixMessageTools()
    ctx = _make_context(
        thread_id=None,
        requester_id=requester_id,
        bot_accounts=bot_accounts,
        mindroom_user=mindroom_user,
    )

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.send_message_result",
            new=AsyncMock(side_effect=delivered_matrix_side_effect("$evt")),
        ) as mock_send,
        tool_runtime_context(ctx),
    ):
        payload = json.loads(
            await tool.matrix_message(
                action="send",
                message="@general continue work",
                ignore_mentions=False,
            ),
        )

    assert payload["status"] == "ok"
    sent_content = mock_send.await_args.args[2]
    assert sent_content["m.mentions"] == {"user_ids": [ctx.client.user_id]}
    assert SKIP_MENTIONS_KEY not in sent_content
    assert ORIGINAL_SENDER_KEY not in sent_content
    assert SOURCE_KIND_KEY not in sent_content


@pytest.mark.asyncio
async def test_matrix_message_send_resolves_room_alias_before_send(tmp_path: Path) -> None:
    """Explicit room aliases should resolve to room IDs before authorization and delivery."""
    tool = MatrixMessageTools()
    ctx = _make_context(storage_path=tmp_path, thread_id=None)
    state = MatrixState()
    state.add_room("ops", room_id="!ops:localhost", alias="#ops:localhost", name="Ops")
    state.save(runtime_paths=ctx.runtime_paths)
    _load_matrix_state_file_cached.cache_clear()

    with (
        patch("mindroom.custom_tools.matrix_message.room_access_allowed", return_value=True) as mock_access,
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.send_message_result",
            new=AsyncMock(side_effect=delivered_matrix_side_effect("$evt")),
        ) as mock_send,
        tool_runtime_context(ctx),
    ):
        payload = json.loads(await tool.matrix_message(action="send", message="hello", room_id="#ops:localhost"))

    mock_access.assert_called_once_with(ctx, "!ops:localhost")
    assert mock_send.await_args.args[1] == "!ops:localhost"
    assert payload["status"] == "ok"
    assert payload["room_id"] == "!ops:localhost"


@pytest.mark.asyncio
async def test_matrix_message_rejects_non_string_room_id_before_resolution(tmp_path: Path) -> None:
    """Explicit room IDs should return structured type errors before alias resolution."""
    tool = MatrixMessageTools()
    ctx = _make_context(storage_path=tmp_path, thread_id=None)

    with (
        patch("mindroom.custom_tools.matrix_message.resolve_optional_room_id") as mock_resolve,
        tool_runtime_context(ctx),
    ):
        payload = json.loads(
            await tool.matrix_message(action="send", message="hello", room_id=123),  # type: ignore[arg-type]
        )

    mock_resolve.assert_not_called()
    assert payload["status"] == "error"
    assert payload["message"] == "room_id must be a string."


@pytest.mark.asyncio
async def test_matrix_message_send_includes_message_extras() -> None:
    """Send action should attach validated MindRoom message extras."""
    tool = MatrixMessageTools()
    ctx = _make_context(thread_id="$ctx-thread:localhost")

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.send_message_result",
            new=AsyncMock(side_effect=delivered_matrix_side_effect("$evt")),
        ) as mock_send,
        tool_runtime_context(ctx),
    ):
        payload = json.loads(
            await tool.matrix_message(
                action="send",
                message="Short answer.",
                message_extras=[
                    {
                        "title": "Evidence",
                        "content_type": "text/html",
                        "content": "<table><tr><td>42</td></tr></table>",
                        "collapsed": False,
                    },
                ],
            ),
        )

    assert payload["status"] == "ok"
    sent_content = mock_send.await_args.args[2]
    assert sent_content["body"] == "Short answer."
    assert sent_content[MINDROOM_MESSAGE_EXTRAS_KEY] == {
        "version": 2,
        "sections": [
            {
                "title": "Evidence",
                "content_type": "text/html",
                "content": "<table><tr><td>42</td></tr></table>",
                "collapsed": False,
            },
        ],
    }


@pytest.mark.asyncio
async def test_matrix_message_send_rejects_message_extras_without_text_event() -> None:
    """Extras should not be silently dropped on attachment-only sends."""
    tool = MatrixMessageTools()
    ctx = _make_context(thread_id="$ctx-thread:localhost")

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.send_message_result",
            new=AsyncMock(side_effect=delivered_matrix_side_effect("$evt")),
        ) as mock_send,
        tool_runtime_context(ctx),
    ):
        payload = json.loads(
            await tool.matrix_message(
                action="send",
                attachment_ids=["att_context_file"],
                message_extras=[
                    {
                        "title": "Evidence",
                        "content": "details",
                    },
                ],
            ),
        )

    assert payload["status"] == "error"
    assert "non-empty message" in payload["message"]
    mock_send.assert_not_awaited()


@pytest.mark.asyncio
async def test_matrix_message_rejects_invalid_message_extras() -> None:
    """Invalid extras should return a tool error instead of sending."""
    tool = MatrixMessageTools()
    ctx = _make_context(thread_id="$ctx-thread:localhost")

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.send_message_result",
            new=AsyncMock(side_effect=delivered_matrix_side_effect("$evt")),
        ) as mock_send,
        tool_runtime_context(ctx),
    ):
        payload = json.loads(
            await tool.matrix_message(
                action="send",
                message="Short answer.",
                message_extras=[
                    {
                        "title": "Raw",
                        "content_type": "application/json",
                        "content": "{}",
                    },
                ],
            ),
        )

    assert payload["status"] == "error"
    assert "content_type" in payload["message"]
    mock_send.assert_not_awaited()


@pytest.mark.asyncio
async def test_matrix_message_send_room_sentinel_stays_room_level() -> None:
    """thread_id='room' should disable thread metadata for sends."""
    tool = MatrixMessageTools()
    event_cache = MagicMock()
    ctx = _make_context(thread_id="$ctx-thread:localhost", event_cache=event_cache)
    ctx.conversation_cache.get_latest_thread_event_id_if_needed = AsyncMock(return_value=None)

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.send_message_result",
            new=AsyncMock(side_effect=delivered_matrix_side_effect("$evt")),
        ) as mock_send,
        tool_runtime_context(ctx),
    ):
        payload = json.loads(
            await tool.matrix_message(action="send", thread_id="room", message="hello"),
        )

    assert payload["status"] == "ok"
    assert payload["action"] == "send"
    assert payload["room_id"] == ctx.room_id
    assert payload["thread_id"] is None
    assert payload["event_id"] == "$evt"
    ctx.conversation_cache.get_latest_thread_event_id_if_needed.assert_awaited_once_with(
        ctx.room_id,
        None,
        caller_label="matrix_message_tool_send",
    )
    sent_content = mock_send.await_args.args[2]
    assert sent_content["body"] == "hello"
    assert "m.relates_to" not in sent_content


@pytest.mark.asyncio
async def test_matrix_message_send_interactive_block_registers_question_and_adds_reactions() -> None:
    """Interactive sends should format the question and add reaction buttons."""
    tool = MatrixMessageTools()
    ctx = _make_context(thread_id="$ctx-thread:localhost")
    interactive_message = """Please choose.

```interactive
{
  "question": "Which option?",
  "options": [
    {"emoji": "✅", "label": "Approve", "value": "approve"},
    {"emoji": "❌", "label": "Reject", "value": "reject"}
  ]
}
```"""
    formatted_text = parse_and_format_interactive(interactive_message, extract_mapping=False).formatted_text

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.send_message_result",
            new=AsyncMock(side_effect=delivered_matrix_side_effect("$evt")),
        ) as mock_send,
        patch("mindroom.custom_tools.matrix_conversation_operations.register_interactive_question") as mock_register,
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.add_reaction_buttons",
            new_callable=AsyncMock,
        ) as mock_add_reactions,
        tool_runtime_context(ctx),
    ):
        payload = json.loads(await tool.matrix_message(action="send", message=interactive_message))

    assert payload["status"] == "ok"
    assert payload["event_id"] == "$evt"
    sent_content = mock_send.await_args.args[2]
    assert sent_content["body"] == formatted_text
    mock_register.assert_called_once_with(
        "$evt",
        ctx.room_id,
        None,
        {
            "✅": "approve",
            "1": "approve",
            "❌": "reject",
            "2": "reject",
        },
        ctx.agent_name,
        question_text="Which option?",
        option_labels={
            "✅": "Approve",
            "1": "Approve",
            "❌": "Reject",
            "2": "Reject",
        },
    )
    mock_add_reactions.assert_awaited_once_with(
        ctx.client,
        ctx.room_id,
        "$evt",
        [
            {"emoji": "✅", "label": "Approve", "value": "approve"},
            {"emoji": "❌", "label": "Reject", "value": "reject"},
        ],
    )


@pytest.mark.asyncio
async def test_matrix_message_send_plain_text_skips_interactive_registration_and_reactions() -> None:
    """Plain-text sends should not register interactive state or add reactions."""
    tool = MatrixMessageTools()
    ctx = _make_context()

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.send_message_result",
            new=AsyncMock(side_effect=delivered_matrix_side_effect("$evt")),
        ),
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.parse_and_format_interactive",
            wraps=parse_and_format_interactive,
        ) as mock_parse,
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.should_create_interactive_question",
            return_value=False,
        ) as mock_should_create,
        patch("mindroom.custom_tools.matrix_conversation_operations.register_interactive_question") as mock_register,
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.add_reaction_buttons",
            new_callable=AsyncMock,
        ) as mock_add_reactions,
        tool_runtime_context(ctx),
    ):
        payload = json.loads(await tool.matrix_message(action="send", message="hello"))

    assert payload["status"] == "ok"
    mock_parse.assert_called_once_with("hello", extract_mapping=False)
    mock_should_create.assert_called_once_with("hello")
    mock_register.assert_not_called()
    mock_add_reactions.assert_not_awaited()


@pytest.mark.asyncio
async def test_matrix_message_send_supports_context_attachments(tmp_path: Path) -> None:
    """Send should accept context att_* IDs and upload them after text."""
    tool = MatrixMessageTools()
    sample_file = tmp_path / "upload.txt"
    sample_file.write_text("payload", encoding="utf-8")
    attachment = register_local_attachment(
        tmp_path,
        sample_file,
        kind="file",
        attachment_id="att_upload",
    )
    assert attachment is not None
    ctx = _make_context(storage_path=tmp_path, attachment_ids=("att_upload",))
    ctx.conversation_cache.get_latest_thread_event_id_if_needed.return_value = "$evt"

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.send_message_result",
            new=AsyncMock(side_effect=delivered_matrix_side_effect("$evt")),
        ) as mock_send,
        patch(
            "mindroom.custom_tools.attachments.send_file_message",
            new=AsyncMock(return_value="$file_evt"),
        ) as mock_send_file,
        tool_runtime_context(ctx),
    ):
        payload = json.loads(
            await tool.matrix_message(
                action="send",
                message="hello",
                attachment_ids=["att_upload"],
            ),
        )

    assert payload["status"] == "ok"
    assert payload["event_id"] == "$evt"
    assert payload["thread_id"] is None
    assert payload["attachment_thread_id"] == "$evt"
    assert payload["attachment_event_ids"] == ["$file_evt"]
    assert payload["resolved_attachment_ids"] == ["att_upload"]
    ctx.conversation_cache.get_latest_thread_event_id_if_needed.assert_has_awaits(
        [
            call(ctx.room_id, None, caller_label="matrix_message_tool_send"),
            call(ctx.room_id, "$evt", caller_label="attachment_tool_send"),
        ],
    )
    mock_send.assert_awaited_once()
    mock_send_file.assert_awaited_once_with(
        ctx.client,
        ctx.room_id,
        attachment.local_path,
        thread_id="$evt",
        latest_thread_event_id="$evt",
        conversation_cache=ctx.conversation_cache,
    )


@pytest.mark.asyncio
async def test_matrix_message_send_with_attachment_in_room_mode_stays_room_level(tmp_path: Path) -> None:
    """Room-mode sends should not auto-thread attachments under the new text event."""
    tool = MatrixMessageTools()
    sample_file = tmp_path / "upload.txt"
    sample_file.write_text("payload", encoding="utf-8")
    attachment = register_local_attachment(
        tmp_path,
        sample_file,
        kind="file",
        attachment_id="att_room_mode",
    )
    assert attachment is not None
    ctx = _make_context(
        storage_path=tmp_path,
        attachment_ids=("att_room_mode",),
        thread_id="$ctx-thread:localhost",
        resolved_thread_id=None,
        agent_thread_mode="room",
    )

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.send_message_result",
            new=AsyncMock(side_effect=delivered_matrix_side_effect("$evt")),
        ) as mock_send,
        patch(
            "mindroom.custom_tools.attachments.send_file_message",
            new=AsyncMock(return_value="$file_evt"),
        ) as mock_send_file,
        tool_runtime_context(ctx),
    ):
        payload = json.loads(
            await tool.matrix_message(
                action="send",
                message="hello",
                attachment_ids=["att_room_mode"],
            ),
        )

    assert payload["status"] == "ok"
    assert payload["event_id"] == "$evt"
    assert payload["thread_id"] is None
    assert payload["attachment_thread_id"] is None
    assert payload["attachment_event_ids"] == ["$file_evt"]
    mock_send.assert_awaited_once()
    mock_send_file.assert_awaited_once_with(
        ctx.client,
        ctx.room_id,
        attachment.local_path,
        thread_id=None,
        latest_thread_event_id=None,
        conversation_cache=ctx.conversation_cache,
    )


@pytest.mark.asyncio
async def test_matrix_message_reply_with_attachments_keeps_existing_thread(tmp_path: Path) -> None:
    """Reply attachments should stay in the existing thread instead of using the text event as a new root."""
    tool = MatrixMessageTools()
    sample_file = tmp_path / "upload.txt"
    sample_file.write_text("payload", encoding="utf-8")
    attachment = register_local_attachment(
        tmp_path,
        sample_file,
        kind="file",
        attachment_id="att_reply",
    )
    assert attachment is not None
    ctx = _make_context(storage_path=tmp_path, attachment_ids=("att_reply",))

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.send_message_result",
            new=AsyncMock(side_effect=delivered_matrix_side_effect("$reply_evt")),
        ) as mock_send,
        patch(
            "mindroom.custom_tools.attachments.send_file_message",
            new=AsyncMock(return_value="$file_evt"),
        ) as mock_send_file,
        tool_runtime_context(ctx),
    ):
        payload = json.loads(
            await tool.matrix_message(
                action="reply",
                message="hello",
                attachment_ids=["att_reply"],
            ),
        )

    assert payload["status"] == "ok"
    assert payload["event_id"] == "$reply_evt"
    assert payload["thread_id"] == ctx.thread_id
    assert payload["attachment_thread_id"] == ctx.thread_id
    assert payload["attachment_event_ids"] == ["$file_evt"]
    mock_send.assert_awaited_once()
    mock_send_file.assert_awaited_once_with(
        ctx.client,
        ctx.room_id,
        attachment.local_path,
        thread_id=ctx.thread_id,
        latest_thread_event_id=ctx.thread_id,
        conversation_cache=ctx.conversation_cache,
    )


@pytest.mark.asyncio
async def test_matrix_message_send_with_explicit_thread_and_attachments_keeps_existing_thread(
    tmp_path: Path,
) -> None:
    """Send attachments should stay in the explicit thread instead of auto-threading under the new text event."""
    tool = MatrixMessageTools()
    sample_file = tmp_path / "upload.txt"
    sample_file.write_text("payload", encoding="utf-8")
    attachment = register_local_attachment(
        tmp_path,
        sample_file,
        kind="file",
        attachment_id="att_explicit_thread",
    )
    assert attachment is not None
    ctx = _make_context(storage_path=tmp_path, attachment_ids=("att_explicit_thread",), thread_id=None)
    explicit_thread_id = "$explicit-thread:localhost"

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.send_message_result",
            new=AsyncMock(side_effect=delivered_matrix_side_effect("$send_evt")),
        ) as mock_send,
        patch(
            "mindroom.custom_tools.attachments.send_file_message",
            new=AsyncMock(return_value="$file_evt"),
        ) as mock_send_file,
        tool_runtime_context(ctx),
    ):
        payload = json.loads(
            await tool.matrix_message(
                action="send",
                message="hello",
                thread_id=explicit_thread_id,
                attachment_ids=["att_explicit_thread"],
            ),
        )

    assert payload["status"] == "ok"
    assert payload["event_id"] == "$send_evt"
    assert payload["thread_id"] == explicit_thread_id
    assert payload["attachment_thread_id"] == explicit_thread_id
    assert payload["attachment_event_ids"] == ["$file_evt"]
    assert payload["resolved_attachment_ids"] == ["att_explicit_thread"]
    mock_send.assert_awaited_once()
    sent_content = mock_send.await_args.args[2]
    relates_to = sent_content.get("m.relates_to", {})
    assert relates_to.get("event_id") == explicit_thread_id
    mock_send_file.assert_awaited_once_with(
        ctx.client,
        ctx.room_id,
        attachment.local_path,
        thread_id=explicit_thread_id,
        latest_thread_event_id=explicit_thread_id,
        conversation_cache=ctx.conversation_cache,
    )


@pytest.mark.asyncio
async def test_matrix_message_send_allows_attachment_only(tmp_path: Path) -> None:
    """Send should allow attachments without a text body."""
    tool = MatrixMessageTools()
    sample_file = tmp_path / "upload.txt"
    sample_file.write_text("payload", encoding="utf-8")
    attachment = register_local_attachment(
        tmp_path,
        sample_file,
        kind="file",
        attachment_id="att_only",
    )
    assert attachment is not None
    ctx = _make_context(storage_path=tmp_path, attachment_ids=("att_only",))

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.send_message_result",
            new=AsyncMock(side_effect=delivered_matrix_side_effect("$evt")),
        ) as mock_send,
        patch(
            "mindroom.custom_tools.attachments.send_file_message",
            new=AsyncMock(return_value="$file_evt"),
        ) as mock_send_file,
        tool_runtime_context(ctx),
    ):
        payload = json.loads(
            await tool.matrix_message(
                action="send",
                attachment_ids=["att_only"],
            ),
        )

    assert payload["status"] == "ok"
    assert payload["event_id"] is None
    assert payload["thread_id"] is None
    assert payload["attachment_thread_id"] is None
    assert payload["attachment_event_ids"] == ["$file_evt"]
    assert payload["resolved_attachment_ids"] == ["att_only"]
    mock_send.assert_not_awaited()
    mock_send_file.assert_awaited_once_with(
        ctx.client,
        ctx.room_id,
        attachment.local_path,
        thread_id=None,
        latest_thread_event_id=None,
        conversation_cache=ctx.conversation_cache,
    )


@pytest.mark.asyncio
async def test_matrix_message_send_multiple_attachments_only_auto_threads_under_first_attachment(
    tmp_path: Path,
) -> None:
    """Attachment-only sends should use the first room-level attachment as the thread root for the rest."""
    tool = MatrixMessageTools()
    first_file = tmp_path / "first.txt"
    second_file = tmp_path / "second.txt"
    first_file.write_text("first", encoding="utf-8")
    second_file.write_text("second", encoding="utf-8")
    first_attachment = register_local_attachment(
        tmp_path,
        first_file,
        kind="file",
        attachment_id="att_first",
    )
    second_attachment = register_local_attachment(
        tmp_path,
        second_file,
        kind="file",
        attachment_id="att_second",
    )
    assert first_attachment is not None
    assert second_attachment is not None
    ctx = _make_context(storage_path=tmp_path, attachment_ids=("att_first", "att_second"), thread_id=None)

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.send_resolved_attachments",
            new=AsyncMock(side_effect=[(["$file_root"], None), (["$file_child"], None)]),
        ) as mock_send_attachments,
        tool_runtime_context(ctx),
    ):
        payload = json.loads(
            await tool.matrix_message(
                action="send",
                attachment_ids=["att_first", "att_second"],
            ),
        )

    assert payload["status"] == "ok"
    assert payload["event_id"] is None
    assert payload["thread_id"] is None
    assert payload["attachment_thread_id"] == "$file_root"
    assert payload["attachment_event_ids"] == ["$file_root", "$file_child"]
    assert payload["resolved_attachment_ids"] == ["att_first", "att_second"]
    assert mock_send_attachments.await_args_list[0].args == (ctx,)
    assert mock_send_attachments.await_args_list[0].kwargs == {
        "room_id": ctx.room_id,
        "thread_id": None,
        "attachments": [first_attachment.local_path],
    }
    assert mock_send_attachments.await_args_list[1].args == (ctx,)
    assert mock_send_attachments.await_args_list[1].kwargs == {
        "room_id": ctx.room_id,
        "thread_id": "$file_root",
        "attachments": [second_attachment.local_path],
    }


@pytest.mark.asyncio
async def test_matrix_message_send_multiple_attachments_only_in_room_mode_stays_room_level(
    tmp_path: Path,
) -> None:
    """Room-mode sends should not create an attachment thread when sending multiple files."""
    tool = MatrixMessageTools()
    first_file = tmp_path / "first.txt"
    second_file = tmp_path / "second.txt"
    first_file.write_text("first", encoding="utf-8")
    second_file.write_text("second", encoding="utf-8")
    first_attachment = register_local_attachment(
        tmp_path,
        first_file,
        kind="file",
        attachment_id="att_room_first",
    )
    second_attachment = register_local_attachment(
        tmp_path,
        second_file,
        kind="file",
        attachment_id="att_room_second",
    )
    assert first_attachment is not None
    assert second_attachment is not None
    ctx = _make_context(
        storage_path=tmp_path,
        attachment_ids=("att_room_first", "att_room_second"),
        thread_id="$ctx-thread:localhost",
        resolved_thread_id=None,
        agent_thread_mode="room",
    )

    with (
        patch(
            "mindroom.custom_tools.attachments.send_file_message",
            new=AsyncMock(side_effect=["$file_one", "$file_two"]),
        ) as mock_send_file,
        tool_runtime_context(ctx),
    ):
        payload = json.loads(
            await tool.matrix_message(
                action="send",
                attachment_ids=["att_room_first", "att_room_second"],
            ),
        )

    assert payload["status"] == "ok"
    assert payload["event_id"] is None
    assert payload["thread_id"] is None
    assert payload["attachment_thread_id"] is None
    assert payload["attachment_event_ids"] == ["$file_one", "$file_two"]
    assert payload["resolved_attachment_ids"] == ["att_room_first", "att_room_second"]
    assert len(mock_send_file.await_args_list) == 2
    first_call = mock_send_file.await_args_list[0]
    second_call = mock_send_file.await_args_list[1]
    assert first_call.args == (ctx.client, ctx.room_id, first_attachment.local_path)
    assert first_call.kwargs == {
        "thread_id": None,
        "latest_thread_event_id": None,
        "conversation_cache": ctx.conversation_cache,
    }
    assert second_call.args == (ctx.client, ctx.room_id, second_attachment.local_path)
    assert second_call.kwargs == {
        "thread_id": None,
        "latest_thread_event_id": "$file_one",
        "conversation_cache": ctx.conversation_cache,
    }


@pytest.mark.asyncio
async def test_matrix_message_send_supports_attachment_file_paths(tmp_path: Path) -> None:
    """Send should auto-register local file paths and upload them."""
    tool = MatrixMessageTools()
    generated_file = tmp_path / "generated.txt"
    generated_file.write_text("artifact", encoding="utf-8")
    ctx = _make_context(storage_path=tmp_path)

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.send_message_result",
            new=AsyncMock(side_effect=delivered_matrix_side_effect("$evt")),
        ) as mock_send,
        patch(
            "mindroom.custom_tools.attachments.send_file_message",
            new=AsyncMock(return_value="$file_evt"),
        ) as mock_send_file,
        tool_runtime_context(ctx),
    ):
        payload = json.loads(
            await tool.matrix_message(
                action="send",
                message="hello",
                attachment_file_paths=[str(generated_file)],
            ),
        )

    assert payload["status"] == "ok"
    assert payload["event_id"] == "$evt"
    assert payload["thread_id"] is None
    assert payload["attachment_thread_id"] == "$evt"
    assert payload["attachment_event_ids"] == ["$file_evt"]
    assert payload["resolved_attachment_ids"][0].startswith("att_")
    assert payload["newly_registered_attachment_ids"] == payload["resolved_attachment_ids"]
    mock_send.assert_awaited_once()
    mock_send_file.assert_awaited_once_with(
        ctx.client,
        ctx.room_id,
        generated_file,
        thread_id="$evt",
        latest_thread_event_id="$evt",
        conversation_cache=ctx.conversation_cache,
    )


@pytest.mark.asyncio
async def test_matrix_message_send_resolves_relative_attachment_file_paths_from_workspace(tmp_path: Path) -> None:
    """Relative attachment_file_paths should resolve from the agent workspace root."""
    workspace_root = tmp_path / "workspace"
    workspace_root.mkdir()
    generated_file = workspace_root / "scratch" / "generated.txt"
    generated_file.parent.mkdir()
    generated_file.write_text("artifact", encoding="utf-8")
    tool = MatrixMessageTools(tool_output_workspace_root=workspace_root)
    ctx = _make_context(storage_path=tmp_path)

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.send_message_result",
            new=AsyncMock(side_effect=delivered_matrix_side_effect("$evt")),
        ),
        patch(
            "mindroom.custom_tools.attachments.send_file_message",
            new=AsyncMock(return_value="$file_evt"),
        ) as mock_send_file,
        tool_runtime_context(ctx),
    ):
        payload = json.loads(
            await tool.matrix_message(
                action="send",
                message="hello",
                attachment_file_paths=["scratch/generated.txt"],
            ),
        )

    assert payload["status"] == "ok"
    assert payload["attachment_event_ids"] == ["$file_evt"]
    mock_send_file.assert_awaited_once_with(
        ctx.client,
        ctx.room_id,
        generated_file.resolve(),
        thread_id="$evt",
        latest_thread_event_id="$evt",
        conversation_cache=ctx.conversation_cache,
    )


@pytest.mark.asyncio
async def test_matrix_message_send_text_failure_does_not_attempt_attachments(tmp_path: Path) -> None:
    """Attachment sends should not start when the text send fails."""
    tool = MatrixMessageTools()
    sample_file = tmp_path / "upload.txt"
    sample_file.write_text("payload", encoding="utf-8")
    attachment = register_local_attachment(
        tmp_path,
        sample_file,
        kind="file",
        attachment_id="att_text_fail",
    )
    assert attachment is not None
    ctx = _make_context(storage_path=tmp_path, attachment_ids=("att_text_fail",))

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.send_message_result",
            new=AsyncMock(return_value=None),
        ) as mock_send,
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.send_context_attachments",
            new=AsyncMock(),
        ) as mock_send_context_attachments,
        tool_runtime_context(ctx),
    ):
        payload = json.loads(
            await tool.matrix_message(
                action="send",
                message="hello",
                attachment_ids=["att_text_fail"],
            ),
        )

    assert payload["status"] == "error"
    assert payload["action"] == "send"
    assert payload["message"] == "Failed to send message to Matrix."
    mock_send.assert_awaited_once()
    mock_send_context_attachments.assert_not_awaited()


@pytest.mark.asyncio
async def test_matrix_message_send_multiple_attachments_only_returns_error_when_first_send_fails(
    tmp_path: Path,
) -> None:
    """Attachment-only auto-threading should stop immediately if the root attachment fails to send."""
    tool = MatrixMessageTools()
    first_file = tmp_path / "first.txt"
    second_file = tmp_path / "second.txt"
    first_file.write_text("first", encoding="utf-8")
    second_file.write_text("second", encoding="utf-8")
    first_attachment = register_local_attachment(
        tmp_path,
        first_file,
        kind="file",
        attachment_id="att_first_fail",
    )
    second_attachment = register_local_attachment(
        tmp_path,
        second_file,
        kind="file",
        attachment_id="att_second_fail",
    )
    assert first_attachment is not None
    assert second_attachment is not None
    ctx = _make_context(
        storage_path=tmp_path,
        attachment_ids=("att_first_fail", "att_second_fail"),
        thread_id=None,
    )

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.send_resolved_attachments",
            new=AsyncMock(return_value=([], "Failed to send attachment: first")),
        ) as mock_send_attachments,
        tool_runtime_context(ctx),
    ):
        payload = json.loads(
            await tool.matrix_message(
                action="send",
                attachment_ids=["att_first_fail", "att_second_fail"],
            ),
        )

    assert payload["status"] == "error"
    assert payload["event_id"] is None
    assert payload["thread_id"] is None
    assert payload["attachment_thread_id"] is None
    assert payload["attachment_event_ids"] == []
    assert payload["resolved_attachment_ids"] == ["att_first_fail", "att_second_fail"]
    assert payload["newly_registered_attachment_ids"] == []
    assert "Failed to send attachment" in payload["message"]
    mock_send_attachments.assert_awaited_once_with(
        ctx,
        room_id=ctx.room_id,
        thread_id=None,
        attachments=[first_attachment.local_path],
    )


@pytest.mark.asyncio
async def test_matrix_message_accepts_register_attachment_ids_across_task_boundaries(tmp_path: Path) -> None:
    """matrix_message should accept attachment IDs registered by a prior tool call in another task."""
    matrix_tool = MatrixMessageTools()
    attachment_tool = AttachmentTools()
    generated_file = tmp_path / "generated.txt"
    generated_file.write_text("artifact", encoding="utf-8")
    ctx = _make_context(storage_path=tmp_path)
    ctx.conversation_cache.get_latest_thread_event_id_if_needed = AsyncMock(return_value=ctx.thread_id)

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.send_message_result",
            new=AsyncMock(side_effect=delivered_matrix_side_effect("$evt")),
        ) as mock_send,
        patch(
            "mindroom.custom_tools.attachments.send_file_message",
            new=AsyncMock(return_value="$file_evt"),
        ) as mock_send_file,
        tool_runtime_context(ctx),
    ):
        register_payload = json.loads(
            await asyncio.create_task(attachment_tool.register_attachment(str(generated_file))),
        )
        attachment_id = register_payload["attachment_id"]
        payload = json.loads(
            await asyncio.create_task(
                matrix_tool.matrix_message(
                    action="thread-reply",
                    message="hello",
                    attachment_ids=[attachment_id],
                ),
            ),
        )

    assert register_payload["status"] == "ok"
    assert payload["status"] == "ok"
    assert payload["event_id"] == "$evt"
    assert payload["attachment_event_ids"] == ["$file_evt"]
    assert payload["resolved_attachment_ids"] == [attachment_id]
    mock_send.assert_awaited_once()
    mock_send_file.assert_awaited_once()


@pytest.mark.asyncio
async def test_matrix_message_reply_defaults_to_context_thread() -> None:
    """Reply action should use current runtime thread when thread_id is omitted."""
    tool = MatrixMessageTools()
    ctx = _make_context(thread_id="$ctx-thread:localhost")

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.send_message_result",
            new=AsyncMock(side_effect=delivered_matrix_side_effect("$evt")),
        ) as mock_send,
        tool_runtime_context(ctx),
    ):
        payload = json.loads(await tool.matrix_message(action="reply", message="hello"))

    assert payload["status"] == "ok"
    assert payload["thread_id"] == "$ctx-thread:localhost"
    sent_content = mock_send.await_args.args[2]
    relates_to = sent_content.get("m.relates_to", {})
    assert relates_to.get("event_id") == "$ctx-thread:localhost"


@pytest.mark.asyncio
async def test_matrix_message_thread_reply_defaults_to_context_thread() -> None:
    """thread-reply action should use current runtime thread when thread_id is omitted."""
    tool = MatrixMessageTools()
    ctx = _make_context(thread_id="$ctx-thread:localhost")

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.send_message_result",
            new=AsyncMock(side_effect=delivered_matrix_side_effect("$evt")),
        ) as mock_send,
        tool_runtime_context(ctx),
    ):
        payload = json.loads(await tool.matrix_message(action="thread-reply", message="hello"))

    assert payload["status"] == "ok"
    assert payload["thread_id"] == "$ctx-thread:localhost"
    sent_content = mock_send.await_args.args[2]
    relates_to = sent_content.get("m.relates_to", {})
    assert relates_to.get("event_id") == "$ctx-thread:localhost"


@pytest.mark.asyncio
async def test_matrix_message_react_happy_path() -> None:
    """React action should send a Matrix annotation event to the target event."""
    tool = MatrixMessageTools()
    ctx = _make_context()
    response = MagicMock(spec=nio.RoomSendResponse)
    response.event_id = "$react"
    ctx.client.room_send.return_value = response

    with tool_runtime_context(ctx):
        payload = json.loads(await tool.matrix_message(action="react", message="🔥", target="$target"))

    assert payload["status"] == "ok"
    assert payload["action"] == "react"
    assert payload["target"] == "$target"
    ctx.client.room_send.assert_awaited_once_with(
        room_id=ctx.room_id,
        message_type="m.reaction",
        content={
            "m.relates_to": {
                "rel_type": "m.annotation",
                "event_id": "$target",
                "key": "🔥",
            },
        },
        ignore_unverified_devices=True,
    )


@pytest.mark.asyncio
async def test_matrix_message_react_skips_interactive_processing() -> None:
    """React action should not touch interactive-question helpers."""
    tool = MatrixMessageTools()
    ctx = _make_context()
    response = MagicMock(spec=nio.RoomSendResponse)
    response.event_id = "$react"
    ctx.client.room_send.return_value = response

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.should_create_interactive_question",
        ) as mock_should_create,
        patch("mindroom.custom_tools.matrix_conversation_operations.parse_and_format_interactive") as mock_parse,
        patch("mindroom.custom_tools.matrix_conversation_operations.register_interactive_question") as mock_register,
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.add_reaction_buttons",
            new_callable=AsyncMock,
        ) as mock_add_reactions,
        tool_runtime_context(ctx),
    ):
        payload = json.loads(await tool.matrix_message(action="react", message="🔥", target="$target"))

    assert payload["status"] == "ok"
    mock_should_create.assert_not_called()
    mock_parse.assert_not_called()
    mock_register.assert_not_called()
    mock_add_reactions.assert_not_awaited()


@pytest.mark.asyncio
async def test_matrix_message_edit_processes_interactive_blocks() -> None:
    """Edit action should format interactive content and register reactions on the target event."""
    tool = MatrixMessageTools()
    ctx = _make_context(thread_id="$ctx-thread:localhost")
    thread_messages = [
        make_visible_message(event_id="$latest", timestamp=1, sender="@alice:localhost", body="latest"),
    ]
    interactive_message = """Updated prompt.

``` Interactive json
{
  "question": "Which option?",
  "options": [
    {"emoji": "✅", "label": "Approve", "value": "approve"},
    {"emoji": "❌", "label": "Reject", "value": "reject"}
  ]
}
```"""
    formatted_text = parse_and_format_interactive(interactive_message, extract_mapping=False).formatted_text
    ctx.conversation_cache.get_thread_history.return_value = thread_messages

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.edit_message_result",
            new=AsyncMock(side_effect=delivered_matrix_side_effect("$edit_evt")),
        ) as mock_edit,
        patch("mindroom.custom_tools.matrix_conversation_operations.register_interactive_question") as mock_register,
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.add_reaction_buttons",
            new_callable=AsyncMock,
        ) as mock_add_reactions,
        tool_runtime_context(ctx),
    ):
        payload = json.loads(await tool.matrix_message(action="edit", message=interactive_message, target="$target"))

    assert payload["status"] == "ok"
    assert payload["event_id"] == "$edit_evt"
    assert mock_edit.await_args.args[4] == formatted_text
    assert mock_edit.await_args.args[3]["body"] == formatted_text
    mock_register.assert_called_once_with(
        "$target",
        ctx.room_id,
        ctx.thread_id,
        {
            "✅": "approve",
            "1": "approve",
            "❌": "reject",
            "2": "reject",
        },
        ctx.agent_name,
        question_text="Which option?",
        option_labels={
            "✅": "Approve",
            "1": "Approve",
            "❌": "Reject",
            "2": "Reject",
        },
    )
    mock_add_reactions.assert_awaited_once_with(
        ctx.client,
        ctx.room_id,
        "$target",
        [
            {"emoji": "✅", "label": "Approve", "value": "approve"},
            {"emoji": "❌", "label": "Reject", "value": "reject"},
        ],
    )


@pytest.mark.asyncio
async def test_matrix_message_edit_includes_message_extras_on_replacement_wrapper() -> None:
    """Edit action should expose extras on both m.new_content and the outer edit event."""
    tool = MatrixMessageTools()
    ctx = _make_context(thread_id="$ctx-thread:localhost")
    thread_messages = [
        make_visible_message(event_id="$latest", timestamp=1, sender="@alice:localhost", body="latest"),
    ]
    ctx.conversation_cache.get_thread_history.return_value = thread_messages

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.edit_message_result",
            new=AsyncMock(side_effect=delivered_matrix_side_effect("$edit_evt")),
        ) as mock_edit,
        tool_runtime_context(ctx),
    ):
        payload = json.loads(
            await tool.matrix_message(
                action="edit",
                message="Updated answer.",
                target="$target",
                message_extras=[
                    {
                        "title": "Evidence",
                        "content": "extra details",
                    },
                ],
            ),
        )

    assert payload["status"] == "ok"
    new_content = mock_edit.await_args.args[3]
    extra_content = mock_edit.await_args.kwargs["extra_content"]
    expected_extras = {
        "version": 2,
        "sections": [
            {
                "title": "Evidence",
                "content_type": "text/markdown",
                "content": "extra details",
                "collapsed": True,
            },
        ],
    }
    assert new_content[MINDROOM_MESSAGE_EXTRAS_KEY] == expected_extras
    assert extra_content == {MINDROOM_MESSAGE_EXTRAS_KEY: expected_extras}


@pytest.mark.asyncio
async def test_matrix_message_edit_rejects_invalid_message_extras() -> None:
    """Invalid edit extras should fail before edit delivery."""
    tool = MatrixMessageTools()
    ctx = _make_context(thread_id="$ctx-thread:localhost")

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.edit_message_result",
            new=AsyncMock(side_effect=delivered_matrix_side_effect("$edit_evt")),
        ) as mock_edit,
        tool_runtime_context(ctx),
    ):
        payload = json.loads(
            await tool.matrix_message(
                action="edit",
                message="Updated answer.",
                target="$target",
                message_extras=[
                    {
                        "title": "Raw",
                        "content_type": "application/json",
                        "content": "{}",
                    },
                ],
            ),
        )

    assert payload["status"] == "error"
    assert "content_type" in payload["message"]
    mock_edit.assert_not_awaited()


@pytest.mark.asyncio
async def test_matrix_message_edit_plain_text_clears_existing_interactive_question() -> None:
    """Editing away an interactive block should clear the tracked question."""
    tool = MatrixMessageTools()
    ctx = _make_context(thread_id="$ctx-thread:localhost")
    thread_messages = [
        make_visible_message(event_id="$latest", timestamp=1, sender="@alice:localhost", body="latest"),
    ]
    ctx.conversation_cache.get_thread_history.return_value = thread_messages
    interactive.register_interactive_question(
        "$target",
        ctx.room_id,
        ctx.thread_id,
        {"✅": "approve", "1": "approve"},
        ctx.agent_name,
    )

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.edit_message_result",
            new=AsyncMock(side_effect=delivered_matrix_side_effect("$edit_evt")),
        ),
        tool_runtime_context(ctx),
    ):
        payload = json.loads(await tool.matrix_message(action="edit", message="updated text", target="$target"))

    assert payload["status"] == "ok"
    assert "$target" not in interactive._active_questions


@pytest.mark.asyncio
async def test_matrix_message_edit_re_registers_interactive_question() -> None:
    """Interactive edits should reformat the message and replace the question mapping."""
    tool = MatrixMessageTools()
    ctx = _make_context(thread_id="$ctx-thread:localhost")
    thread_messages = [
        make_visible_message(event_id="$latest", timestamp=1, sender="@alice:localhost", body="latest"),
    ]
    interactive_message = """Please choose.

```interactive
{
  "question": "Which option?",
  "options": [
    {"emoji": "✅", "label": "Approve", "value": "approve"},
    {"emoji": "❌", "label": "Reject", "value": "reject"}
  ]
}
```"""
    formatted_text = parse_and_format_interactive(interactive_message, extract_mapping=False).formatted_text
    ctx.conversation_cache.get_thread_history.return_value = thread_messages

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.edit_message_result",
            new=AsyncMock(side_effect=delivered_matrix_side_effect("$edit_evt")),
        ) as mock_edit,
        patch("mindroom.custom_tools.matrix_conversation_operations.clear_interactive_question") as mock_clear,
        patch("mindroom.custom_tools.matrix_conversation_operations.register_interactive_question") as mock_register,
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.add_reaction_buttons",
            new_callable=AsyncMock,
        ) as mock_add_reactions,
        tool_runtime_context(ctx),
    ):
        payload = json.loads(await tool.matrix_message(action="edit", message=interactive_message, target="$target"))

    assert payload["status"] == "ok"
    mock_clear.assert_called_once_with("$target")
    assert mock_edit.await_args.args[4] == formatted_text
    mock_register.assert_called_once_with(
        "$target",
        ctx.room_id,
        ctx.thread_id,
        {
            "✅": "approve",
            "1": "approve",
            "❌": "reject",
            "2": "reject",
        },
        ctx.agent_name,
        question_text="Which option?",
        option_labels={
            "✅": "Approve",
            "1": "Approve",
            "❌": "Reject",
            "2": "Reject",
        },
    )
    mock_add_reactions.assert_awaited_once_with(
        ctx.client,
        ctx.room_id,
        "$target",
        [
            {"emoji": "✅", "label": "Approve", "value": "approve"},
            {"emoji": "❌", "label": "Reject", "value": "reject"},
        ],
    )


def test_resolved_visible_message_to_dict_includes_msgtype() -> None:
    """Thread-list serialization should preserve the visible Matrix msgtype."""
    message = make_visible_message(
        event_id="$notice",
        body="notice",
        content={"body": "notice", "msgtype": "m.notice"},
    )

    assert message.to_dict()["msgtype"] == "m.notice"


@pytest.mark.asyncio
async def test_matrix_message_read_thread_enforces_max_limit() -> None:
    """Thread reads should be bounded by the configured max limit."""
    tool = MatrixMessageTools()
    ctx = _make_context()
    thread_messages = [
        make_visible_message(event_id=f"${index}", timestamp=index, body=f"m{index}") for index in range(100)
    ]
    ctx.conversation_cache.get_thread_history.return_value = thread_messages

    with tool_runtime_context(ctx):
        payload = json.loads(await tool.matrix_message(action="read", limit=999))

    assert payload["status"] == "ok"
    assert payload["limit"] == MatrixMessageTools._MAX_READ_LIMIT
    assert len(payload["messages"]) == MatrixMessageTools._MAX_READ_LIMIT
    assert "edit_options" in payload
    ctx.conversation_cache.get_thread_history.assert_awaited_once_with(
        ctx.room_id,
        ctx.thread_id,
        caller_label="matrix_message_tool",
    )


@pytest.mark.asyncio
async def test_matrix_message_read_thread_includes_edit_options() -> None:
    """Thread reads should include event IDs that can be edited."""
    tool = MatrixMessageTools()
    ctx = _make_context()
    ctx.client.user_id = "@mindroom_general:localhost"
    thread_messages = [
        make_visible_message(event_id="$one", timestamp=1, sender="@alice:localhost", body="earlier message"),
        make_visible_message(
            event_id="$two",
            timestamp=2,
            sender="@mindroom_general:localhost",
            body="latest message",
        ),
    ]
    ctx.conversation_cache.get_thread_history.return_value = thread_messages

    with tool_runtime_context(ctx):
        payload = json.loads(await tool.matrix_message(action="read"))

    assert payload["status"] == "ok"
    assert payload["action"] == "read"
    assert payload["thread_id"] == ctx.thread_id
    assert payload["edit_options"][0]["event_id"] == "$two"
    assert payload["edit_options"][0]["can_edit"] is True
    assert payload["edit_options"][0]["edit_action"] == {"action": "edit", "target": "$two"}
    assert payload["edit_options"][1]["event_id"] == "$one"
    assert payload["edit_options"][1]["can_edit"] is False
    assert "edit_action" not in payload["edit_options"][1]


@pytest.mark.asyncio
async def test_matrix_message_thread_list_requires_thread_context_or_target() -> None:
    """thread-list should fail when no thread can be resolved."""
    tool = MatrixMessageTools()
    ctx = _make_context(thread_id=None)

    with tool_runtime_context(ctx):
        payload = json.loads(await tool.matrix_message(action="thread-list"))

    assert payload["status"] == "error"
    assert payload["action"] == "thread-list"
    assert "thread_id is required" in payload["message"]


@pytest.mark.asyncio
async def test_matrix_message_thread_list_returns_thread_messages() -> None:
    """thread-list should return thread messages and edit options."""
    tool = MatrixMessageTools()
    ctx = _make_context(thread_id=None)
    ctx.client.user_id = "@mindroom_general:localhost"
    thread_messages = [
        make_visible_message(event_id="$one", timestamp=1, sender="@mindroom_general:localhost", body="first"),
        make_visible_message(event_id="$two", timestamp=2, sender="@alice:localhost", body="second"),
    ]
    ctx.conversation_cache.get_thread_history.return_value = thread_messages

    with tool_runtime_context(ctx):
        payload = json.loads(
            await tool.matrix_message(
                action="thread-list",
                thread_id="$thread-other:localhost",
                limit=1,
            ),
        )

    assert payload["status"] == "ok"
    assert payload["action"] == "thread-list"
    assert payload["thread_id"] == "$thread-other:localhost"
    assert payload["messages"] == [thread_messages[-1].to_dict()]
    assert payload["edit_options"][0]["event_id"] == "$two"
    ctx.conversation_cache.get_thread_history.assert_awaited_once_with(
        ctx.room_id,
        "$thread-other:localhost",
        caller_label="matrix_message_tool",
    )


@pytest.mark.asyncio
async def test_matrix_message_thread_list_preserves_notice_messages() -> None:
    """thread-list should surface notice msgtypes unchanged."""
    tool = MatrixMessageTools()
    ctx = _make_context(thread_id=None)
    thread_messages = [
        make_visible_message(event_id="$one", timestamp=1, sender="@alice:localhost", body="first"),
        make_visible_message(
            event_id="$notice",
            timestamp=2,
            sender="@mindroom_general:localhost",
            body="Compacted 12 messages",
            content={"body": "Compacted 12 messages", "msgtype": "m.notice"},
        ),
    ]
    ctx.conversation_cache.get_thread_history.return_value = thread_messages

    with tool_runtime_context(ctx):
        payload = json.loads(
            await tool.matrix_message(
                action="thread-list",
                thread_id="$thread-other:localhost",
                limit=2,
            ),
        )

    assert payload["status"] == "ok"
    assert payload["messages"] == [message.to_dict() for message in thread_messages]
    assert payload["messages"][1]["msgtype"] == "m.notice"
    ctx.conversation_cache.get_thread_history.assert_awaited_once_with(
        ctx.room_id,
        "$thread-other:localhost",
        caller_label="matrix_message_tool",
    )


@pytest.mark.asyncio
async def test_matrix_message_room_threads_returns_paginated_thread_roots() -> None:
    """room-threads should serialize thread roots and forward page tokens."""
    tool = MatrixMessageTools()
    ctx = _make_context()
    page_marker = "page_1"
    next_page = "page_2"
    thread_root = _make_room_thread_root(
        event_id="$thread-root",
        sender="@alice:localhost",
        timestamp=1234,
        body="Root message body",
        reply_count=4,
    )

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.get_room_threads_page",
            new=AsyncMock(return_value=([thread_root], next_page)),
        ) as mock_get_page,
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.thread_root_body_preview",
            new=AsyncMock(return_value="Resolved root message body"),
        ) as mock_preview,
        tool_runtime_context(ctx),
    ):
        payload = json.loads(
            await tool.matrix_message(
                action="room-threads",
                limit=7,
                page_token=page_marker,
            ),
        )

    assert payload["status"] == "ok"
    assert payload["action"] == "room-threads"
    assert payload["room_id"] == ctx.room_id
    assert payload["count"] == 1
    assert payload["threads"] == [
        {
            "thread_id": "$thread-root",
            "sender": "@alice:localhost",
            "timestamp": 1234,
            "body_preview": "Resolved root message body",
            "reply_count": 4,
        },
    ]
    assert payload["next_token"] == next_page
    assert payload["has_more"] is True
    mock_get_page.assert_awaited_once_with(
        ctx.client,
        ctx.room_id,
        limit=7,
        page_token=page_marker,
    )
    mock_preview.assert_awaited_once()


@pytest.mark.asyncio
async def test_matrix_message_room_threads_includes_latest_activity_ts() -> None:
    """room-threads should expose latest activity separately from root creation time."""
    tool = MatrixMessageTools()
    ctx = _make_context()
    thread_root = _make_room_thread_root(
        event_id="$thread-root",
        sender="@alice:localhost",
        timestamp=1234,
        body="Root message body",
        reply_count=4,
    )
    thread_root.source["unsigned"] = {
        "m.relations": {
            "m.thread": {
                "count": 4,
                "latest_event": {
                    "event_id": "$thread-reply",
                    "origin_server_ts": 5678,
                },
            },
        },
    }

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.get_room_threads_page",
            new=AsyncMock(return_value=([thread_root], None)),
        ),
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.thread_root_body_preview",
            new=AsyncMock(return_value="Resolved root message body"),
        ) as mock_preview,
        tool_runtime_context(ctx),
    ):
        payload = json.loads(await tool.matrix_message(action="room-threads", limit=1))

    assert payload["status"] == "ok"
    assert payload["threads"] == [
        {
            "thread_id": "$thread-root",
            "sender": "@alice:localhost",
            "timestamp": 1234,
            "latest_activity_ts": 5678,
            "body_preview": "Resolved root message body",
            "reply_count": 4,
        },
    ]
    mock_preview.assert_awaited_once()


@pytest.mark.asyncio
async def test_matrix_message_room_threads_uses_bundled_replacement_preview_for_text_root() -> None:
    """room-threads should prefer bundled replacement bodies for text roots."""
    tool = MatrixMessageTools()
    ctx = _make_context()
    thread_root = _make_room_thread_root(
        event_id="$thread-root",
        sender="@alice:localhost",
        timestamp=1234,
        body="Thinking...",
        reply_count=4,
    )
    thread_root.source["unsigned"] = {
        "m.relations": {
            "m.thread": {"count": 4},
            "m.replace": _make_bundled_replacement(
                event_id="$thread-root",
                body="Final root message",
                msgtype="m.text",
            ),
        },
    }

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.get_room_threads_page",
            new=AsyncMock(return_value=([thread_root], None)),
        ) as mock_get_page,
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.extract_and_resolve_message",
            new=AsyncMock(),
        ) as mock_extract,
        tool_runtime_context(ctx),
    ):
        payload = json.loads(await tool.matrix_message(action="room-threads", limit=1))

    assert payload["status"] == "ok"
    assert payload["threads"] == [
        {
            "thread_id": "$thread-root",
            "sender": "@alice:localhost",
            "timestamp": 1234,
            "body_preview": "Final root message",
            "reply_count": 4,
        },
    ]
    mock_get_page.assert_awaited_once_with(
        ctx.client,
        ctx.room_id,
        limit=1,
        page_token=None,
    )
    mock_extract.assert_not_awaited()


@pytest.mark.asyncio
async def test_matrix_message_room_threads_prefers_trusted_canonical_bundled_preview() -> None:
    """room-threads should hide transient warmup text for trusted bundled local edits."""
    tool = MatrixMessageTools()
    ctx = _make_context()
    thread_root = _make_room_thread_root(
        event_id="$thread-root",
        sender="@alice:localhost",
        timestamp=1234,
        body="Thinking...",
        reply_count=4,
    )
    thread_root.source["unsigned"] = {
        "m.relations": {
            "m.thread": {"count": 4},
            "m.replace": _make_bundled_replacement(
                event_id="$thread-root",
                body="Final root message\n\n⏳ Preparing isolated worker...",
                msgtype="m.text",
                sender="@mindroom_general:localhost",
                visible_body="Final root message",
            ),
        },
    }

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.get_room_threads_page",
            new=AsyncMock(return_value=([thread_root], None)),
        ),
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.extract_and_resolve_message",
            new=AsyncMock(),
        ) as mock_extract,
        tool_runtime_context(ctx),
    ):
        payload = json.loads(await tool.matrix_message(action="room-threads", limit=1))

    assert payload["status"] == "ok"
    assert payload["threads"][0]["body_preview"] == "Final root message"
    mock_extract.assert_not_awaited()


@pytest.mark.asyncio
@pytest.mark.parametrize("bundle_key", ["event", "latest_event"])
async def test_matrix_message_room_threads_uses_nested_bundled_replacement_preview_for_notice_root(
    bundle_key: str,
) -> None:
    """room-threads should read notice previews from nested bundled replacement events."""
    tool = MatrixMessageTools()
    ctx = _make_context()
    thread_root = nio.RoomMessageNotice.from_dict(
        {
            "event_id": "$thread-notice",
            "sender": "@alice:localhost",
            "origin_server_ts": 1234,
            "content": {"msgtype": "m.notice", "body": "Thinking..."},
        },
    )
    thread_root.source["unsigned"] = {
        "m.relations": {
            "m.thread": {"count": 2},
            "m.replace": _make_bundled_replacement(
                event_id="$thread-notice",
                body="Compacted 12 messages",
                msgtype="m.notice",
                bundle_key=bundle_key,
            ),
        },
    }

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.get_room_threads_page",
            new=AsyncMock(return_value=([thread_root], None)),
        ) as mock_get_page,
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.extract_and_resolve_message",
            new=AsyncMock(),
        ) as mock_extract,
        tool_runtime_context(ctx),
    ):
        payload = json.loads(await tool.matrix_message(action="room-threads", limit=1))

    assert payload["status"] == "ok"
    assert payload["threads"] == [
        {
            "thread_id": "$thread-notice",
            "sender": "@alice:localhost",
            "timestamp": 1234,
            "body_preview": "Compacted 12 messages",
            "reply_count": 2,
        },
    ]
    mock_get_page.assert_awaited_once_with(
        ctx.client,
        ctx.room_id,
        limit=1,
        page_token=None,
    )
    mock_extract.assert_not_awaited()


@pytest.mark.asyncio
async def test_matrix_message_room_threads_resolves_notice_root_without_replacement() -> None:
    """room-threads should resolve notice roots through the canonical message path."""
    tool = MatrixMessageTools()
    ctx = _make_context()
    thread_root = nio.RoomMessageNotice.from_dict(
        {
            "event_id": "$thread-notice",
            "sender": "@alice:localhost",
            "origin_server_ts": 1234,
            "content": {"msgtype": "m.notice", "body": "Thinking..."},
        },
    )
    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.get_room_threads_page",
            new=AsyncMock(return_value=([thread_root], None)),
        ) as mock_get_page,
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.thread_root_body_preview",
            new=AsyncMock(return_value="Resolved notice body"),
        ) as mock_preview,
        tool_runtime_context(ctx),
    ):
        payload = json.loads(await tool.matrix_message(action="room-threads", limit=1))

    assert payload["status"] == "ok"
    assert payload["threads"] == [
        {
            "thread_id": "$thread-notice",
            "sender": "@alice:localhost",
            "timestamp": 1234,
            "body_preview": "Resolved notice body",
            "reply_count": 0,
        },
    ]
    mock_get_page.assert_awaited_once_with(
        ctx.client,
        ctx.room_id,
        limit=1,
        page_token=None,
    )
    mock_preview.assert_awaited_once_with(
        thread_root,
        client=ctx.client,
        config=ctx.config,
        runtime_paths=ctx.runtime_paths,
        trusted_sender_ids=ANY,
    )


@pytest.mark.asyncio
async def test_matrix_message_room_threads_resolves_large_file_root_through_canonical_visible_body() -> None:
    """room-threads should hydrate large m.file roots before building previews."""
    tool = MatrixMessageTools()
    ctx = _make_context()
    ctx.client.download = AsyncMock(
        return_value=MagicMock(
            spec=nio.DownloadResponse,
            body=json.dumps(
                {
                    "msgtype": "m.text",
                    "body": "Final large root message\n\n⏳ Preparing isolated worker...",
                    "io.mindroom.visible_body": "Final large root message",
                },
            ).encode("utf-8"),
        ),
    )
    thread_root = nio.RoomMessageFile.from_dict(
        {
            "type": "m.room.message",
            "event_id": "$thread-large",
            "sender": "@mindroom_general:localhost",
            "origin_server_ts": 1234,
            "content": {
                "msgtype": "m.file",
                "body": "Preview root...",
                "info": {"mimetype": "application/json"},
                "io.mindroom.long_text": {
                    "version": 2,
                    "encoding": "matrix_event_content_json",
                },
                "url": "mxc://server/thread-large",
            },
            "unsigned": {"m.relations": {"m.thread": {"count": 4}}},
        },
    )

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.get_room_threads_page",
            new=AsyncMock(return_value=([thread_root], None)),
        ),
        tool_runtime_context(ctx),
    ):
        payload = json.loads(await tool.matrix_message(action="room-threads", limit=1))

    assert payload["status"] == "ok"
    assert payload["threads"][0]["body_preview"] == "Final large root message"


@pytest.mark.asyncio
async def test_matrix_message_room_threads_resolves_large_bundled_replacement_through_canonical_visible_body() -> None:
    """room-threads should hydrate large bundled latest edits before building previews."""
    tool = MatrixMessageTools()
    ctx = _make_context()
    ctx.client.download = AsyncMock(
        return_value=MagicMock(
            spec=nio.DownloadResponse,
            body=json.dumps(
                {
                    "msgtype": "m.text",
                    "body": "Final bundled edit\n\n⏳ Preparing isolated worker...",
                    "io.mindroom.visible_body": "Final bundled edit",
                },
            ).encode("utf-8"),
        ),
    )
    thread_root = _make_room_thread_root(
        event_id="$thread-root",
        sender="@alice:localhost",
        timestamp=1234,
        body="Original root",
        reply_count=4,
    )
    thread_root.source["unsigned"] = {
        "m.relations": {
            "m.thread": {"count": 4},
            "m.replace": _make_bundled_replacement(
                event_id="$thread-root",
                body="Preview latest edit...",
                msgtype="m.file",
                sender="@mindroom_general:localhost",
                long_text={
                    "version": 2,
                    "encoding": "matrix_event_content_json",
                },
                url="mxc://server/thread-root-edit",
            ),
        },
    }

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.get_room_threads_page",
            new=AsyncMock(return_value=([thread_root], None)),
        ),
        tool_runtime_context(ctx),
    ):
        payload = json.loads(await tool.matrix_message(action="room-threads", limit=1))

    assert payload["status"] == "ok"
    assert payload["threads"][0]["body_preview"] == "Final bundled edit"


@pytest.mark.asyncio
async def test_matrix_message_room_threads_skips_malformed_roots() -> None:
    """room-threads should skip malformed roots instead of crashing the whole action."""
    tool = MatrixMessageTools()
    ctx = _make_context()
    thread_root = _make_room_thread_root(
        event_id="$thread-root",
        sender="@alice:localhost",
        timestamp=1234,
        body="Root message body",
        reply_count=4,
    )

    class MalformedThreadRoot:
        event_id: ClassVar[object] = None
        sender: ClassVar[str] = "@broken:localhost"
        server_timestamp: ClassVar[int] = 1234
        source: ClassVar[dict[str, object]] = {
            "type": "m.room.message",
            "content": {"msgtype": "m.text", "body": "broken"},
        }

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.get_room_threads_page",
            new=AsyncMock(return_value=([MalformedThreadRoot(), thread_root], None)),
        ),
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.thread_root_body_preview",
            new=AsyncMock(return_value="Resolved root message body"),
        ) as mock_preview,
        patch("mindroom.custom_tools.matrix_conversation_operations.logger.warning") as mock_warning,
        tool_runtime_context(ctx),
    ):
        payload = json.loads(await tool.matrix_message(action="room-threads"))

    assert payload["status"] == "ok"
    assert payload["count"] == 1
    assert payload["threads"] == [
        {
            "thread_id": "$thread-root",
            "sender": "@alice:localhost",
            "timestamp": 1234,
            "body_preview": "Resolved root message body",
            "reply_count": 4,
        },
    ]
    mock_preview.assert_awaited_once_with(
        thread_root,
        client=ctx.client,
        config=ctx.config,
        runtime_paths=ctx.runtime_paths,
        trusted_sender_ids=ANY,
    )
    mock_warning.assert_called_once()


@pytest.mark.asyncio
async def test_matrix_message_room_threads_has_more_false_without_next_token() -> None:
    """room-threads should derive has_more solely from next_token."""
    tool = MatrixMessageTools()
    ctx = _make_context()
    thread_roots = [
        _make_room_thread_root(
            event_id="$thread-one",
            sender="@alice:localhost",
            timestamp=1,
            body="First thread",
        ),
        _make_room_thread_root(
            event_id="$thread-two",
            sender="@bob:localhost",
            timestamp=2,
            body="Second thread",
        ),
    ]

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.get_room_threads_page",
            new=AsyncMock(return_value=(thread_roots, None)),
        ),
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.extract_and_resolve_message",
            new=AsyncMock(side_effect=[{"body": "First thread"}, {"body": "Second thread"}]),
        ),
        tool_runtime_context(ctx),
    ):
        payload = json.loads(await tool.matrix_message(action="room-threads", limit=2))

    assert payload["status"] == "ok"
    assert payload["count"] == 2
    assert payload["next_token"] is None
    assert payload["has_more"] is False


@pytest.mark.asyncio
async def test_matrix_message_room_threads_empty_room() -> None:
    """room-threads should return an empty success payload for rooms without threads."""
    tool = MatrixMessageTools()
    ctx = _make_context()

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.get_room_threads_page",
            new=AsyncMock(return_value=([], None)),
        ) as mock_get_page,
        tool_runtime_context(ctx),
    ):
        payload = json.loads(await tool.matrix_message(action="room-threads"))

    assert payload == {
        "action": "room-threads",
        "count": 0,
        "has_more": False,
        "next_token": None,
        "room_id": ctx.room_id,
        "status": "ok",
        "threads": [],
        "tool": "matrix_message",
    }
    mock_get_page.assert_awaited_once_with(
        ctx.client,
        ctx.room_id,
        limit=MatrixMessageTools._DEFAULT_READ_LIMIT,
        page_token=None,
    )


@pytest.mark.asyncio
async def test_matrix_message_room_threads_returns_structured_api_error() -> None:
    """room-threads should surface Matrix API failures without a fallback scan."""
    tool = MatrixMessageTools()
    ctx = _make_context()
    stale_page = "stale"

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.get_room_threads_page",
            new=AsyncMock(
                side_effect=RoomThreadsPageError(
                    response="RoomThreadsError: M_INVALID_PARAM Unknown or invalid from token",
                    errcode="M_INVALID_PARAM",
                ),
            ),
        ),
        tool_runtime_context(ctx),
    ):
        payload = json.loads(await tool.matrix_message(action="room-threads", page_token=stale_page))

    assert payload["status"] == "error"
    assert payload["action"] == "room-threads"
    assert payload["room_id"] == ctx.room_id
    assert payload["response"] == "RoomThreadsError: M_INVALID_PARAM Unknown or invalid from token"
    assert payload["errcode"] == "M_INVALID_PARAM"


@pytest.mark.asyncio
async def test_matrix_message_room_threads_preserves_rate_limit_details() -> None:
    """room-threads should preserve retry metadata from Matrix rate limits."""
    tool = MatrixMessageTools()
    ctx = _make_context()

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.get_room_threads_page",
            new=AsyncMock(
                side_effect=RoomThreadsPageError(
                    response="RoomThreadsError: M_LIMIT_EXCEEDED Too many requests - retry after 1500ms",
                    errcode="M_LIMIT_EXCEEDED",
                    retry_after_ms=1500,
                ),
            ),
        ),
        tool_runtime_context(ctx),
    ):
        payload = json.loads(await tool.matrix_message(action="room-threads"))

    assert payload["status"] == "error"
    assert payload["action"] == "room-threads"
    assert payload["room_id"] == ctx.room_id
    assert payload["response"] == "RoomThreadsError: M_LIMIT_EXCEEDED Too many requests - retry after 1500ms"
    assert payload["errcode"] == "M_LIMIT_EXCEEDED"
    assert payload["retry_after_ms"] == 1500


@pytest.mark.asyncio
async def test_matrix_message_room_threads_returns_structured_transport_error() -> None:
    """room-threads should convert transport exceptions into structured tool errors."""
    tool = MatrixMessageTools()
    ctx = _make_context()

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.get_room_threads_page",
            new=AsyncMock(
                side_effect=RoomThreadsPageError(
                    response="TimeoutError: request timed out",
                ),
            ),
        ),
        tool_runtime_context(ctx),
    ):
        payload = json.loads(await tool.matrix_message(action="room-threads"))

    assert payload["status"] == "error"
    assert payload["action"] == "room-threads"
    assert payload["room_id"] == ctx.room_id
    assert payload["response"] == "TimeoutError: request timed out"
    assert "errcode" not in payload
    assert "retry_after_ms" not in payload


@pytest.mark.asyncio
async def test_matrix_message_room_threads_clamps_limit() -> None:
    """room-threads should reuse the existing read-limit clamp."""
    tool = MatrixMessageTools()
    ctx = _make_context()

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.get_room_threads_page",
            new=AsyncMock(return_value=([], None)),
        ) as mock_get_page,
        tool_runtime_context(ctx),
    ):
        payload = json.loads(await tool.matrix_message(action="room-threads", limit=999))

    assert payload["status"] == "ok"
    mock_get_page.assert_awaited_once_with(
        ctx.client,
        ctx.room_id,
        limit=MatrixMessageTools._MAX_READ_LIMIT,
        page_token=None,
    )


@pytest.mark.asyncio
async def test_matrix_message_room_threads_encrypted_preview_is_redacted() -> None:
    """Encrypted thread roots should use the explicit encrypted preview."""
    tool = MatrixMessageTools()
    ctx = _make_context()
    encrypted_root = _make_room_thread_root(
        event_id="$thread-encrypted",
        sender="@alice:localhost",
        timestamp=1234,
        encrypted=True,
    )

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.get_room_threads_page",
            new=AsyncMock(return_value=([encrypted_root], None)),
        ),
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.extract_and_resolve_message",
            new=AsyncMock(),
        ) as mock_extract,
        tool_runtime_context(ctx),
    ):
        payload = json.loads(await tool.matrix_message(action="room-threads"))

    assert payload["status"] == "ok"
    assert payload["threads"][0]["body_preview"] == "[encrypted]"
    assert payload["threads"][0]["reply_count"] == 0
    mock_extract.assert_not_awaited()


@pytest.mark.asyncio
async def test_matrix_message_read_room_happy_path() -> None:
    """Room reads should resolve message events when no thread is active."""
    tool = MatrixMessageTools()
    ctx = _make_context(thread_id=None)
    response = nio.RoomMessagesResponse.from_dict(
        {
            "chunk": [
                {
                    "type": "m.room.message",
                    "event_id": "$evt",
                    "sender": "@alice:localhost",
                    "origin_server_ts": 1,
                    "content": {"msgtype": "m.text", "body": "hello"},
                },
            ],
            "start": "s",
            "end": "e",
        },
        ctx.room_id,
    )
    ctx.client.room_messages.return_value = response

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.extract_and_resolve_message",
            new=AsyncMock(return_value={"event_id": "$evt", "body": "hello"}),
        ) as mock_extract,
        tool_runtime_context(ctx),
    ):
        payload = json.loads(await tool.matrix_message(action="read", limit=5))

    assert payload["status"] == "ok"
    assert payload["limit"] == 5
    assert payload["messages"] == [{"event_id": "$evt", "body": "hello"}]
    ctx.client.room_messages.assert_awaited_once_with(
        ctx.room_id,
        limit=5,
        direction=nio.MessageDirection.back,
        message_filter={"types": ["m.room.message"]},
    )
    mock_extract.assert_awaited_once()


@pytest.mark.asyncio
async def test_matrix_message_read_room_includes_notice_events() -> None:
    """Room reads should keep both text and notice events."""
    tool = MatrixMessageTools()
    ctx = _make_context(thread_id=None)
    response = nio.RoomMessagesResponse.from_dict(
        {
            "chunk": [
                {
                    "type": "m.room.message",
                    "event_id": "$notice",
                    "sender": "@mindroom:localhost",
                    "origin_server_ts": 2,
                    "content": {"msgtype": "m.notice", "body": "Compacted 12 messages"},
                },
                {
                    "type": "m.room.message",
                    "event_id": "$text",
                    "sender": "@alice:localhost",
                    "origin_server_ts": 1,
                    "content": {"msgtype": "m.text", "body": "hello"},
                },
            ],
            "start": "s",
            "end": "e",
        },
        ctx.room_id,
    )
    ctx.client.room_messages.return_value = response
    extracted_messages = {
        "$text": {"event_id": "$text", "body": "hello"},
        "$notice": {"event_id": "$notice", "body": "Compacted 12 messages", "msgtype": "m.notice"},
    }

    async def _extract(
        event: nio.Event,
        _client: nio.AsyncClient,
        *,
        config: Config,
        runtime_paths: object,
        trusted_sender_ids: frozenset[str],
    ) -> dict[str, object]:
        assert config is ctx.config
        assert runtime_paths == ctx.runtime_paths
        assert trusted_sender_ids
        return extracted_messages[event.event_id]

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.extract_and_resolve_message",
            new=AsyncMock(side_effect=_extract),
        ) as mock_extract,
        tool_runtime_context(ctx),
    ):
        payload = json.loads(await tool.matrix_message(action="read", limit=5))

    assert payload["status"] == "ok"
    assert payload["messages"] == [
        {"event_id": "$text", "body": "hello"},
        {"event_id": "$notice", "body": "Compacted 12 messages", "msgtype": "m.notice"},
    ]
    assert mock_extract.await_count == 2


@pytest.mark.asyncio
async def test_matrix_message_read_room_precomputes_trusted_sender_ids_once() -> None:
    """Room reads should resolve the trust set once and pass it through every extraction."""
    tool = MatrixMessageTools()
    ctx = _make_context(thread_id=None)
    response = nio.RoomMessagesResponse.from_dict(
        {
            "chunk": [
                {
                    "type": "m.room.message",
                    "event_id": "$notice",
                    "sender": "@mindroom:localhost",
                    "origin_server_ts": 2,
                    "content": {"msgtype": "m.notice", "body": "Compacted 12 messages"},
                },
                {
                    "type": "m.room.message",
                    "event_id": "$text",
                    "sender": "@alice:localhost",
                    "origin_server_ts": 1,
                    "content": {"msgtype": "m.text", "body": "hello"},
                },
            ],
            "start": "s",
            "end": "e",
        },
        ctx.room_id,
    )
    ctx.client.room_messages.return_value = response
    trusted_sender_ids = frozenset({"@mindroom_general:localhost"})

    async def _extract(
        event: nio.Event,
        _client: nio.AsyncClient,
        *,
        config: Config,
        runtime_paths: object,
        trusted_sender_ids: frozenset[str],
    ) -> dict[str, object]:
        assert config is ctx.config
        assert runtime_paths == ctx.runtime_paths
        assert trusted_sender_ids is trusted_sender_ids_for_assertion
        return {"event_id": event.event_id, "body": event.source["content"]["body"]}

    trusted_sender_ids_for_assertion = trusted_sender_ids

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.trusted_visible_sender_ids",
            return_value=trusted_sender_ids,
        ) as mock_trusted_sender_ids,
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.extract_and_resolve_message",
            new=AsyncMock(side_effect=_extract),
        ) as mock_extract,
        tool_runtime_context(ctx),
    ):
        payload = json.loads(await tool.matrix_message(action="read", limit=5))

    assert payload["status"] == "ok"
    assert payload["messages"] == [
        {"event_id": "$text", "body": "hello"},
        {"event_id": "$notice", "body": "Compacted 12 messages"},
    ]
    mock_trusted_sender_ids.assert_called_once_with(ctx.config, ctx.runtime_paths)
    assert mock_extract.await_count == 2


@pytest.mark.asyncio
async def test_matrix_message_read_room_sentinel_uses_room_timeline() -> None:
    """thread_id='room' should bypass the current thread and read the room timeline."""
    tool = MatrixMessageTools()
    ctx = _make_context(thread_id="$ctx-thread:localhost")
    response = nio.RoomMessagesResponse.from_dict(
        {
            "chunk": [
                {
                    "type": "m.room.message",
                    "event_id": "$evt",
                    "sender": "@alice:localhost",
                    "origin_server_ts": 1,
                    "content": {"msgtype": "m.text", "body": "hello from room"},
                },
            ],
            "start": "s",
            "end": "e",
        },
        ctx.room_id,
    )
    ctx.client.room_messages.return_value = response

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.extract_and_resolve_message",
            new=AsyncMock(return_value={"event_id": "$evt", "body": "hello from room"}),
        ) as mock_extract,
        tool_runtime_context(ctx),
    ):
        payload = json.loads(await tool.matrix_message(action="read", thread_id="room", limit=5))

    assert payload["status"] == "ok"
    assert payload["action"] == "read"
    assert payload["limit"] == 5
    assert payload["messages"] == [{"event_id": "$evt", "body": "hello from room"}]
    assert "thread_id" not in payload
    ctx.client.room_messages.assert_awaited_once_with(
        ctx.room_id,
        limit=5,
        direction=nio.MessageDirection.back,
        message_filter={"types": ["m.room.message"]},
    )
    mock_extract.assert_awaited_once()
    ctx.conversation_cache.get_thread_history.assert_not_awaited()


@pytest.mark.asyncio
async def test_matrix_message_read_explicit_thread_id_still_reads_that_thread() -> None:
    """Explicit thread IDs should win over runtime thread fallback for read."""
    tool = MatrixMessageTools()
    ctx = _make_context(thread_id="$ctx-thread:localhost")
    thread_messages = [
        make_visible_message(event_id="$one", timestamp=1, body="first"),
        make_visible_message(event_id="$two", timestamp=2, body="second"),
    ]
    ctx.conversation_cache.get_thread_history.return_value = thread_messages

    with tool_runtime_context(ctx):
        payload = json.loads(
            await tool.matrix_message(action="read", thread_id="$thread-other:localhost", limit=1),
        )

    assert payload["status"] == "ok"
    assert payload["action"] == "read"
    assert payload["thread_id"] == "$thread-other:localhost"
    assert payload["messages"] == [thread_messages[-1].to_dict()]
    ctx.conversation_cache.get_thread_history.assert_awaited_once_with(
        ctx.room_id,
        "$thread-other:localhost",
        caller_label="matrix_message_tool",
    )
    ctx.client.room_messages.assert_not_awaited()


@pytest.mark.asyncio
async def test_matrix_message_edit_happy_path() -> None:
    """Edit should update an existing message by target event ID."""
    tool = MatrixMessageTools()
    event_cache = MagicMock()
    ctx = _make_context(thread_id="$ctx-thread:localhost", event_cache=event_cache)
    ctx.conversation_cache.get_latest_thread_event_id_if_needed = AsyncMock(return_value="$latest")
    sent_content: dict[str, object] = {}

    async def _deliver_edit(
        _client: object,
        _room_id: str,
        content: dict[str, object],
        **_kwargs: object,
    ) -> DeliveredMatrixEvent:
        sent_content.update(content)
        return delivered_matrix_event("$edit_evt", content)

    with (
        patch(
            "mindroom.matrix.client_delivery.send_message_result",
            new=AsyncMock(side_effect=_deliver_edit),
        ),
        tool_runtime_context(ctx),
    ):
        payload = json.loads(await tool.matrix_message(action="edit", message="updated text", target="$target"))

    assert payload["status"] == "ok"
    assert payload["action"] == "edit"
    assert payload["target"] == "$target"
    assert payload["event_id"] == "$edit_evt"
    relation = sent_content["m.relates_to"]
    assert relation == {"rel_type": "m.replace", "event_id": "$target"}
    replacement = sent_content["m.new_content"]
    assert isinstance(replacement, dict)
    assert replacement["body"] == "updated text"
    assert "m.relates_to" not in replacement
    ctx.conversation_cache.get_latest_thread_event_id_if_needed.assert_not_awaited()
    ctx.conversation_cache.get_thread_history.assert_not_awaited()


@pytest.mark.asyncio
async def test_matrix_message_edit_requires_target() -> None:
    """Edit action should require target event ID."""
    tool = MatrixMessageTools()
    ctx = _make_context()

    with tool_runtime_context(ctx):
        payload = json.loads(await tool.matrix_message(action="edit", message="updated text"))

    assert payload["status"] == "error"
    assert "target event_id is required for edit" in payload["message"]


@pytest.mark.asyncio
async def test_matrix_message_edit_requires_message() -> None:
    """Edit action should require non-empty replacement text."""
    tool = MatrixMessageTools()
    ctx = _make_context()

    with tool_runtime_context(ctx):
        payload = json.loads(await tool.matrix_message(action="edit", target="$target", message="  "))

    assert payload["status"] == "error"
    assert "message is required for edit" in payload["message"]


@pytest.mark.asyncio
async def test_matrix_message_send_validates_non_empty_message() -> None:
    """Send should reject calls where both message and attachments are empty."""
    tool = MatrixMessageTools()
    ctx = _make_context()

    with tool_runtime_context(ctx):
        payload = json.loads(await tool.matrix_message(action="send", message="  "))

    assert payload["status"] == "error"
    assert "At least one of message, attachment_ids, or attachment_file_paths" in payload["message"]


@pytest.mark.asyncio
async def test_matrix_message_rejects_attachments_for_non_send_actions(tmp_path: Path) -> None:
    """Attachments should be accepted only by send/reply/thread-reply actions."""
    tool = MatrixMessageTools()
    ctx = _make_context(storage_path=tmp_path)

    with tool_runtime_context(ctx):
        payload = json.loads(
            await tool.matrix_message(
                action="react",
                target="$target",
                attachment_ids=["att_upload"],
            ),
        )

    assert payload["status"] == "error"
    assert "only supported for send, reply, and thread-reply" in payload["message"]


@pytest.mark.asyncio
async def test_matrix_message_rejects_non_att_attachment_references(tmp_path: Path) -> None:
    """Attachment refs should require context-scoped att_* IDs."""
    tool = MatrixMessageTools()
    ctx = _make_context(storage_path=tmp_path)

    with tool_runtime_context(ctx):
        payload = json.loads(
            await tool.matrix_message(
                action="send",
                attachment_ids=["output.txt"],
            ),
        )

    assert payload["status"] == "error"
    assert "must be context attachment IDs" in payload["message"]


@pytest.mark.asyncio
async def test_matrix_message_rejects_attachment_count_over_limit(tmp_path: Path) -> None:
    """Send should enforce a maximum attachment count per call."""
    tool = MatrixMessageTools()
    ctx = _make_context(storage_path=tmp_path)

    with tool_runtime_context(ctx):
        payload = json.loads(
            await tool.matrix_message(
                action="send",
                attachment_ids=["att_over"] * 6,
            ),
        )

    assert payload["status"] == "error"
    assert "cannot exceed 5" in payload["message"]


@pytest.mark.asyncio
async def test_matrix_message_reply_requires_thread_when_context_has_none() -> None:
    """Reply action should fail when no thread is provided or active."""
    tool = MatrixMessageTools()
    ctx = _make_context(thread_id=None)

    with tool_runtime_context(ctx):
        payload = json.loads(await tool.matrix_message(action="reply", message="hello"))

    assert payload["status"] == "error"
    assert "thread_id is required" in payload["message"]


@pytest.mark.asyncio
async def test_matrix_message_reply_room_sentinel_disables_context_thread_fallback() -> None:
    """thread_id='room' should disable reply thread inheritance and keep reply invalid."""
    tool = MatrixMessageTools()
    ctx = _make_context(thread_id="$ctx-thread:localhost")

    with (
        patch("mindroom.custom_tools.matrix_conversation_operations.send_message_result", new=AsyncMock()) as mock_send,
        tool_runtime_context(ctx),
    ):
        payload = json.loads(await tool.matrix_message(action="reply", thread_id="room", message="hello"))

    assert payload["status"] == "error"
    assert payload["action"] == "reply"
    assert "thread_id is required" in payload["message"]
    mock_send.assert_not_awaited()


@pytest.mark.asyncio
async def test_matrix_message_react_requires_target() -> None:
    """React action should validate that target event ID is provided."""
    tool = MatrixMessageTools()
    ctx = _make_context()

    with tool_runtime_context(ctx):
        payload = json.loads(await tool.matrix_message(action="react", message="👍"))

    assert payload["status"] == "error"
    assert "target event_id is required" in payload["message"]


@pytest.mark.asyncio
async def test_matrix_message_explicit_room_target_requires_authorization() -> None:
    """Explicit room targeting should enforce authorization checks."""
    tool = MatrixMessageTools()
    ctx = _make_context()

    with tool_runtime_context(ctx):
        payload = json.loads(await tool.matrix_message(action="send", message="hello", room_id="!other:localhost"))

    assert payload["status"] == "error"
    assert "Not authorized" in payload["message"]


@pytest.mark.asyncio
async def test_matrix_message_rejects_unsupported_action() -> None:
    """Unsupported actions should return a clear validation error."""
    tool = MatrixMessageTools()
    ctx = _make_context()

    with tool_runtime_context(ctx):
        payload = json.loads(await tool.matrix_message(action="delete", message="hello"))

    assert payload["status"] == "error"
    assert payload["action"] == "delete"
    assert "Unsupported action" in payload["message"]
    assert "reply" in payload["message"]


@pytest.mark.asyncio
async def test_matrix_message_rate_limit_guardrail() -> None:
    """Tool should block rapid repeated actions in the same room context."""
    tool = MatrixMessageTools()
    ctx = _make_context()

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.send_message_result",
            new=AsyncMock(side_effect=delivered_matrix_side_effect("$evt")),
        ),
        patch.object(MatrixMessageTools, "_RATE_LIMIT_MAX_ACTIONS", 1),
        patch.object(MatrixMessageTools, "_RATE_LIMIT_WINDOW_SECONDS", 60.0),
        tool_runtime_context(ctx),
    ):
        first = json.loads(await tool.matrix_message(action="send", message="first"))
        second = json.loads(await tool.matrix_message(action="send", message="second"))

    assert first["status"] == "ok"
    assert second["status"] == "error"
    assert "Rate limit exceeded" in second["message"]


@pytest.mark.asyncio
async def test_matrix_message_rate_limit_counts_attachments_weight(tmp_path: Path) -> None:
    """Rate limiting should charge one tick for message plus one per attachment."""
    tool = MatrixMessageTools()
    sample_file = tmp_path / "upload.txt"
    sample_file.write_text("payload", encoding="utf-8")
    attachment = register_local_attachment(
        tmp_path,
        sample_file,
        kind="file",
        attachment_id="att_weighted",
    )
    assert attachment is not None
    ctx = _make_context(storage_path=tmp_path, attachment_ids=("att_weighted",))

    with (
        patch(
            "mindroom.custom_tools.matrix_conversation_operations.send_message_result",
            new=AsyncMock(side_effect=delivered_matrix_side_effect("$evt")),
        ),
        patch(
            "mindroom.custom_tools.attachments.send_file_message",
            new=AsyncMock(return_value="$file_evt"),
        ),
        patch.object(MatrixMessageTools, "_RATE_LIMIT_MAX_ACTIONS", 2),
        patch.object(MatrixMessageTools, "_RATE_LIMIT_WINDOW_SECONDS", 60.0),
        tool_runtime_context(ctx),
    ):
        first = json.loads(
            await tool.matrix_message(
                action="send",
                message="first",
                attachment_ids=["att_weighted"],
            ),
        )
        second = json.loads(await tool.matrix_message(action="send", message="second"))

    assert first["status"] == "ok"
    assert second["status"] == "error"
    assert "Rate limit exceeded" in second["message"]


@pytest.mark.asyncio
async def test_matrix_message_context_returns_runtime_metadata() -> None:
    """Context action should expose room/thread/event identifiers for targeting."""
    tool = MatrixMessageTools()
    ctx = _make_context(thread_id="$thread-root:localhost", reply_to_event_id="$event:localhost")

    with tool_runtime_context(ctx):
        payload = json.loads(await tool.matrix_message(action="context"))

    assert payload["status"] == "ok"
    assert payload["action"] == "context"
    assert payload["room_id"] == ctx.room_id
    assert payload["thread_id"] == "$thread-root:localhost"
    assert payload["reply_to_event_id"] == "$event:localhost"


@pytest.mark.asyncio
async def test_matrix_message_context_room_sentinel_normalizes_to_room_level() -> None:
    """Context should not leak the room sentinel as a fake thread ID."""
    tool = MatrixMessageTools()
    ctx = _make_context(thread_id="$thread-root:localhost", reply_to_event_id="$event:localhost")

    with tool_runtime_context(ctx):
        payload = json.loads(await tool.matrix_message(action="context", thread_id="room"))

    assert payload["status"] == "ok"
    assert payload["action"] == "context"
    assert payload["room_id"] == ctx.room_id
    assert payload["thread_id"] is None
    assert payload["reply_to_event_id"] == "$event:localhost"


@pytest.mark.asyncio
async def test_matrix_message_cross_room_reply_does_not_inherit_context_thread() -> None:
    """Authorized cross-room reply should not inherit the origin room's thread."""
    tool = MatrixMessageTools()
    ctx = _make_context(thread_id="$origin-thread:localhost")

    with (
        patch("mindroom.custom_tools.matrix_message.room_access_allowed", return_value=True),
        tool_runtime_context(ctx),
    ):
        payload = json.loads(
            await tool.matrix_message(action="reply", message="hello", room_id="!other:localhost"),
        )

    assert payload["status"] == "error"
    assert "thread_id is required" in payload["message"]


@pytest.mark.asyncio
async def test_matrix_message_cross_room_read_defaults_to_room_level() -> None:
    """Authorized cross-room read should not use the origin room's thread."""
    tool = MatrixMessageTools()
    ctx = _make_context(thread_id="$origin-thread:localhost")
    response = MagicMock(spec=nio.RoomMessagesResponse)
    response.chunk = []
    ctx.client.room_messages.return_value = response

    with (
        patch("mindroom.custom_tools.matrix_message.room_access_allowed", return_value=True),
        tool_runtime_context(ctx),
    ):
        payload = json.loads(
            await tool.matrix_message(action="read", room_id="!other:localhost"),
        )

    assert payload["status"] == "ok"
    assert payload["action"] == "read"
    assert "thread_id" not in payload
    ctx.client.room_messages.assert_awaited_once()


@pytest.mark.asyncio
async def test_matrix_message_cross_room_context_does_not_leak_thread() -> None:
    """Authorized cross-room context should not return the origin room's thread."""
    tool = MatrixMessageTools()
    ctx = _make_context(thread_id="$origin-thread:localhost", reply_to_event_id="$evt:localhost")

    with (
        patch("mindroom.custom_tools.matrix_message.room_access_allowed", return_value=True),
        tool_runtime_context(ctx),
    ):
        payload = json.loads(
            await tool.matrix_message(action="context", room_id="!other:localhost"),
        )

    assert payload["status"] == "ok"
    assert payload["thread_id"] is None
    assert payload["reply_to_event_id"] is None
