"""Tests for centralized message content extraction with large message support."""

import json
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, patch

import nio
import pytest
from nio import crypto

import mindroom.matrix.message_content as message_content_module
from mindroom.config.agent import AgentConfig
from mindroom.config.main import Config
from mindroom.constants import STREAM_STATUS_KEY, STREAM_WARMUP_SUFFIX_KEY, RuntimePaths
from mindroom.entity_resolution import entity_identity_registry
from mindroom.matrix.cache.event_cache_events import event_mxc_urls
from mindroom.matrix.cache.sqlite_event_cache import SqliteEventCache
from mindroom.matrix.client_visible_messages import (
    extract_visible_edit_body,
    message_preview,
    resolve_visible_event_source,
    thread_root_body_preview,
)
from mindroom.matrix.membership_fence import UNCERTIFIED_MEMBERSHIP_EPOCH
from mindroom.matrix.message_content import (
    SidecarHydrationBatch,
    _download_mxc_text,
    extract_and_resolve_message,
    extract_edit_body,
    prepare_sidecar_hydration_batch,
    resolve_event_source_content,
)
from mindroom.matrix.sidecar_content import sidecar_mxc_url
from mindroom.matrix.state import MatrixState
from mindroom.matrix.visible_body import (
    strip_matrix_rich_reply_fallback,
    visible_body_from_event_source,
    visible_content_from_content,
)
from tests.conftest import bind_runtime_paths, make_matrix_client_mock, runtime_paths_for, test_runtime_paths
from tests.identity_helpers import persist_entity_accounts


def _trusted_entity_sender_ids(config: Config, runtime_paths: RuntimePaths) -> frozenset[str]:
    return entity_identity_registry(config, runtime_paths)._internal_sender_ids


def _make_message_event(
    *,
    body: str,
    content: dict[str, object],
    event_id: str = "$event",
    sender: str = "@alice:example.com",
    timestamp_ms: int = 1234567890,
) -> nio.RoomMessageText:
    """Create a Matrix text event for message content tests."""
    event = nio.RoomMessageText(
        source={
            "content": content,
            "event_id": event_id,
            "sender": sender,
            "origin_server_ts": timestamp_ms,
            "type": "m.room.message",
        },
        body=body,
        formatted_body=None,
        format=None,
    )
    event.sender = sender
    return event


def _make_client() -> AsyncMock:
    """Return one AsyncClient-shaped test mock with a local agent user ID."""
    return make_matrix_client_mock(user_id="@mindroom_general:localhost")


