"""Tests for voice message handling functionality."""

from __future__ import annotations

import asyncio
import tempfile
from pathlib import Path
from typing import TYPE_CHECKING
from unittest.mock import AsyncMock, MagicMock, patch

import httpx
import nio
import pytest
from agno.media import Audio

from mindroom import voice_handler
from mindroom.config.agent import AgentConfig
from mindroom.config.main import Config
from mindroom.config.voice import VoiceConfig, VoiceSTTConfig, _VoiceLLMConfig
from mindroom.constants import ATTACHMENT_IDS_KEY, VOICE_RAW_AUDIO_FALLBACK_KEY
from mindroom.model_defaults import LOCAL_OPENAI_API_KEY_DEFAULT
from tests.conftest import bind_runtime_paths, runtime_paths_for, test_runtime_paths
from tests.identity_helpers import persist_actual_entity_accounts

if TYPE_CHECKING:
    from collections.abc import Callable

TEST_VOICE_ACCOUNT_PASSWORD = "pw"  # noqa: S105


def _runtime_bound_config(config: Config) -> Config:
    """Return a runtime-bound config for voice handler tests."""
    bound = bind_runtime_paths(config, test_runtime_paths(Path(tempfile.mkdtemp())))
    _persist_voice_handler_accounts(bound)
    return bound


def _persist_voice_handler_accounts(config: Config) -> None:
    runtime_paths = runtime_paths_for(config)
    persist_actual_entity_accounts(config, runtime_paths, password=TEST_VOICE_ACCOUNT_PASSWORD)


def _recording_stt_client(response_text: str) -> tuple[Callable[[], httpx.AsyncClient], list[httpx.Request]]:
    """Return an HTTP client factory and its captured STT requests."""
    requests: list[httpx.Request] = []
    async_client_type = httpx.AsyncClient

    def handle_request(request: httpx.Request) -> httpx.Response:
        requests.append(request)
        return httpx.Response(200, json={"text": response_text})

    transport = httpx.MockTransport(handle_request)

    def client_factory() -> httpx.AsyncClient:
        return async_client_type(transport=transport)

    return client_factory, requests


def _matrix_room(
    room_id: str,
    *,
    members: tuple[str, ...] = (),
    members_synced: bool = True,
) -> nio.MatrixRoom:
    room = nio.MatrixRoom(room_id=room_id, own_user_id="@mindroom_router:localhost")
    for member_id in members:
        room.add_member(member_id, None, None)
    room.members_synced = members_synced
    return room


async def _handle_voice_message(
    client: nio.AsyncClient,
    room: nio.MatrixRoom,
    event: nio.RoomMessageAudio | nio.RoomEncryptedAudio,
    config: Config,
    *,
    audio: Audio | None = None,
) -> str | None:
    """Run voice handling with the explicit runtime bound to the test config."""
    return await voice_handler._handle_voice_message(
        client,
        room,
        event,
        config,
        runtime_paths_for(config),
        audio=audio,
    )


async def _process_transcription(transcription: str, config: Config, **kwargs: object) -> str:
    """Run transcription processing with the explicit runtime bound to the test config."""
    return await voice_handler._process_transcription(
        transcription,
        config,
        runtime_paths_for(config),
        **kwargs,
    )


async def _prepare_voice_message(
    client: nio.AsyncClient,
    storage_path: Path,
    room: nio.MatrixRoom,
    event: nio.RoomMessageAudio | nio.RoomEncryptedAudio,
    config: Config,
    *,
    thread_id: str | None,
) -> voice_handler._PreparedVoiceMessage | None:
    """Prepare one voice message with the explicit runtime bound to the test config."""
    return await voice_handler.prepare_voice_message(
        client,
        storage_path,
        room,
        event,
        config,
        runtime_paths=runtime_paths_for(config),
        thread_id=thread_id,
    )