class TestResolvedMessageExtraction:
    """Tests for coherent visible message extraction."""

    def test_download_and_durable_ownership_share_sidecar_url_validation(self) -> None:
        """Hydration and reference tracking must choose the same valid MXC URL."""
        content = {
            "body": "Preview body",
            "msgtype": "m.file",
            "io.mindroom.long_text": {
                "version": 2,
                "encoding": "matrix_event_content_json",
            },
            "url": "https://example.test/not-mxc",
            "file": {"url": "mxc://server/encrypted-sidecar"},
        }

        assert sidecar_mxc_url(content) == "mxc://server/encrypted-sidecar"
        assert event_mxc_urls({"content": content}) == frozenset({"mxc://server/encrypted-sidecar"})

    @pytest.mark.parametrize("malformed_url", ["mxc://", "mxc://server"])
    def test_sidecar_url_validation_rejects_incomplete_mxc_uris(self, malformed_url: str) -> None:
        """Incomplete content URIs must not enter hydration or ownership indexes."""
        metadata = {
            "version": 2,
            "encoding": "matrix_event_content_json",
        }
        direct_content = {
            "io.mindroom.long_text": metadata,
            "url": malformed_url,
        }
        encrypted_content = {
            "io.mindroom.long_text": metadata,
            "file": {"url": malformed_url},
        }

        assert sidecar_mxc_url(direct_content) is None
        assert sidecar_mxc_url(encrypted_content) is None
        assert event_mxc_urls({"content": direct_content}) == frozenset()
        assert event_mxc_urls({"content": encrypted_content}) == frozenset()

    @pytest.mark.asyncio
    async def test_extract_and_resolve_message_hydrates_v2_sidecar_content(self) -> None:
        """Regular v2 sidecars should return the canonical content and body."""
        original_content = {
            "msgtype": "m.text",
            "body": "Full response body",
            "io.mindroom.tool_trace": {"version": 1, "events": [{"tool": "shell"}]},
        }
        event = _make_message_event(
            body="Preview body",
            content={
                "msgtype": "m.file",
                "body": "Preview body",
                "info": {"mimetype": "application/json"},
                "io.mindroom.long_text": {
                    "version": 2,
                    "encoding": "matrix_event_content_json",
                },
                "url": "mxc://server/sidecar",
            },
        )
        client = _make_client()
        client.download = AsyncMock(
            return_value=MagicMock(
                spec=nio.DownloadResponse,
                body=json.dumps(original_content).encode("utf-8"),
            ),
        )
        event_cache = AsyncMock()

        resolved = await extract_and_resolve_message(
            event,
            client,
            event_cache=event_cache,
            room_id="!room:localhost",
        )

        assert resolved["body"] == "Full response body"
        assert resolved["content"] == original_content
        event_cache.store_event.assert_not_awaited()
        event_cache.get_mxc_text.assert_not_awaited()
        event_cache.store_mxc_text.assert_not_awaited()

    @pytest.mark.asyncio
    async def test_extract_and_resolve_message_hydrates_v2_edit_wrapper(self) -> None:
        """Edit-sidecar events should resolve to the canonical outer replacement payload."""
        canonical_content = {
            "msgtype": "m.text",
            "body": "* Full edit body",
            "m.new_content": {
                "msgtype": "m.text",
                "body": "Full edit body",
                "io.mindroom.tool_trace": {"version": 1, "events": [{"tool": "shell"}]},
            },
            "m.relates_to": {"rel_type": "m.replace", "event_id": "$original"},
        }
        event = _make_message_event(
            body="* Preview edit",
            content={
                "msgtype": "m.text",
                "body": "* Preview edit",
                "m.new_content": {
                    "msgtype": "m.file",
                    "body": "Preview edit",
                    "info": {"mimetype": "application/json"},
                    "io.mindroom.long_text": {
                        "version": 2,
                        "encoding": "matrix_event_content_json",
                    },
                    "url": "mxc://server/edit-sidecar",
                },
                "m.relates_to": {"rel_type": "m.replace", "event_id": "$original"},
            },
        )
        client = _make_client()
        client.download = AsyncMock(
            return_value=MagicMock(
                spec=nio.DownloadResponse,
                body=json.dumps(canonical_content).encode("utf-8"),
            ),
        )

        resolved = await extract_and_resolve_message(event, client)

        assert resolved["body"] == "* Full edit body"
        assert resolved["content"] == canonical_content
        assert resolved["content"]["body"] == resolved["body"]

    @pytest.mark.asyncio
    async def test_extract_edit_body_hydrates_v2_edit_sidecar(self) -> None:
        """Edit extraction should return the canonical m.new_content from a v2 sidecar."""
        canonical_content = {
            "msgtype": "m.text",
            "body": "* Full edit body",
            "m.new_content": {
                "msgtype": "m.text",
                "body": "Full edit body",
                "io.mindroom.tool_trace": {"version": 1, "events": [{"tool": "shell"}]},
            },
            "m.relates_to": {"rel_type": "m.replace", "event_id": "$original"},
        }
        client = _make_client()
        client.download = AsyncMock(
            return_value=MagicMock(
                spec=nio.DownloadResponse,
                body=json.dumps(canonical_content).encode("utf-8"),
            ),
        )

        body, content = await extract_edit_body(
            {
                "content": {
                    "msgtype": "m.text",
                    "body": "* Preview edit",
                    "m.new_content": {
                        "msgtype": "m.file",
                        "body": "Preview edit",
                        "info": {"mimetype": "application/json"},
                        "io.mindroom.long_text": {
                            "version": 2,
                            "encoding": "matrix_event_content_json",
                        },
                        "url": "mxc://server/edit-sidecar",
                    },
                    "m.relates_to": {"rel_type": "m.replace", "event_id": "$original"},
                },
            },
            client,
        )

        assert body == "Full edit body"
        assert content == canonical_content["m.new_content"]

    @pytest.mark.asyncio
    async def test_extract_and_resolve_message_leaves_legacy_v1_preview_untouched(self) -> None:
        """Unsupported v1 sidecars should stay on the preview payload without download."""
        event = _make_message_event(
            body="Preview body",
            content={
                "msgtype": "m.file",
                "body": "Preview body",
                "io.mindroom.long_text": {
                    "version": 1,
                    "original_size": 100000,
                },
                "url": "mxc://server/legacy-sidecar",
            },
        )
        client = _make_client()
        client.download = AsyncMock()

        resolved = await extract_and_resolve_message(event, client)

        assert resolved["body"] == "Preview body"
        assert resolved["content"]["body"] == "Preview body"
        client.download.assert_not_called()

    @pytest.mark.asyncio
    async def test_extract_edit_body_leaves_legacy_v1_preview_untouched(self) -> None:
        """Unsupported v1 edit sidecars should keep the preview body/content coherent."""
        client = _make_client()
        client.download = AsyncMock()

        body, content = await extract_edit_body(
            {
                "content": {
                    "msgtype": "m.text",
                    "body": "* Preview edit",
                    "m.new_content": {
                        "msgtype": "m.file",
                        "body": "Preview edit",
                        "io.mindroom.long_text": {
                            "version": 1,
                            "original_size": 100000,
                        },
                        "url": "mxc://server/legacy-edit-sidecar",
                    },
                    "m.relates_to": {"rel_type": "m.replace", "event_id": "$original"},
                },
            },
            client,
        )

        assert body == "Preview edit"
        assert content == {
            "msgtype": "m.file",
            "body": "Preview edit",
            "io.mindroom.long_text": {
                "version": 1,
                "original_size": 100000,
            },
            "url": "mxc://server/legacy-edit-sidecar",
        }
        client.download.assert_not_called()

    @pytest.mark.asyncio
    async def test_resolve_event_source_content_hydrates_v2_edit_payload(self) -> None:
        """Event-source hydration should expose canonical edit metadata for mention routing."""
        canonical_content = {
            "msgtype": "m.text",
            "body": "* @agent full edit",
            "m.new_content": {
                "msgtype": "m.text",
                "body": "@agent full edit",
                "m.mentions": {"user_ids": ["@mindroom_agent:example.com"]},
            },
            "m.relates_to": {"rel_type": "m.replace", "event_id": "$original"},
        }
        client = _make_client()
        client.download = AsyncMock(
            return_value=MagicMock(
                spec=nio.DownloadResponse,
                body=json.dumps(canonical_content).encode("utf-8"),
            ),
        )

        event_source = await resolve_event_source_content(
            {
                "content": {
                    "msgtype": "m.text",
                    "body": "* Preview edit",
                    "m.new_content": {
                        "msgtype": "m.file",
                        "body": "Preview edit",
                        "info": {"mimetype": "application/json"},
                        "io.mindroom.long_text": {
                            "version": 2,
                            "encoding": "matrix_event_content_json",
                        },
                        "url": "mxc://server/context-edit-sidecar",
                    },
                    "m.relates_to": {"rel_type": "m.replace", "event_id": "$original"},
                },
            },
            client,
        )

        assert event_source["content"] == canonical_content

    def test_visible_body_from_event_source_prefers_visible_edit_content(self) -> None:
        """Visible-body extraction should use m.new_content when present."""
        event_source = {
            "content": {
                "msgtype": "m.text",
                "body": "* Preview edit",
                "m.new_content": {
                    "msgtype": "m.text",
                    "body": "Full edit body",
                },
            },
        }

        assert visible_body_from_event_source(event_source, "* Preview edit") == "Full edit body"

    def test_visible_body_from_event_source_prefers_canonical_stream_body(self) -> None:
        """Visible-body extraction should prefer canonical stream text over transient warmup suffixes."""
        event_source = {
            "sender": "@mindroom_general:localhost",
            "content": {
                "msgtype": "m.text",
                "body": "hello\n\n⏳ Preparing isolated worker...",
                "io.mindroom.visible_body": "hello",
            },
        }

        assert (
            visible_body_from_event_source(
                event_source,
                "hello",
                trusted_sender_ids={"@mindroom_general:localhost"},
            )
            == "hello"
        )

    def test_visible_body_from_event_source_uses_explicit_warmup_suffix_metadata(self) -> None:
        """Trusted streamed previews may remove only the exact suffix that was explicitly appended."""
        warmup_suffix = "⏳ Preparing isolated worker..."
        event_source = {
            "sender": "@mindroom_general:localhost",
            "content": {
                "msgtype": "m.text",
                "body": f"hello\n\n{warmup_suffix}",
                STREAM_WARMUP_SUFFIX_KEY: warmup_suffix,
            },
        }

        assert (
            visible_body_from_event_source(
                event_source,
                "hello",
                trusted_sender_ids={"@mindroom_general:localhost"},
            )
            == "hello"
        )

    def test_visible_body_from_event_source_ignores_empty_canonical_stream_body(self) -> None:
        """Empty canonical stream metadata should fall back to the actual Matrix body."""
        event_source = {
            "sender": "@mindroom_general:localhost",
            "content": {
                "msgtype": "m.text",
                "body": "Thinking...",
                "io.mindroom.visible_body": "",
            },
        }

        assert visible_body_from_event_source(
            event_source,
            "Thinking...",
            trusted_sender_ids={"@mindroom_general:localhost"},
        ) == ("Thinking...")

    def test_visible_body_from_event_source_ignores_untrusted_visible_body(self) -> None:
        """Untrusted inbound events should not override the real Matrix body via visible_body."""
        event_source = {
            "sender": "@mindroom_fake:localhost",
            "content": {
                "msgtype": "m.text",
                "body": "benign body",
                "io.mindroom.visible_body": "spoofed body",
            },
        }

        assert visible_body_from_event_source(
            event_source,
            "benign body",
            trusted_sender_ids={"@mindroom_general:localhost"},
        ) == ("benign body")

    def test_visible_body_from_event_source_does_not_strip_literal_status_text_without_explicit_metadata(self) -> None:
        """Legitimate final content should stay intact when no explicit warmup metadata is present."""
        event_source = {
            "sender": "@mindroom_general:localhost",
            "content": {
                "msgtype": "m.text",
                "body": "Diagnosis follows\n\n⚠️ Worker startup failed for shell.run: intentional example.",
                STREAM_STATUS_KEY: "completed",
            },
        }

        assert visible_body_from_event_source(
            event_source,
            "Diagnosis follows",
            trusted_sender_ids={"@mindroom_general:localhost"},
        ) == ("Diagnosis follows\n\n⚠️ Worker startup failed for shell.run: intentional example.")

    def test_strip_matrix_rich_reply_fallback_removes_quoted_prefix(self) -> None:
        """Rich-reply denial reasons should keep only the user-authored reply body."""
        body = "> <@alice:localhost> Approval required\n> quoted details\n\nNo, too risky."

        assert strip_matrix_rich_reply_fallback(body) == "No, too risky."

    def test_strip_matrix_rich_reply_fallback_allows_empty_reply_body(self) -> None:
        """Quote-only rich replies should not preserve the Matrix fallback."""
        body = "> <@alice:localhost> Approval required\n> quoted details\n\n"

        assert strip_matrix_rich_reply_fallback(body) == ""

    def test_strip_matrix_rich_reply_fallback_leaves_plain_quotes_alone(self) -> None:
        """Quoted text without the Matrix blank separator is normal message content."""
        body = "> keep this quoted line\nNo Matrix rich-reply separator"

        assert strip_matrix_rich_reply_fallback(body) == body

    def test_visible_content_from_content_prefers_replacement_content(self) -> None:
        """Matrix edit content unwrapping should be shared across consumers."""
        content = {
            "body": "* old",
            "m.new_content": {"body": "new", "status": "expired"},
        }

        assert visible_content_from_content(content) == {"body": "new", "status": "expired"}

    def test_visible_body_from_event_source_ignores_removed_agent_sender_ids(self, tmp_path: Path) -> None:
        """Removed managed senders must not keep overriding canonical-body metadata."""
        config = bind_runtime_paths(
            Config(agents={"general": AgentConfig(display_name="General Agent")}),
            test_runtime_paths(tmp_path),
        )
        runtime_paths = runtime_paths_for(config)
        persist_entity_accounts(config, runtime_paths)
        state = MatrixState.load(runtime_paths=runtime_paths)
        state.add_account("agent_removed", "mindroom_removed", "pw", domain="legacy.example.com")
        state.save(runtime_paths=runtime_paths)

        event_source = {
            "sender": "@mindroom_removed:legacy.example.com",
            "content": {
                "msgtype": "m.text",
                "body": "hello\n\n⏳ Preparing isolated worker...",
                "io.mindroom.visible_body": "hello",
            },
        }

        assert (
            visible_body_from_event_source(
                event_source,
                "hello",
                trusted_sender_ids=_trusted_entity_sender_ids(config, runtime_paths),
            )
            == "hello\n\n⏳ Preparing isolated worker..."
        )

    def test_visible_body_from_event_source_trusts_persisted_runtime_usernames(self, tmp_path: Path) -> None:
        """Persisted current usernames should stay trusted on the current runtime domain."""
        config = bind_runtime_paths(
            Config(agents={"general": AgentConfig(display_name="General Agent")}),
            test_runtime_paths(tmp_path),
        )
        runtime_paths = runtime_paths_for(config)
        persist_entity_accounts(config, runtime_paths, usernames={"general": "mindroom_general_oldns"})
        state = MatrixState.load(runtime_paths=runtime_paths)
        state.add_account("agent_general", "mindroom_general_oldns", "pw", domain=config.get_domain(runtime_paths))
        state.save(runtime_paths=runtime_paths)
        current_domain = config.get_domain(runtime_paths)

        event_source = {
            "sender": f"@mindroom_general_oldns:{current_domain}",
            "content": {
                "msgtype": "m.text",
                "body": "hello\n\n⏳ Preparing isolated worker...",
                "io.mindroom.visible_body": "hello",
            },
        }

        assert (
            visible_body_from_event_source(
                event_source,
                "hello",
                trusted_sender_ids=_trusted_entity_sender_ids(config, runtime_paths),
            )
            == "hello"
        )

    def test_visible_body_from_event_source_ignores_previous_persisted_sender_ids(self, tmp_path: Path) -> None:
        """Earlier persisted usernames must not stay trusted after a rename."""
        config = bind_runtime_paths(
            Config(agents={"general": AgentConfig(display_name="General Agent")}),
            test_runtime_paths(tmp_path),
        )
        runtime_paths = runtime_paths_for(config)
        persist_entity_accounts(config, runtime_paths, usernames={"general": "mindroom_general_v2"})
        state = MatrixState.load(runtime_paths=runtime_paths)
        state.add_account("agent_general", "mindroom_general_v1", "pw", domain="legacy.example.com")
        state.add_account("agent_general", "mindroom_general_v2", "pw", domain=config.get_domain(runtime_paths))
        state.save(runtime_paths=runtime_paths)

        event_source = {
            "sender": f"@mindroom_general_v1:{config.get_domain(runtime_paths)}",
            "content": {
                "msgtype": "m.text",
                "body": "hello\n\n⏳ Preparing isolated worker...",
                "io.mindroom.visible_body": "hello",
            },
        }

        assert (
            visible_body_from_event_source(
                event_source,
                "hello",
                trusted_sender_ids=_trusted_entity_sender_ids(config, runtime_paths),
            )
            == "hello\n\n⏳ Preparing isolated worker..."
        )

    @pytest.mark.asyncio
    async def test_resolve_visible_event_source_trusts_runtime_sender_ids(self, tmp_path: Path) -> None:
        """High-level visible-source resolution should derive trust from runtime config."""
        config = bind_runtime_paths(
            Config(agents={"general": AgentConfig(display_name="General Agent")}),
            test_runtime_paths(tmp_path),
        )
        runtime_paths = runtime_paths_for(config)
        persist_entity_accounts(
            config,
            runtime_paths,
            usernames={"router": "mindroom_router", "general": "mindroom_general"},
        )
        current_domain = config.get_domain(runtime_paths)
        event_source = {
            "sender": f"@mindroom_general:{current_domain}",
            "content": {
                "msgtype": "m.text",
                "body": "hello\n\n⏳ Preparing isolated worker...",
                "io.mindroom.visible_body": "hello",
            },
        }

        resolved_source, visible_body = await resolve_visible_event_source(
            event_source,
            None,
            fallback_body="hello",
            config=config,
            runtime_paths=runtime_paths,
        )

        assert resolved_source == event_source
        assert visible_body == "hello"

    @pytest.mark.asyncio
    async def test_extract_visible_edit_body_trusts_runtime_sender_ids(self, tmp_path: Path) -> None:
        """High-level edit extraction should derive trusted visible-body rules from runtime config."""
        config = bind_runtime_paths(
            Config(agents={"general": AgentConfig(display_name="General Agent")}),
            test_runtime_paths(tmp_path),
        )
        runtime_paths = runtime_paths_for(config)
        persist_entity_accounts(
            config,
            runtime_paths,
            usernames={"router": "mindroom_router", "general": "mindroom_general"},
        )
        current_domain = config.get_domain(runtime_paths)

        body, content = await extract_visible_edit_body(
            {
                "sender": f"@mindroom_general:{current_domain}",
                "content": {
                    "msgtype": "m.text",
                    "body": "* Preview edit",
                    "m.new_content": {
                        "msgtype": "m.text",
                        "body": "hello\n\n⏳ Preparing isolated worker...",
                        "io.mindroom.visible_body": "hello",
                    },
                    "m.relates_to": {"rel_type": "m.replace", "event_id": "$original"},
                },
            },
            None,
            config=config,
            runtime_paths=runtime_paths,
        )

        assert body == "hello"
        assert content == {
            "msgtype": "m.text",
            "body": "hello",
            "io.mindroom.visible_body": "hello",
        }

    @pytest.mark.asyncio
    async def test_thread_root_body_preview_uses_runtime_sender_ids_for_bundled_edits(
        self,
        tmp_path: Path,
    ) -> None:
        """Thread previews should resolve trusted bundled edits without raw trusted-sender plumbing."""
        config = bind_runtime_paths(
            Config(agents={"general": AgentConfig(display_name="General Agent")}),
            test_runtime_paths(tmp_path),
        )
        runtime_paths = runtime_paths_for(config)
        persist_entity_accounts(
            config,
            runtime_paths,
            usernames={"router": "mindroom_router", "general": "mindroom_general"},
        )
        current_domain = config.get_domain(runtime_paths)
        event = _make_message_event(
            body="Original root",
            content={"msgtype": "m.text", "body": "Original root"},
            event_id="$thread-root",
            sender="@user:example.com",
        )
        event.source["unsigned"] = {
            "m.relations": {
                "m.replace": {
                    "event_id": "$thread-root-edit",
                    "sender": f"@mindroom_general:{current_domain}",
                    "origin_server_ts": 2000,
                    "type": "m.room.message",
                    "content": {
                        "body": "* Edited body\n\n⏳ Preparing isolated worker...",
                        "msgtype": "m.text",
                        "m.new_content": {
                            "body": "Edited body\n\n⏳ Preparing isolated worker...",
                            "msgtype": "m.text",
                            "io.mindroom.visible_body": "Edited body",
                        },
                        "m.relates_to": {"rel_type": "m.replace", "event_id": "$thread-root"},
                    },
                },
            },
        }

        preview = await thread_root_body_preview(
            event,
            client=_make_client(),
            config=config,
            runtime_paths=runtime_paths,
        )

        assert preview == "Edited body"

    @pytest.mark.asyncio
    async def test_thread_root_body_preview_passes_precomputed_trusted_sender_ids_to_nested_helpers(
        self,
        tmp_path: Path,
    ) -> None:
        """Thread previews should reuse one caller-provided trust set through nested helpers."""
        config = bind_runtime_paths(
            Config(agents={"general": AgentConfig(display_name="General Agent")}),
            test_runtime_paths(tmp_path),
        )
        runtime_paths = runtime_paths_for(config)
        persist_entity_accounts(
            config,
            runtime_paths,
            usernames={"router": "mindroom_router", "general": "mindroom_general"},
        )
        event = _make_message_event(
            body="Original root",
            content={"msgtype": "m.text", "body": "Original root"},
            event_id="$thread-root",
            sender="@user:example.com",
        )
        client = _make_client()
        trusted_sender_ids = frozenset({"@mindroom_general:localhost"})

        with (
            patch(
                "mindroom.matrix.client_visible_messages.bundled_replacement_body",
                new=AsyncMock(return_value=None),
            ) as mock_bundled,
            patch(
                "mindroom.matrix.client_visible_messages.resolve_visible_event_source",
                new=AsyncMock(return_value=(event.source, "Resolved root")),
            ) as mock_resolve,
        ):
            preview = await thread_root_body_preview(
                event,
                client=client,
                config=config,
                runtime_paths=runtime_paths,
                trusted_sender_ids=trusted_sender_ids,
            )

        assert preview == "Resolved root"
        mock_bundled.assert_awaited_once_with(
            event.source,
            client=client,
            config=config,
            runtime_paths=runtime_paths,
            event_cache=None,
            room_id=None,
            trusted_sender_ids=trusted_sender_ids,
        )
        mock_resolve.assert_awaited_once_with(
            event.source,
            client,
            fallback_body="Original root",
            config=config,
            runtime_paths=runtime_paths,
            event_cache=None,
            room_id=None,
            trusted_sender_ids=trusted_sender_ids,
        )

    def test_message_preview_compacts_whitespace_and_truncates(self) -> None:
        """Shared preview compaction should live in the Matrix visible-message layer."""
        assert message_preview("  alpha   beta  \n gamma  ", max_length=12) == "alpha bet..."


class TestSidecarHydrationBatch:
    """Tests for request-scoped batched sidecar hydration."""

    @pytest.mark.asyncio
    async def test_batch_reference_miss_skips_redundant_point_read(self) -> None:
        """A reference covered by the batch may download directly after a proven cache miss."""
        client = AsyncMock()
        response = MagicMock(spec=nio.DownloadResponse)
        response.body = b"fresh"
        client.download.return_value = response
        event_cache = AsyncMock()
        event_cache.get_mxc_text.return_value = "stale"
        reference = ("$event", "mxc://server/covered")
        hydration_batch = SidecarHydrationBatch(
            cached_texts={},
            references=frozenset({reference}),
            owner_event_ids=frozenset({"$event"}),
        )

        assert (
            await _download_mxc_text(
                client,
                reference[1],
                event_cache=event_cache,
                room_id="!room:localhost",
                event_id=reference[0],
                expected_membership_epoch=1,
                hydration_batch=hydration_batch,
            )
            == "fresh"
        )
        event_cache.get_mxc_text.assert_not_awaited()

    @pytest.mark.asyncio
    async def test_reference_outside_batch_keeps_point_read_fallback(self) -> None:
        """An uncovered bundled event must retain the exact point-cache lookup."""
        client = AsyncMock()
        client.download.side_effect = AssertionError("point-cache hit must not download")
        event_cache = AsyncMock()
        event_cache.get_mxc_text.return_value = "cached"
        hydration_batch = SidecarHydrationBatch(
            cached_texts={},
            references=frozenset({("$other", "mxc://server/other")}),
            owner_event_ids=frozenset({"$other"}),
        )

        assert (
            await _download_mxc_text(
                client,
                "mxc://server/uncovered",
                event_cache=event_cache,
                room_id="!room:localhost",
                event_id="$event",
                expected_membership_epoch=1,
                hydration_batch=hydration_batch,
            )
            == "cached"
        )
        event_cache.get_mxc_text.assert_awaited_once_with(
            "!room:localhost",
            "$event",
            "mxc://server/uncovered",
        )