class TestVoiceHandler:
    """Test voice message handler functionality."""

    def test_voice_handler_disabled_by_default(self) -> None:
        """Test that voice handler is disabled when not configured."""
        config = Config()
        assert not config.voice.enabled
        assert config.voice.visible_router_echo

    def test_voice_handler_enabled_with_config(self) -> None:
        """Test that voice handler is enabled when configured."""
        config = _runtime_bound_config(
            Config(
                voice=VoiceConfig(
                    enabled=True,
                    stt=VoiceSTTConfig(provider="openai", model="whisper-1"),
                    intelligence=_VoiceLLMConfig(model="default"),
                ),
            ),
        )
        assert config.voice.enabled
        assert config.voice.stt.provider == "openai"
        assert config.voice.stt.model == "whisper-1"
        assert config.voice.intelligence.model == "default"

    def test_voice_stt_keeps_model_default_for_partial_config(self) -> None:
        """Promoting the shared speech config does not break existing partial voice STT blocks."""
        config = VoiceConfig.model_validate({"stt": {"provider": "openai"}})

        assert config.stt.model == "gpt-4o-transcribe"

    @pytest.mark.parametrize(
        ("matrix_mime_type", "expected_filename", "expected_mime_type"),
        [
            ("audio/mp4", "audio.m4a", "audio/mp4"),
            ("audio/m4a", "audio.m4a", "audio/m4a"),
            ("audio/x-m4a", "audio.m4a", "audio/x-m4a"),
            ("audio/mp3", "audio.mp3", "audio/mp3"),
            ("audio/webm", "audio.webm", "audio/webm"),
            ("audio/ogg; codecs=opus", "audio.ogg", "audio/ogg"),
            ("  AUDIO/WAV  ", "audio.wav", "audio/wav"),
        ],
    )
    def test_stt_upload_metadata_preserves_matrix_audio_mime_type(
        self,
        matrix_mime_type: str,
        expected_filename: str,
        expected_mime_type: str,
    ) -> None:
        """Matrix audio MIME types should be uploaded to STT with matching metadata."""
        filename, mime_type = voice_handler._stt_upload_filename_and_mime_type(matrix_mime_type)

        assert filename == expected_filename
        assert mime_type == expected_mime_type

    def test_stt_upload_metadata_defaults_to_ogg(self) -> None:
        """Missing Matrix MIME metadata should keep the legacy OGG upload shape."""
        filename, mime_type = voice_handler._stt_upload_filename_and_mime_type(None)

        assert filename == "audio.ogg"
        assert mime_type == "audio/ogg"

    def test_stt_upload_metadata_defaults_to_ogg_for_non_audio_mime_type(self) -> None:
        """Non-audio Matrix MIME metadata should keep the legacy OGG upload shape."""
        filename, mime_type = voice_handler._stt_upload_filename_and_mime_type("text/plain")

        assert filename == "audio.ogg"
        assert mime_type == "audio/ogg"

    def test_stt_upload_metadata_preserves_unknown_audio_mime_type(self) -> None:
        """Unknown audio MIME metadata should preserve the audio type with a generic extension."""
        filename, mime_type = voice_handler._stt_upload_filename_and_mime_type("audio/unknown")

        assert filename == "audio.bin"
        assert mime_type == "audio/unknown"

    @pytest.mark.asyncio
    async def test_transcribe_audio_uses_configured_stt_host_and_named_credential(self) -> None:
        """OpenAI proxy requests use their named credential and TLS verification."""
        config = _runtime_bound_config(
            Config(
                voice=VoiceConfig(
                    enabled=True,
                    stt=VoiceSTTConfig(
                        provider="openai_compatible",
                        host="https://stt.example.test/v1",
                        model="whisper-1",
                        credentials_service="openai-voice",
                        extra_kwargs={"language": "nl", "temperature": 0},
                    ),
                ),
            ),
        )
        client_init_kwargs: list[dict[str, object]] = []
        post_calls: list[dict[str, object]] = []

        class FakeResponse:
            status_code = 200

            def json(self) -> dict[str, str]:
                return {"text": " transcribed text "}

        class FakeAsyncClient:
            def __init__(self, **kwargs: object) -> None:
                client_init_kwargs.append(kwargs)

            async def __aenter__(self) -> FakeAsyncClient:
                return self

            async def __aexit__(self, *args: object) -> None:
                return None

            async def post(
                self,
                url: str,
                *,
                headers: dict[str, str],
                files: dict[str, tuple[str, bytes, str]],
                data: dict[str, str],
            ) -> FakeResponse:
                post_calls.append(
                    {
                        "url": url,
                        "headers": headers,
                        "files": files,
                        "data": data,
                    },
                )
                return FakeResponse()

        with (
            patch("mindroom.voice_handler.httpx.AsyncClient", FakeAsyncClient),
            patch("mindroom.voice_handler.get_api_key_for_service", return_value="openai-provider-key") as get_key,
        ):
            transcription = await voice_handler._transcribe_audio(
                b"audio-bytes",
                config,
                runtime_paths_for(config),
                mime_type="audio/ogg",
            )

        assert transcription == "transcribed text"
        get_key.assert_called_once_with("openai-voice", runtime_paths_for(config))
        assert len(client_init_kwargs) == 1
        assert client_init_kwargs[0].get("verify", True) is not False
        assert post_calls == [
            {
                "url": "https://stt.example.test/v1/audio/transcriptions",
                "headers": {"Authorization": "Bearer openai-provider-key"},
                "files": {"file": ("audio.ogg", b"audio-bytes", "audio/ogg")},
                "data": {"model": "whisper-1", "language": "nl", "temperature": 0},
            },
        ]

    @pytest.mark.asyncio
    async def test_transcribe_audio_uses_placeholder_for_keyless_openai_compatible_host(self) -> None:
        """Keyless custom STT requests should use the non-secret local placeholder."""
        config = _runtime_bound_config(
            Config(
                voice=VoiceConfig(
                    enabled=True,
                    stt=VoiceSTTConfig(
                        provider="openai_compatible",
                        host="http://localhost:10301",
                        model="large-v3",
                    ),
                ),
            ),
        )
        client_factory, requests = _recording_stt_client("local transcript")
        with (
            patch("mindroom.voice_handler.httpx.AsyncClient", side_effect=client_factory),
            patch("mindroom.voice_handler.get_api_key_for_service", return_value=None) as get_key,
        ):
            transcription = await voice_handler._transcribe_audio(
                b"audio-bytes",
                config,
                runtime_paths_for(config),
            )

        assert transcription == "local transcript"
        get_key.assert_not_called()
        assert len(requests) == 1
        assert str(requests[0].url) == "http://localhost:10301/v1/audio/transcriptions"
        assert requests[0].headers["Authorization"] == f"Bearer {LOCAL_OPENAI_API_KEY_DEFAULT}"

    @pytest.mark.asyncio
    async def test_transcribe_audio_rejects_keyless_cloud_openai(self) -> None:
        """Cloud OpenAI STT should not send a request without resolved credentials."""
        config = _runtime_bound_config(
            Config(
                voice=VoiceConfig(
                    enabled=True,
                    stt=VoiceSTTConfig(provider="openai", model="gpt-4o-transcribe"),
                ),
            ),
        )

        with (
            patch("mindroom.voice_handler.httpx.AsyncClient") as async_client,
            patch("mindroom.voice_handler.get_api_key_for_service", return_value=None) as get_key,
        ):
            transcription = await voice_handler._transcribe_audio(
                b"audio-bytes",
                config,
                runtime_paths_for(config),
            )

        assert transcription is None
        get_key.assert_called_once_with("openai", runtime_paths_for(config))
        async_client.assert_not_called()

    @pytest.mark.asyncio
    async def test_transcribe_audio_preserves_explicit_custom_api_key(self) -> None:
        """Custom STT requests should preserve an explicitly configured API key."""
        config = _runtime_bound_config(
            Config(
                voice=VoiceConfig(
                    enabled=True,
                    stt=VoiceSTTConfig(
                        provider="openai_compatible",
                        host="https://stt.example.test",
                        model="large-v3",
                        api_key="configured-test-placeholder",
                    ),
                ),
            ),
        )
        client_factory, requests = _recording_stt_client("custom transcript")
        with (
            patch("mindroom.voice_handler.httpx.AsyncClient", side_effect=client_factory),
            patch("mindroom.voice_handler.get_api_key_for_service") as get_key,
        ):
            transcription = await voice_handler._transcribe_audio(
                b"audio-bytes",
                config,
                runtime_paths_for(config),
            )

        assert transcription == "custom transcript"
        get_key.assert_not_called()
        assert len(requests) == 1
        assert requests[0].headers["Authorization"] == "Bearer configured-test-placeholder"

    def test_sanitize_unavailable_mentions_uses_exact_aliases(self) -> None:
        """Voice mention sanitizing should match exact Matrix mention aliases."""
        config = _runtime_bound_config(
            Config(
                agents={
                    "foo": AgentConfig(display_name="Foo"),
                    "mindroom_foo": AgentConfig(display_name="Prefixed Foo"),
                },
            ),
        )

        sanitized = voice_handler._sanitize_unavailable_mentions(
            "@mindroom_foo help @mindroom_mindroom_foo",
            allowed_entities={"foo"},
            config=config,
            runtime_paths=runtime_paths_for(config),
        )

        assert sanitized == "mindroom_foo help @mindroom_mindroom_foo"

    @pytest.mark.asyncio
    async def test_voice_handler_ignores_when_disabled(self) -> None:
        """Test that voice handler does nothing when disabled."""
        config = _runtime_bound_config(Config())

        # Mock objects
        client = AsyncMock()
        room = MagicMock()
        event = MagicMock()

        # Should return immediately without processing
        await _handle_voice_message(client, room, event, config)

        # Verify no processing occurred
        client.download.assert_not_called()

    @pytest.mark.asyncio
    async def test_process_transcription_basic(self) -> None:
        """Test basic transcription processing."""
        from mindroom.config.agent import AgentConfig, TeamConfig  # noqa: PLC0415

        config = _runtime_bound_config(
            Config(
                voice=VoiceConfig(enabled=True),
                agents={
                    "research": AgentConfig(display_name="ResearchAgent", role="Research agent"),
                    "code": AgentConfig(display_name="CodeAgent", role="Code agent"),
                },
                teams={
                    "dev_team": TeamConfig(
                        display_name="Development Team",
                        role="Dev team",
                        agents=["code"],
                    ),
                },
            ),
        )

        # Mock the AI model
        with patch("mindroom.voice_handler._process_transcription") as mock_process:
            mock_process.return_value = "@research help me with this"

            result = await _process_transcription("research help me with this", config)
            assert "@research" in result

    @pytest.mark.asyncio
    async def test_voice_handler_uses_room_scoped_entities_for_transcription(self) -> None:
        """Test voice transcription prompt is scoped to entities present in the room."""
        config = _runtime_bound_config(
            Config(
                voice=VoiceConfig(enabled=True),
                agents={
                    "openclaw": AgentConfig(display_name="OpenClaw Agent", role="OpenClaw role"),
                    "code": AgentConfig(display_name="Code Agent", role="Coding role"),
                },
            ),
        )

        client = AsyncMock()
        domain = config.get_domain(runtime_paths_for(config))
        room = _matrix_room(
            "!voice:localhost",
            members=(
                f"@actual_openclaw:{domain}",
                f"@actual_router:{domain}",
                "@alice:example.com",
            ),
        )
        event = MagicMock(spec=nio.RoomMessageAudio)
        event.event_id = "$voice"
        event.sender = "@alice:example.com"
        event.body = "voice.ogg"
        event.source = {"content": {"body": "voice.ogg"}}

        with (
            patch(
                "mindroom.voice_handler._download_audio",
                new=AsyncMock(return_value=Audio(content=b"audio", mime_type="audio/ogg")),
            ),
            patch("mindroom.voice_handler._transcribe_audio", return_value="help me") as mock_transcribe,
            patch("mindroom.voice_handler._process_transcription", new_callable=AsyncMock) as mock_process,
        ):
            mock_process.return_value = "@openclaw help me"
            result = await _handle_voice_message(client, room, event, config)

        assert result == "🎤 @openclaw help me"
        mock_transcribe.assert_awaited_once()
        assert mock_transcribe.await_args.kwargs["mime_type"] == "audio/ogg"
        assert mock_process.await_count == 1
        assert mock_process.await_args.kwargs["available_agent_names"] == ["openclaw"]
        assert mock_process.await_args.kwargs["available_team_names"] == []

    @pytest.mark.asyncio
    async def test_download_audio_unencrypted(self) -> None:
        """Test downloading unencrypted audio messages."""
        _runtime_bound_config(Config(voice=VoiceConfig(enabled=True)))  # Just to verify it works, not used in test

        # Mock client and event
        client = AsyncMock()
        event = MagicMock(spec=nio.RoomMessageAudio)

        with (
            patch(
                "mindroom.voice_handler.download_media_bytes",
                new=AsyncMock(return_value=b"audio_data"),
            ) as mock_download,
            patch("mindroom.voice_handler.media_mime_type", return_value="audio/ogg"),
        ):
            result = await voice_handler._download_audio(client, event)

        assert result is not None
        assert result.content == b"audio_data"
        assert result.mime_type == "audio/ogg"
        mock_download.assert_awaited_once_with(client, event)

    @pytest.mark.asyncio
    async def test_download_audio_encrypted(self) -> None:
        """Test downloading and decrypting encrypted audio messages."""
        _runtime_bound_config(Config(voice=VoiceConfig(enabled=True)))  # Just to verify it works, not used in test

        # Mock client and encrypted event
        client = AsyncMock()
        event = MagicMock(spec=nio.RoomEncryptedAudio)

        with (
            patch(
                "mindroom.voice_handler.download_media_bytes",
                new=AsyncMock(return_value=b"decrypted_audio_data"),
            ) as mock_download,
            patch("mindroom.voice_handler.media_mime_type", return_value="audio/mpeg"),
        ):
            result = await voice_handler._download_audio(client, event)

        assert result is not None
        assert result.content == b"decrypted_audio_data"
        assert result.mime_type == "audio/mpeg"
        mock_download.assert_awaited_once_with(client, event)

    @pytest.mark.asyncio
    async def test_download_audio_returns_none_when_media_cap_rejects_payload(self) -> None:
        """Voice downloads should inherit Matrix media byte caps from the shared helper."""
        client = AsyncMock()
        event = MagicMock(spec=nio.RoomMessageAudio)

        with patch("mindroom.voice_handler.download_media_bytes", new=AsyncMock(return_value=None)) as mock_download:
            result = await voice_handler._download_audio(client, event)

        assert result is None
        mock_download.assert_awaited_once_with(client, event)

    @pytest.mark.asyncio
    async def test_prepare_voice_message_clears_inflight_task_after_failed_download(self, tmp_path: Path) -> None:
        """Failed normalization should not leave stale in-flight task entries behind."""
        config = _runtime_bound_config(Config(authorization={"default_room_access": True}))
        client = AsyncMock()
        room = _matrix_room("!test:server", members=("@alice:example.com",))
        event = MagicMock(spec=nio.RoomMessageAudio)
        event.event_id = "$voice123"
        event.sender = "@alice:example.com"
        event.body = "voice.ogg"
        event.source = {"content": {"body": "voice.ogg"}}

        voice_handler._voice_normalization_cache.clear()
        voice_handler._voice_normalization_tasks.clear()
        cache_key = voice_handler._voice_cache_key(tmp_path, room.room_id, event.event_id, None)

        with patch("mindroom.voice_handler._download_audio", new=AsyncMock(return_value=None)):
            prepared = await _prepare_voice_message(
                client,
                tmp_path,
                room,
                event,
                config,
                thread_id=None,
            )

        assert prepared is None
        assert cache_key not in voice_handler._voice_normalization_tasks
        assert cache_key not in voice_handler._voice_normalization_cache

    @pytest.mark.asyncio
    async def test_prepare_voice_message_cancellation_does_not_cancel_shared_normalization(
        self,
        tmp_path: Path,
    ) -> None:
        """Canceling one waiter should not cancel the shared normalization task for others."""
        config = _runtime_bound_config(Config(authorization={"default_room_access": True}))
        client = AsyncMock()
        room = _matrix_room("!test:server", members=("@alice:example.com",))
        event = MagicMock(spec=nio.RoomMessageAudio)
        event.event_id = "$voice123"
        event.sender = "@alice:example.com"
        event.body = "voice.ogg"
        event.source = {"content": {"body": "voice.ogg"}}

        voice_handler._voice_normalization_cache.clear()
        voice_handler._voice_normalization_tasks.clear()
        cache_key = voice_handler._voice_cache_key(tmp_path, room.room_id, event.event_id, None)
        started = asyncio.Event()
        release = asyncio.Event()
        compute_calls = 0

        async def fake_compute(
            *_args: object,
            **_kwargs: object,
        ) -> voice_handler._NormalizedVoiceMessage:
            nonlocal compute_calls
            compute_calls += 1
            started.set()
            await release.wait()
            return voice_handler._NormalizedVoiceMessage(
                attachment_id="att-123",
                transcribed_message="🎤 hello",
            )

        with patch("mindroom.voice_handler._compute_normalized_voice_message", side_effect=fake_compute):
            first_waiter = asyncio.create_task(
                _prepare_voice_message(
                    client,
                    tmp_path,
                    room,
                    event,
                    config,
                    thread_id=None,
                ),
            )
            await started.wait()

            second_waiter = asyncio.create_task(
                _prepare_voice_message(
                    client,
                    tmp_path,
                    room,
                    event,
                    config,
                    thread_id=None,
                ),
            )
            await asyncio.sleep(0)

            first_waiter.cancel()
            with pytest.raises(asyncio.CancelledError):
                await first_waiter

            release.set()
            prepared = await second_waiter

        assert prepared is not None
        assert prepared.text == "🎤 hello"
        assert prepared.source["content"][ATTACHMENT_IDS_KEY] == ["att-123"]
        assert compute_calls == 1
        assert cache_key in voice_handler._voice_normalization_cache
        assert cache_key not in voice_handler._voice_normalization_tasks

    @pytest.mark.asyncio
    async def test_process_transcription_returns_raw_transcription_when_normalizer_llm_hangs(
        self,
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        """A hung normalizer LLM call must fail open to the raw transcription."""
        config = _runtime_bound_config(
            Config(
                voice=VoiceConfig(enabled=True),
                agents={"code": AgentConfig(display_name="CodeAgent", role="Code agent")},
            ),
        )
        monkeypatch.setattr(voice_handler, "_VOICE_NORMALIZER_LLM_TIMEOUT_SECONDS", 0.05, raising=False)
        arun_started = asyncio.Event()

        class HungAgent:
            def __init__(self, *_args: object, **_kwargs: object) -> None:
                pass

            async def arun(self, *_args: object, **_kwargs: object) -> None:
                arun_started.set()
                await asyncio.Event().wait()

        with (
            patch("mindroom.voice_handler.model_loading.get_model_instance", return_value=MagicMock()),
            patch("mindroom.voice_handler.Agent", HungAgent),
        ):
            result = await asyncio.wait_for(
                _process_transcription("turn on the lights", config),
                timeout=2.0,
            )

        assert arun_started.is_set()
        assert result == "turn on the lights"

    @pytest.mark.asyncio
    async def test_normalize_voice_message_fails_when_normalization_hangs(
        self,
        monkeypatch: pytest.MonkeyPatch,
        tmp_path: Path,
    ) -> None:
        """A hung download must fail the shared normalization task instead of wedging waiters."""
        config = _runtime_bound_config(Config(voice=VoiceConfig(enabled=True)))
        monkeypatch.setattr(voice_handler, "_VOICE_NORMALIZATION_TOTAL_TIMEOUT_SECONDS", 0.05, raising=False)
        client = AsyncMock()
        room = _matrix_room("!test:server", members=("@alice:example.com",))
        event = MagicMock(spec=nio.RoomMessageAudio)
        event.event_id = "$hung_voice"
        event.sender = "@alice:example.com"
        event.body = "voice.ogg"
        event.source = {"content": {"body": "voice.ogg"}}

        voice_handler._voice_normalization_cache.clear()
        voice_handler._voice_normalization_tasks.clear()

        async def hung_download(*_args: object, **_kwargs: object) -> None:
            await asyncio.Event().wait()

        try:
            with patch("mindroom.voice_handler._download_audio", side_effect=hung_download):
                with pytest.raises(TimeoutError):
                    await asyncio.wait_for(
                        voice_handler._normalize_voice_message(
                            client,
                            tmp_path,
                            room,
                            event,
                            config,
                            runtime_paths_for(config),
                            thread_id=None,
                        ),
                        timeout=2.0,
                    )
                await asyncio.sleep(0.01)
                assert voice_handler._voice_normalization_tasks == {}
        finally:
            for task in voice_handler._voice_normalization_tasks.values():
                task.cancel()
            voice_handler._voice_normalization_tasks.clear()

    @pytest.mark.asyncio
    async def test_prepare_raw_voice_fallback_message_times_out_hung_download(
        self,
        monkeypatch: pytest.MonkeyPatch,
        tmp_path: Path,
    ) -> None:
        """A hung fallback download must degrade to text instead of wedging the ingress lane."""
        config = _runtime_bound_config(Config(voice=VoiceConfig(enabled=True)))
        monkeypatch.setattr(voice_handler, "_VOICE_NORMALIZATION_TOTAL_TIMEOUT_SECONDS", 0.05, raising=False)
        client = AsyncMock()
        room = _matrix_room("!test:server", members=("@alice:example.com",))
        event = MagicMock(spec=nio.RoomMessageAudio)
        event.event_id = "$hung_fallback"
        event.sender = "@alice:example.com"
        event.body = "voice.ogg"
        event.server_timestamp = 1_000
        event.source = {"content": {"body": "voice.ogg", "msgtype": "m.audio"}}
        download_started = asyncio.Event()

        async def hung_download(*_args: object, **_kwargs: object) -> None:
            download_started.set()
            await asyncio.Event().wait()

        with patch("mindroom.voice_handler._download_audio", side_effect=hung_download):
            prepared = await asyncio.wait_for(
                voice_handler.prepare_raw_voice_fallback_message(
                    client,
                    tmp_path,
                    room,
                    event,
                    config,
                    runtime_paths=runtime_paths_for(config),
                    thread_id=None,
                ),
                timeout=2.0,
            )

        assert download_started.is_set()
        assert prepared.source["content"][VOICE_RAW_AUDIO_FALLBACK_KEY] is True
        assert ATTACHMENT_IDS_KEY not in prepared.source["content"]