@pytest.mark.asyncio
async def test_sidecar_batch_only_skips_registration_for_verified_owners() -> None:
    """A read-only batch must not claim ownership for plaintext cache misses."""
    hit_reference = ("$hit", "mxc://server/hit")
    miss_reference = ("$miss", "mxc://server/miss")
    sources = [
        {
            "event_id": event_id,
            "content": {
                "msgtype": "m.file",
                "body": "preview",
                "url": mxc_url,
                "io.mindroom.long_text": {
                    "version": 2,
                    "encoding": "matrix_event_content_json",
                },
            },
        }
        for event_id, mxc_url in (hit_reference, miss_reference)
    ]
    event_cache = AsyncMock()
    event_cache.get_mxc_texts.return_value = {hit_reference: "cached"}

    hydration_batch = await prepare_sidecar_hydration_batch(
        sources,
        event_cache=event_cache,
        room_id="!room:localhost",
        expected_membership_epoch=1,
        register_owners=False,
    )

    assert hydration_batch is not None
    assert hydration_batch.references == frozenset({hit_reference, miss_reference})
    assert hydration_batch.owner_event_ids == frozenset({"$hit"})


class TestDownloadMxcText:
    """Tests for _download_mxc_text function."""

    @pytest.mark.asyncio
    async def test_invalid_mxc_url(self) -> None:
        """Test handling of invalid MXC URL."""
        client = AsyncMock()
        result = await _download_mxc_text(client, "http://not-mxc-url")
        assert result is None

    @pytest.mark.asyncio
    async def test_malformed_mxc_url(self) -> None:
        """Test handling of malformed MXC URL."""
        client = AsyncMock()
        result = await _download_mxc_text(client, "mxc://no-media-id")
        assert result is None

    @pytest.mark.asyncio
    async def test_successful_download(self) -> None:
        """Test successful text download."""
        client = AsyncMock()
        client.user_id = "@alice:localhost"
        response = MagicMock(spec=nio.DownloadResponse)
        response.body = b"Downloaded text content"
        client.download.return_value = response

        result = await _download_mxc_text(
            client,
            "mxc://server/media123",
            room_id="!room:localhost",
            event_id="$event",
        )
        assert result == "Downloaded text content"
        client.download.assert_called_once_with(mxc="mxc://server/media123")
        assert (
            await _download_mxc_text(
                client,
                "mxc://server/media123",
                room_id="!room:localhost",
                event_id="$event",
            )
            == "Downloaded text content"
        )
        assert client.download.await_count == 2

    @pytest.mark.asyncio
    async def test_successful_encrypted_download(self) -> None:
        """Encrypted sidecars should use the JWK key value emitted by nio."""
        plaintext = b"Downloaded encrypted text content"
        encrypted, file_info = crypto.attachments.encrypt_attachment(plaintext)
        client = AsyncMock()
        response = MagicMock(spec=nio.DownloadResponse)
        response.body = encrypted
        client.download.return_value = response

        assert await _download_mxc_text(client, "mxc://server/encrypted", file_info) == plaintext.decode()

    @pytest.mark.asyncio
    async def test_hydration_without_durable_cache_does_not_reuse_plaintext(self) -> None:
        """Hydration without room and event identity returns text without retaining it."""
        client = AsyncMock()
        client.user_id = "@alice:localhost"
        first_response = MagicMock(spec=nio.DownloadResponse)
        first_response.body = b"first"
        second_response = MagicMock(spec=nio.DownloadResponse)
        second_response.body = b"second"
        client.download.side_effect = [first_response, second_response]

        assert await _download_mxc_text(client, "mxc://server/incomplete") == "first"
        assert await _download_mxc_text(client, "mxc://server/incomplete") == "second"
        assert client.download.await_count == 2

    @pytest.mark.asyncio
    async def test_download_returns_text_without_caching_when_durable_store_fails(self) -> None:
        """A cache outage must not suppress freshly downloaded visible content."""
        client = AsyncMock()
        client.user_id = "@alice:localhost"
        first_response = MagicMock(spec=nio.DownloadResponse)
        first_response.body = b"first"
        second_response = MagicMock(spec=nio.DownloadResponse)
        second_response.body = b"second"
        client.download.side_effect = [first_response, second_response]
        event_cache = AsyncMock()
        event_cache.principal_id = "@alice:localhost"
        event_cache.get_mxc_text.return_value = None
        event_cache.store_mxc_text.side_effect = RuntimeError("cache unavailable")
        kwargs = {
            "event_cache": event_cache,
            "room_id": "!room:localhost",
            "event_id": "$event",
        }

        assert await _download_mxc_text(client, "mxc://server/outage", **kwargs) == "first"
        assert await _download_mxc_text(client, "mxc://server/outage", **kwargs) == "second"
        assert client.download.await_count == 2

    @pytest.mark.asyncio
    async def test_download_returns_fresh_text_without_disabled_advisory_cache(self) -> None:
        """A permanent cache disable must not suppress authenticated fresh content."""
        principal_id = "@alice:localhost"
        room_id = "!room:localhost"
        event_id = "$event"
        mxc_url = "mxc://server/cache-disabled"
        client = AsyncMock()
        client.user_id = principal_id
        response = MagicMock(spec=nio.DownloadResponse)
        response.body = b"fresh plaintext"
        client.download.return_value = response
        event_cache = AsyncMock()
        event_cache.principal_id = principal_id
        event_cache.durable_writes_available = False
        event_cache.get_mxc_text.return_value = None
        event_cache.store_mxc_text.return_value = False

        assert (
            await _download_mxc_text(
                client,
                mxc_url,
                event_cache=event_cache,
                room_id=room_id,
                event_id=event_id,
            )
            == "fresh plaintext"
        )

    @pytest.mark.asyncio
    @pytest.mark.parametrize("membership_epoch", [None, UNCERTIFIED_MEMBERSHIP_EPOCH])
    async def test_uncertified_refill_bypasses_durable_plaintext_cache(self, membership_epoch: int | None) -> None:
        """Missing durable membership proof must use fresh plaintext without caching it."""
        client = AsyncMock()
        response = MagicMock(spec=nio.DownloadResponse)
        response.body = b"fresh plaintext"
        client.download.return_value = response
        event_cache = AsyncMock()
        event_cache.get_mxc_text.return_value = "stale plaintext"

        assert (
            await _download_mxc_text(
                client,
                "mxc://server/uncertified",
                event_cache=event_cache,
                room_id="!room:localhost",
                event_id="$event",
                expected_membership_epoch=membership_epoch,
            )
            == "fresh plaintext"
        )
        event_cache.get_mxc_text.assert_not_awaited()
        event_cache.store_mxc_text.assert_not_awaited()
        client.download.assert_awaited_once_with(mxc="mxc://server/uncertified")

    @pytest.mark.asyncio
    async def test_departed_room_fence_rejects_late_sidecar_until_rejoin(self, tmp_path: Path) -> None:
        """Late hydration cannot repopulate plaintext after leave, while a real rejoin can."""
        principal_id = "@alice:localhost"
        room_id = "!left:localhost"
        event_id = "$sidecar"
        mxc_url = "mxc://server/departed"
        event = {
            "event_id": event_id,
            "sender": principal_id,
            "origin_server_ts": 1,
            "type": "m.room.message",
            "content": {
                "body": "preview",
                "msgtype": "m.file",
                "url": mxc_url,
                "io.mindroom.long_text": {
                    "version": 2,
                    "encoding": "matrix_event_content_json",
                },
            },
        }
        root = SqliteEventCache(tmp_path / "event-cache.db")
        await root.initialize()
        cache = root.for_principal(principal_id)
        await cache.store_event(event_id, room_id, event)
        initial_membership_epoch = await cache.room_membership_epoch(room_id)
        assert initial_membership_epoch is not None
        await cache.purge_room(room_id)
        client = AsyncMock()
        response = MagicMock(spec=nio.DownloadResponse)
        response.body = b"departed plaintext"
        client.download.return_value = response
        try:
            assert (
                await _download_mxc_text(
                    client,
                    mxc_url,
                    event_cache=cache,
                    room_id=room_id,
                    event_id=event_id,
                    expected_membership_epoch=initial_membership_epoch,
                )
                is None
            )
            assert await cache.get_mxc_text(room_id, event_id, mxc_url) is None

            await cache.mark_room_joined(
                room_id,
                expected_departure_epoch=cache.room_departure_epoch(room_id),
            )
            await cache.store_event(event_id, room_id, event)
            rejoined_membership_epoch = await cache.room_membership_epoch(room_id)
            assert rejoined_membership_epoch is not None

            assert (
                await _download_mxc_text(
                    client,
                    mxc_url,
                    event_cache=cache,
                    room_id=room_id,
                    event_id=event_id,
                    expected_membership_epoch=rejoined_membership_epoch,
                )
                == "departed plaintext"
            )
        finally:
            await root.close()

    @pytest.mark.asyncio
    async def test_durable_hit_returns_authorized_plaintext_without_rewriting(self) -> None:
        """A durable ownership-joined cache hit needs no second authorization write."""
        principal_id = "@alice:localhost"
        room_id = "!room:localhost"
        event_id = "$event"
        mxc_url = "mxc://server/redacted-during-read"
        client = AsyncMock()
        client.user_id = principal_id
        event_cache = AsyncMock()
        event_cache.principal_id = principal_id

        event_cache.get_mxc_text.return_value = "cached plaintext"

        assert (
            await _download_mxc_text(
                client,
                mxc_url,
                event_cache=event_cache,
                room_id=room_id,
                event_id=event_id,
                expected_membership_epoch=7,
            )
            == "cached plaintext"
        )
        event_cache.store_mxc_text.assert_not_awaited()
        client.download.assert_not_awaited()

    @pytest.mark.asyncio
    async def test_active_event_cache_without_event_identity_does_not_suppress_download(self) -> None:
        """An incomplete owner may hydrate for this call but cannot populate either cache."""
        client = AsyncMock()
        client.user_id = "@alice:localhost"
        first_response = MagicMock(spec=nio.DownloadResponse)
        first_response.body = b"first"
        second_response = MagicMock(spec=nio.DownloadResponse)
        second_response.body = b"second"
        client.download.side_effect = [first_response, second_response]
        event_cache = AsyncMock()
        event_cache.principal_id = "@alice:localhost"

        assert (
            await _download_mxc_text(
                client,
                "mxc://server/incomplete-owner",
                event_cache=event_cache,
                room_id="!room:localhost",
            )
            == "first"
        )
        assert (
            await _download_mxc_text(
                client,
                "mxc://server/incomplete-owner",
                event_cache=event_cache,
                room_id="!room:localhost",
            )
            == "second"
        )
        event_cache.get_mxc_text.assert_not_awaited()
        event_cache.store_mxc_text.assert_not_awaited()
        assert client.download.await_count == 2

    @pytest.mark.asyncio
    async def test_download_failure(self) -> None:
        """Test handling of download failure."""
        client = AsyncMock()
        client.download.return_value = MagicMock(spec=nio.DownloadError)

        result = await _download_mxc_text(client, "mxc://server/media123")
        assert result is None

    @pytest.mark.asyncio
    async def test_download_rejects_plaintext_over_byte_limit(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """Oversized sidecar bytes should not be decoded, cached, or persisted."""
        monkeypatch.setattr(message_content_module, "_MXC_TEXT_MAX_BYTES", 5)
        client = AsyncMock()
        response = MagicMock(spec=nio.DownloadResponse)
        response.body = b"123456"
        client.download.return_value = response
        event_cache = AsyncMock()
        event_cache.principal_id = "@alice:localhost"
        event_cache.get_mxc_text.return_value = None

        result = await _download_mxc_text(
            client,
            "mxc://server/oversized",
            event_cache=event_cache,
            room_id="!room:server",
            event_id="$oversized",
        )

        assert result is None
        event_cache.store_mxc_text.assert_not_awaited()

    @pytest.mark.asyncio
    async def test_download_rejects_encrypted_sidecar_over_byte_limit_before_decrypt(
        self,
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        """Oversized encrypted sidecars should be rejected before decryption allocates plaintext."""
        monkeypatch.setattr(message_content_module, "_MXC_TEXT_MAX_BYTES", 5)
        client = AsyncMock()
        response = MagicMock(spec=nio.DownloadResponse)
        response.body = b"123456"
        client.download.return_value = response
        file_info = {"key": {"k": "key"}, "hashes": {"sha256": "hash"}, "iv": "iv"}

        with patch("mindroom.matrix.message_content.crypto.attachments.decrypt_attachment") as mock_decrypt:
            result = await _download_mxc_text(client, "mxc://server/encrypted-oversized", file_info)

        assert result is None
        mock_decrypt.assert_not_called()

    @pytest.mark.asyncio
    async def test_download_rejects_decrypted_sidecar_over_byte_limit(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """Decrypted sidecar bytes should be capped before UTF-8 decode and JSON parsing."""
        monkeypatch.setattr(message_content_module, "_MXC_TEXT_MAX_BYTES", 5)
        client = AsyncMock()
        response = MagicMock(spec=nio.DownloadResponse)
        response.body = b"small"
        client.download.return_value = response
        file_info = {"key": {"k": "key"}, "hashes": {"sha256": "hash"}, "iv": "iv"}
        event_cache = AsyncMock()
        event_cache.principal_id = "@alice:localhost"
        event_cache.get_mxc_text.return_value = None

        with patch("mindroom.matrix.message_content.crypto.attachments.decrypt_attachment", return_value=b"123456"):
            result = await _download_mxc_text(
                client,
                "mxc://server/decrypted-oversized",
                file_info,
                event_cache=event_cache,
                room_id="!room:server",
                event_id="$decrypted-oversized",
            )

        assert result is None


class TestCanonicalContentResolution:
    """Tests for sidecar-backed canonical content extraction."""

    @pytest.mark.asyncio
    async def test_extract_and_resolve_message_hydrates_v2_content_metadata(self) -> None:
        """Large-message v2 previews should resolve canonical content keys from the sidecar."""
        client = AsyncMock()
        response = MagicMock(spec=nio.DownloadResponse)
        response.body = b'{"body":"Full body","msgtype":"m.text","io.mindroom.tool_trace":{"version":1,"events":[{"tool":"shell"}]}}'
        client.download.return_value = response
        event = nio.RoomMessageText.from_dict(
            {
                "content": {
                    "msgtype": "m.file",
                    "body": "Preview...",
                    "info": {"mimetype": "application/json"},
                    "io.mindroom.long_text": {"version": 2, "encoding": "matrix_event_content_json"},
                    "io.mindroom.ai_run": {"version": 1, "run_id": "run-preview"},
                    "url": "mxc://server/file-json",
                },
                "event_id": "$event",
                "sender": "@agent:example.com",
                "origin_server_ts": 123,
                "type": "m.room.message",
                "room_id": "!room:example.com",
            },
        )

        result = await extract_and_resolve_message(event, client)

        assert result["body"] == "Full body"
        assert result["content"]["io.mindroom.tool_trace"] == {"version": 1, "events": [{"tool": "shell"}]}
        assert "io.mindroom.long_text" not in result["content"]

    @pytest.mark.asyncio
    async def test_extract_edit_body_hydrates_v2_sidecar_new_content(self) -> None:
        """Edit extraction should use canonical m.new_content from a v2 sidecar payload."""
        client = AsyncMock()
        response = MagicMock(spec=nio.DownloadResponse)
        response.body = (
            b'{"msgtype":"m.text","body":"* Full edit wrapper","m.new_content":{"body":"Full edit body","msgtype":"m.text",'
            b'"io.mindroom.tool_trace":{"version":1,"events":[{"tool":"web_search"}]}}}'
        )
        client.download.return_value = response
        event_source = {
            "content": {
                "body": "* Preview edit",
                "msgtype": "m.text",
                "m.new_content": {
                    "body": "Preview edit...",
                    "msgtype": "m.file",
                    "info": {"mimetype": "application/json"},
                    "io.mindroom.long_text": {"version": 2, "encoding": "matrix_event_content_json"},
                    "url": "mxc://server/edit-json",
                },
                "m.relates_to": {"rel_type": "m.replace", "event_id": "$original"},
            },
        }

        body, resolved_content = await extract_edit_body(event_source, client)

        assert body == "Full edit body"
        assert resolved_content == {
            "body": "Full edit body",
            "msgtype": "m.text",
            "io.mindroom.tool_trace": {"version": 1, "events": [{"tool": "web_search"}]},
        }

    @pytest.mark.asyncio
    async def test_extract_edit_body_prefers_canonical_stream_body(self) -> None:
        """Edit extraction should drop transient warmup suffixes when canonical stream text is present."""
        event_source = {
            "sender": "@mindroom_general:localhost",
            "content": {
                "body": "* hello",
                "msgtype": "m.text",
                "m.new_content": {
                    "body": "hello\n\n⏳ Preparing isolated worker...",
                    "msgtype": "m.text",
                    "io.mindroom.visible_body": "hello",
                },
                "m.relates_to": {"rel_type": "m.replace", "event_id": "$original"},
            },
        }

        body, resolved_content = await extract_edit_body(
            event_source,
            trusted_sender_ids={"@mindroom_general:localhost"},
        )

        assert body == "hello"
        assert resolved_content == {
            "body": "hello",
            "msgtype": "m.text",
            "io.mindroom.visible_body": "hello",
        }

    @pytest.mark.asyncio
    async def test_extract_edit_body_ignores_untrusted_visible_body(self) -> None:
        """Edit extraction should not trust canonical-body overrides from arbitrary room senders."""
        event_source = {
            "sender": "@alice:localhost",
            "content": {
                "body": "* hello",
                "msgtype": "m.text",
                "m.new_content": {
                    "body": "hello\n\n⏳ Preparing isolated worker...",
                    "msgtype": "m.text",
                    "io.mindroom.visible_body": "hello",
                },
                "m.relates_to": {"rel_type": "m.replace", "event_id": "$original"},
            },
        }

        body, resolved_content = await extract_edit_body(
            event_source,
            trusted_sender_ids={"@mindroom_general:localhost"},
        )

        assert body == "hello\n\n⏳ Preparing isolated worker..."
        assert resolved_content == {
            "body": "hello\n\n⏳ Preparing isolated worker...",
            "msgtype": "m.text",
            "io.mindroom.visible_body": "hello",
        }

    @pytest.mark.asyncio
    async def test_extract_edit_body_preserves_explicit_empty_string_body(self) -> None:
        """Edit extraction should keep explicit empty-string bodies instead of dropping the edit."""
        event_source = {
            "sender": "@mindroom_general:localhost",
            "content": {
                "body": "* Preview edit",
                "msgtype": "m.text",
                "m.new_content": {
                    "body": "",
                    "msgtype": "m.text",
                },
                "m.relates_to": {"rel_type": "m.replace", "event_id": "$original"},
            },
        }

        body, resolved_content = await extract_edit_body(
            event_source,
            trusted_sender_ids={"@mindroom_general:localhost"},
        )

        assert body == ""
        assert resolved_content == {
            "body": "",
            "msgtype": "m.text",
        }


class TestExtractAndResolveMessage:
    """Tests for extracted read/thread payload formatting."""

    @pytest.mark.asyncio
    async def test_text_message_includes_msgtype(self) -> None:
        """Plain text messages should preserve their Matrix msgtype."""
        event = nio.RoomMessageText.from_dict(
            {
                "type": "m.room.message",
                "event_id": "$text",
                "sender": "@alice:localhost",
                "origin_server_ts": 1,
                "content": {"msgtype": "m.text", "body": "hello"},
            },
        )

        result = await extract_and_resolve_message(event)

        assert result == {
            "sender": "@alice:localhost",
            "body": "hello",
            "timestamp": 1,
            "event_id": "$text",
            "content": {"msgtype": "m.text", "body": "hello"},
            "msgtype": "m.text",
        }

    @pytest.mark.asyncio
    async def test_notice_message_includes_msgtype(self) -> None:
        """Notices should expose msgtype so callers can distinguish them from text."""
        event = nio.RoomMessageNotice.from_dict(
            {
                "type": "m.room.message",
                "event_id": "$notice",
                "sender": "@mindroom:localhost",
                "origin_server_ts": 2,
                "content": {"msgtype": "m.notice", "body": "Compacted 12 messages"},
            },
        )

        result = await extract_and_resolve_message(event)

        assert result == {
            "sender": "@mindroom:localhost",
            "body": "Compacted 12 messages",
            "timestamp": 2,
            "event_id": "$notice",
            "content": {"msgtype": "m.notice", "body": "Compacted 12 messages"},
            "msgtype": "m.notice",
        }

    @pytest.mark.asyncio
    async def test_extract_and_resolve_message_prefers_canonical_body_for_trusted_edit_event(self) -> None:
        """Trusted local agent edit events should resolve to canonical body text."""
        event = nio.RoomMessageText.from_dict(
            {
                "type": "m.room.message",
                "event_id": "$edit",
                "sender": "@mindroom_general:localhost",
                "origin_server_ts": 3,
                "content": {
                    "msgtype": "m.text",
                    "body": "* hello",
                    "m.new_content": {
                        "msgtype": "m.text",
                        "body": "hello\n\n⏳ Preparing isolated worker...",
                        "io.mindroom.visible_body": "hello",
                        STREAM_STATUS_KEY: "streaming",
                    },
                    "m.relates_to": {"rel_type": "m.replace", "event_id": "$original"},
                },
            },
        )

        result = await extract_and_resolve_message(
            event,
            trusted_sender_ids={"@mindroom_general:localhost"},
        )

        assert result["body"] == "hello"

    @pytest.mark.asyncio
    async def test_extract_and_resolve_message_ignores_spoofed_visible_body(self) -> None:
        """Arbitrary inbound events should not override the real body via visible_body."""
        event = nio.RoomMessageText.from_dict(
            {
                "type": "m.room.message",
                "event_id": "$spoof",
                "sender": "@alice:localhost",
                "origin_server_ts": 4,
                "content": {
                    "msgtype": "m.text",
                    "body": "benign body",
                    "io.mindroom.visible_body": "spoofed body",
                },
            },
        )

        result = await extract_and_resolve_message(
            event,
            trusted_sender_ids={"@mindroom_general:localhost"},
        )

        assert result["body"] == "benign body"
