"""Tests for live debounce-based message coalescing."""

from __future__ import annotations

import asyncio
import json
import time
from typing import TYPE_CHECKING, Literal, cast, get_args
from unittest.mock import AsyncMock, MagicMock, patch

import nio
import pytest
from pydantic import ValidationError
from structlog.testing import capture_logs

from mindroom.attachments import _attachment_id_for_event, load_attachment, register_local_attachment
from mindroom.bot import AgentBot
from mindroom.coalescing import CoalescingGate, ReadyPendingEvent, is_coalescing_exempt_source_kind
from mindroom.coalescing_batch import (
    CoalescedBatch,
    CoalescingKey,
    PendingEvent,
    build_coalesced_batch,
)
from mindroom.config.agent import AgentConfig
from mindroom.config.auth import AuthorizationConfig
from mindroom.config.main import Config
from mindroom.config.models import DefaultsConfig, ModelConfig
from mindroom.constants import (
    ATTACHMENT_IDS_KEY,
    HOOK_MESSAGE_RECEIVED_DEPTH_KEY,
    ORIGINAL_SENDER_KEY,
    SKIP_MENTIONS_KEY,
    SOURCE_KIND_KEY,
    STREAM_STATUS_KEY,
    STREAM_STATUS_PENDING,
    STREAM_STATUS_STREAMING,
    VISIBLE_ROUTER_VOICE_ECHO_KEY,
    VOICE_RAW_AUDIO_FALLBACK_KEY,
    VOICE_TRANSCRIPT_KEY,
)
from mindroom.conversation_resolver import MessageContext
from mindroom.dispatch_callback_outcome import TurnDispatchOutcome
from mindroom.dispatch_handoff import (
    DispatchEvent,
    DispatchIngressMetadata,
    DispatchPayloadMetadata,
    PendingDispatchMetadata,
    PreparedTextEvent,
    _build_batch_dispatch_event,
    build_dispatch_handoff,
)
from mindroom.dispatch_obligations import DispatchCallbackKind
from mindroom.dispatch_replay_guard import has_newer_unresponded_in_thread
from mindroom.dispatch_source import (
    ACTIVE_THREAD_FOLLOW_UP_SOURCE_KIND,
    IMAGE_SOURCE_KIND,
    MESSAGE_SOURCE_KIND,
    TRUSTED_INTERNAL_RELAY_SOURCE_KIND,
    VOICE_SOURCE_KIND,
)
from mindroom.handled_turns import SourceEventMetadata, TurnRecord
from mindroom.hooks import MessageEnvelope
from mindroom.inbound_turn_normalizer import (
    BatchMediaAttachmentRequest,
    DispatchPayload,
    DispatchPayloadWithAttachmentsRequest,
    _BatchMediaAttachmentResult,
)
from mindroom.matrix.cache import ThreadHistoryResult
from mindroom.matrix.client import ResolvedVisibleMessage
from mindroom.matrix.event_info import EventInfo
from mindroom.matrix.identity import MatrixID
from mindroom.matrix.thread_diagnostics import (
    THREAD_HISTORY_DEGRADED_DIAGNOSTIC,
    THREAD_HISTORY_ERROR_DIAGNOSTIC,
    THREAD_HISTORY_SOURCE_DEGRADED,
    THREAD_HISTORY_SOURCE_DIAGNOSTIC,
)
from mindroom.matrix.users import AgentMatrixUser
from mindroom.message_target import MessageTarget
from mindroom.response_payload_preparation import ResponsePayloadPreparer
from mindroom.turn_controller import _IngressAdmissionOutcome, _PrecheckedEvent
from mindroom.turn_policy import PreparedDispatch, _DispatchPlan
from tests.conftest import (
    TEST_PASSWORD,
    bind_runtime_paths,
    dispatch_context_result,
    install_generate_response_mock,
    install_send_response_mock,
    make_conversation_cache_mock,
    make_matrix_client_mock,
    message_origin,
    prepare_payload_via_seam,
    prepared_dispatch_result,
    replace_turn_controller_deps,
    runtime_paths_for,
    test_runtime_paths,
    unwrap_extracted_collaborator,
    wrap_extracted_collaborators,
)

if TYPE_CHECKING:
    from collections.abc import Callable, Sequence
    from pathlib import Path


def _coalescing_gate_is_idle(gate: CoalescingGate) -> bool:
    return not gate._gates


def _make_config(
    tmp_path: Path,
    *,
    debounce_ms: int = 10,
) -> Config:
    """Build a config with configurable live coalescing timings."""
    return bind_runtime_paths(
        Config(
            agents={"test_agent": AgentConfig(display_name="TestAgent", rooms=["!room:localhost"])},
            teams={},
            models={"default": ModelConfig(provider="test", id="test-model")},
            defaults=DefaultsConfig(
                coalescing={
                    "debounce_ms": debounce_ms,
                },
            ),
            authorization=AuthorizationConfig(default_room_access=True),
        ),
        test_runtime_paths(tmp_path),
    )


def _make_bot(
    tmp_path: Path,
    *,
    debounce_ms: int = 10,
    agent_name: str = "test_agent",
) -> AgentBot:
    """Create a bot instance wired to a temporary runtime root."""
    config = _make_config(tmp_path, debounce_ms=debounce_ms)
    agent_user = AgentMatrixUser(
        agent_name=agent_name,
        password=TEST_PASSWORD,
        display_name="TestAgent",
        user_id=f"@mindroom_{agent_name}:localhost",
    )
    bot = AgentBot(agent_user, tmp_path, config, runtime_paths_for(config), rooms=["!room:localhost"])
    bot.client = make_matrix_client_mock(user_id=agent_user.user_id)
    wrap_extracted_collaborators(bot)
    replace_turn_controller_deps(
        bot,
        turn_policy=bot._turn_policy,
        delivery_gateway=bot._delivery_gateway,
        response_runner=bot._response_runner,
        resolver=bot._conversation_resolver,
        normalizer=bot._inbound_turn_normalizer,
        state_writer=bot._conversation_state_writer,
    )
    return bot


async def _enqueue_for_dispatch(
    bot: AgentBot,
    event: DispatchEvent,
    room: nio.MatrixRoom,
    *,
    source_kind: str,
    requester_user_id: str,
    dispatch_policy_source_kind: str | None = None,
    hook_source: str | None = None,
    message_received_depth: int = 0,
    coalescing_key: CoalescingKey | None = None,
    trust_internal_payload_metadata: bool | None = None,
) -> _IngressAdmissionOutcome:
    """Test helper for the reserved Matrix-ingress enqueue path."""
    reservation_owner = bot._turn_controller.reserve_prompt_ingress_order(room, requester_user_id)
    try:
        return await bot._turn_controller._enqueue_for_dispatch(
            event,
            room,
            reservation_owner=reservation_owner,
            source_kind=source_kind,
            requester_user_id=requester_user_id,
            dispatch_policy_source_kind=dispatch_policy_source_kind,
            hook_source=hook_source,
            message_received_depth=message_received_depth,
            coalescing_key=coalescing_key,
            trust_internal_payload_metadata=trust_internal_payload_metadata,
        )
    finally:
        await reservation_owner.release()


async def _admit_ready(
    gate: CoalescingGate,
    key: CoalescingKey,
    pending_event: PendingEvent,
) -> None:
    """Admit one already-ready event through the canonical gate API."""
    await gate.admit(
        key,
        source_event_id=pending_event.event.event_id,
        source_kind=pending_event.source_kind,
        ready_result=ReadyPendingEvent(pending_event=pending_event),
    )


def test_coalescing_config_rejects_removed_enabled_flag(tmp_path: Path) -> None:
    """Reject the removed defaults.coalescing.enabled toggle."""
    with pytest.raises(ValidationError, match="enabled"):
        Config.validate_with_runtime(
            {
                "agents": {"test_agent": {"display_name": "TestAgent"}},
                "models": {"default": {"provider": "test", "id": "test-model"}},
                "defaults": {"coalescing": {"enabled": True}},
            },
            test_runtime_paths(tmp_path),
        )


def _respond_dispatch_plan(action: object | None = None) -> _DispatchPlan:
    """Return a plan that continues into the response executor path."""
    return _DispatchPlan(
        kind="respond",
        response_action=action or MagicMock(kind="individual"),
    )


def _handled_turn_source_event_ids(handled_turn: TurnRecord | None) -> list[str]:
    """Return source event IDs from one handled-turn carrier for test assertions."""
    return list(handled_turn.source_event_ids) if handled_turn is not None else []


def _make_room(room_id: str = "!room:localhost") -> MagicMock:
    room = MagicMock(spec=nio.MatrixRoom)
    room.room_id = room_id
    room.canonical_alias = None
    return room


@pytest.mark.asyncio
async def test_duplicate_delivery_is_claimed_before_dispatch_resolution(tmp_path: Path) -> None:
    """A replay arriving during cache resolution must not repeat that work."""
    bot = _make_bot(tmp_path, debounce_ms=0)
    room = _make_room()
    event = _text_event(event_id="$duplicate", body="@test_agent hello")
    resolution_started = asyncio.Event()
    duplicate_reservation = MagicMock()

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

    normalizer = MagicMock()
    normalizer.resolve_text_event = AsyncMock(side_effect=slow_normalize)
    controller = replace_turn_controller_deps(bot, normalizer=normalizer)
    first = asyncio.create_task(
        controller._dispatch_text_message(room, event, "@user:localhost"),
    )
    await resolution_started.wait()
    await controller._dispatch_text_message(
        room,
        event,
        "@user:localhost",
        queued_notice_reservation=duplicate_reservation,
    )
    first.cancel()
    with pytest.raises(asyncio.CancelledError):
        await first

    normalizer.resolve_text_event.assert_awaited_once()
    duplicate_reservation.cancel.assert_called_once_with()


async def _wait_for(condition: Callable[[], bool], *, deadline_seconds: float = 0.5) -> None:
    """Poll until a test condition becomes true."""
    ready = asyncio.Event()
    loop = asyncio.get_running_loop()

    def _mark_ready() -> None:
        if condition():
            ready.set()
            return
        loop.call_later(0.001, _mark_ready)

    _mark_ready()
    try:
        async with asyncio.timeout(deadline_seconds):
            await ready.wait()
    except TimeoutError as exc:
        msg = "Timed out waiting for async test condition"
        raise AssertionError(msg) from exc


def _text_event(
    *,
    event_id: str,
    body: str,
    sender: str = "@user:localhost",
    server_timestamp: int = 1000,
    thread_id: str | None = None,
    source_kind: str | None = None,
    original_sender: str | None = None,
) -> nio.RoomMessageText:
    """Build a synthetic inbound text event for coalescing tests."""
    content: dict[str, object] = {
        "msgtype": "m.text",
        "body": body,
    }
    if thread_id is not None:
        content["m.relates_to"] = {"rel_type": "m.thread", "event_id": thread_id}
    if source_kind is not None:
        content[SOURCE_KIND_KEY] = source_kind
    if original_sender is not None:
        content[ORIGINAL_SENDER_KEY] = original_sender
    return cast(
        "nio.RoomMessageText",
        nio.RoomMessageText.from_dict(
            {
                "event_id": event_id,
                "sender": sender,
                "origin_server_ts": server_timestamp,
                "room_id": "!room:localhost",
                "type": "m.room.message",
                "content": content,
            },
        ),
    )


def _reply_event(
    *,
    event_id: str,
    body: str,
    reply_to_event_id: str,
    sender: str = "@user:localhost",
    server_timestamp: int = 1000,
) -> nio.RoomMessageText:
    """Build a synthetic inbound plain reply event for coalescing tests."""
    return cast(
        "nio.RoomMessageText",
        nio.RoomMessageText.from_dict(
            {
                "event_id": event_id,
                "sender": sender,
                "origin_server_ts": server_timestamp,
                "room_id": "!room:localhost",
                "type": "m.room.message",
                "content": {
                    "msgtype": "m.text",
                    "body": body,
                    "m.relates_to": {
                        "m.in_reply_to": {"event_id": reply_to_event_id},
                    },
                },
            },
        ),
    )


def _image_event(
    *,
    event_id: str,
    body: str = "photo.jpg",
    sender: str = "@user:localhost",
    server_timestamp: int = 1000,
    thread_id: str | None = None,
    source_kind: str | None = None,
) -> nio.RoomMessageImage:
    """Build a synthetic inbound image event for coalescing tests."""
    content: dict[str, object] = {
        "msgtype": "m.image",
        "body": body,
        "filename": body,
        "url": "mxc://localhost/test-image",
        "info": {"mimetype": "image/jpeg"},
    }
    if thread_id is not None:
        content["m.relates_to"] = {"rel_type": "m.thread", "event_id": thread_id}
    if source_kind is not None:
        content[SOURCE_KIND_KEY] = source_kind
    return cast(
        "nio.RoomMessageImage",
        nio.RoomMessageImage.from_dict(
            {
                "event_id": event_id,
                "sender": sender,
                "origin_server_ts": server_timestamp,
                "room_id": "!room:localhost",
                "type": "m.room.message",
                "content": content,
            },
        ),
    )


def _file_event(
    *,
    event_id: str,
    body: str = "document.pdf",
    sender: str = "@user:localhost",
    server_timestamp: int = 1000,
    thread_id: str | None = None,
) -> nio.RoomMessageFile:
    """Build a synthetic inbound file event for coalescing tests."""
    content: dict[str, object] = {
        "msgtype": "m.file",
        "body": body,
        "filename": body,
        "url": "mxc://localhost/test-file",
        "info": {"mimetype": "application/pdf"},
    }
    if thread_id is not None:
        content["m.relates_to"] = {"rel_type": "m.thread", "event_id": thread_id}
    return cast(
        "nio.RoomMessageFile",
        nio.RoomMessageFile.from_dict(
            {
                "event_id": event_id,
                "sender": sender,
                "origin_server_ts": server_timestamp,
                "room_id": "!room:localhost",
                "type": "m.room.message",
                "content": content,
            },
        ),
    )


def _prepared_dispatch(
    *,
    event_id: str,
    requester_user_id: str = "@user:localhost",
    body: str = "hello",
    thread_id: str | None = None,
    source_kind: str = "message",
    dispatch_policy_source_kind: str | None = None,
) -> PreparedDispatch:
    history: list[ResolvedVisibleMessage] = []
    context = MessageContext(
        am_i_mentioned=True,
        is_thread=thread_id is not None,
        thread_id=thread_id,
        thread_history=history,
        replay_guard_history=history,
        mentioned_agents=[],
        has_non_agent_mentions=False,
    )
    target = MessageTarget.resolve(
        room_id="!room:localhost",
        thread_id=thread_id,
        reply_to_event_id=event_id,
    )
    return PreparedDispatch(
        requester_user_id=requester_user_id,
        context=context,
        target=target,
        correlation_id=event_id,
        envelope=MessageEnvelope(
            source_event_id=event_id,
            target=target,
            body=body,
            attachment_ids=(),
            mentioned_agents=(),
            agent_name="test_agent",
            dispatch_policy_source_kind=dispatch_policy_source_kind,
            origin=message_origin(sender_id=requester_user_id, requester_id=requester_user_id, source_kind=source_kind),
        ),
    )


def _set_context_histories(dispatch: PreparedDispatch, history: Sequence[ResolvedVisibleMessage]) -> None:
    """Keep replay-snapshot and planning history aligned for tests that need both."""
    dispatch.context.thread_history = list(history)
    dispatch.context.replay_guard_history = list(history)


@pytest.mark.asyncio
@pytest.mark.parametrize("terminal_path", ["policy_ignore", "deep_synthetic", "non_owner_command"])
async def test_post_gate_terminal_drop_settles_real_deferred_dispatch_obligation(
    tmp_path: Path,
    terminal_path: str,
) -> None:
    """Every successful post-gate drop must provide terminal truth to durable dispatch."""
    bot = _make_bot(tmp_path, debounce_ms=0)
    room = _make_room()
    event = _text_event(
        event_id=f"$terminal-{terminal_path}",
        body="!help" if terminal_path == "non_owner_command" else "ignore me",
    )
    dispatch = _prepared_dispatch(event_id=event.event_id, body=event.body)
    runner = bot._dispatch_obligation_runner
    runner._retry_initial_delay_seconds = 0.001
    runner._retry_max_delay_seconds = 0.001
    await runner.persist(room, event, DispatchCallbackKind.MESSAGE)

    plan_turn = AsyncMock(return_value=_DispatchPlan(kind="ignore"))
    with (
        patch.object(
            bot._turn_controller,
            "_prepare_dispatch",
            new=AsyncMock(return_value=prepared_dispatch_result(dispatch)),
        ),
        patch.object(
            bot._turn_controller,
            "_should_skip_deep_synthetic_full_dispatch",
            return_value=terminal_path == "deep_synthetic",
        ),
        patch.object(
            bot._turn_policy,
            "plan_turn",
            new=plan_turn,
        ),
    ):
        await _enqueue_for_dispatch(
            bot,
            event,
            room,
            source_kind="message",
            requester_user_id=event.sender,
        )
        await bot._coalescing_gate.drain_all()

    if terminal_path == "policy_ignore":
        plan_turn.assert_awaited_once()
    else:
        plan_turn.assert_not_awaited()
    assert not bot._turn_store.is_durably_handled(event.event_id)
    await _wait_for(
        lambda: not runner.store.has_pending(event.event_id, DispatchCallbackKind.MESSAGE),
        deadline_seconds=1,
    )


@pytest.mark.asyncio
async def test_single_message_dispatches_after_debounce_window(tmp_path: Path) -> None:
    """Dispatch one text message once the debounce window elapses."""
    bot = _make_bot(tmp_path)
    room = _make_room()
    event = _text_event(event_id="$m1", body="hello")
    calls: list[tuple[str, list[str], list[object]]] = []

    async def record_dispatch(
        _room: nio.MatrixRoom,
        dispatched_event: nio.RoomMessageText,
        _requester_user_id: str,
        *,
        media_events: list[object] | None = None,
        handled_turn: TurnRecord | None = None,
        **_metadata: object,
    ) -> None:
        _ = handled_turn
        calls.append((dispatched_event.body, _handled_turn_source_event_ids(handled_turn), media_events or []))

    with patch.object(bot._turn_controller, "_dispatch_text_message", new=AsyncMock(side_effect=record_dispatch)):
        await _enqueue_for_dispatch(
            bot,
            event,
            room,
            source_kind="message",
            requester_user_id="@user:localhost",
        )
        assert calls == []
        await _wait_for(lambda: calls == [("hello", ["$m1"], [])])

    assert calls == [("hello", ["$m1"], [])]


@pytest.mark.asyncio
async def test_two_rapid_text_messages_dispatch_one_combined_turn(tmp_path: Path) -> None:
    """Coalesce two quick thread messages into one combined prompt."""
    bot = _make_bot(tmp_path)
    room = _make_room()
    first = _text_event(event_id="$m1", body="first", server_timestamp=1000, thread_id="$thread")
    second = _text_event(event_id="$m2", body="second", server_timestamp=1001, thread_id="$thread")
    calls: list[tuple[str, str, list[str]]] = []

    async def record_dispatch(
        _room: nio.MatrixRoom,
        dispatched_event: nio.RoomMessageText,
        _requester_user_id: str,
        *,
        media_events: list[object] | None = None,
        handled_turn: TurnRecord | None = None,
        **_metadata: object,
    ) -> None:
        _ = media_events, handled_turn
        calls.append((dispatched_event.event_id, dispatched_event.body, _handled_turn_source_event_ids(handled_turn)))

    with patch.object(bot._turn_controller, "_dispatch_text_message", new=AsyncMock(side_effect=record_dispatch)):
        await _enqueue_for_dispatch(
            bot,
            first,
            room,
            source_kind="message",
            requester_user_id="@user:localhost",
        )
        await _enqueue_for_dispatch(
            bot,
            second,
            room,
            source_kind="message",
            requester_user_id="@user:localhost",
        )
        await bot._coalescing_gate.drain_all()

    assert calls == [
        (
            "$m2",
            "The user sent the following messages in quick succession. "
            "Treat them as one turn and respond once:\n\n"
            "<messages>\n"
            '<msg event_id="$m1" from="@user:localhost" ts="1970-01-01 00:00 UTC"><![CDATA[first]]></msg>\n'
            '<msg event_id="$m2" from="@user:localhost" ts="1970-01-01 00:00 UTC"><![CDATA[second]]></msg>\n'
            "</messages>",
            ["$m1", "$m2"],
        ),
    ]


@pytest.mark.asyncio
async def test_two_rapid_text_messages_forward_prompt_map_to_dispatch(tmp_path: Path) -> None:
    """Thread-scoped coalesced dispatch should forward the per-source prompt map."""
    bot = _make_bot(tmp_path)
    room = _make_room()
    first = _text_event(event_id="$m1", body="first", server_timestamp=1000, thread_id="$thread")
    second = _text_event(event_id="$m2", body="second", server_timestamp=1001, thread_id="$thread")

    with patch.object(bot._turn_controller, "_dispatch_text_message", new=AsyncMock()) as mock_dispatch:
        await _enqueue_for_dispatch(
            bot,
            first,
            room,
            source_kind="message",
            requester_user_id="@user:localhost",
        )
        await _enqueue_for_dispatch(
            bot,
            second,
            room,
            source_kind="message",
            requester_user_id="@user:localhost",
        )
        await bot._coalescing_gate.drain_all()

    assert mock_dispatch.await_count == 1
    handled_turn = mock_dispatch.await_args.kwargs["handled_turn"]
    assert list(handled_turn.source_event_ids) == ["$m1", "$m2"]
    assert handled_turn.source_event_prompts == {
        "$m1": "first",
        "$m2": "second",
    }


@pytest.mark.asyncio
async def test_image_and_text_coalesce_into_single_dispatch(tmp_path: Path) -> None:
    """Coalesce thread image uploads and follow-up text into one dispatch."""
    bot = _make_bot(tmp_path)
    room = _make_room()
    image_event = _image_event(event_id="$img1", server_timestamp=1000, thread_id="$thread")
    text_event = _text_event(event_id="$m2", body="describe it", server_timestamp=1001, thread_id="$thread")
    calls: list[tuple[str, list[str], int]] = []

    async def record_dispatch(
        _room: nio.MatrixRoom,
        dispatched_event: nio.RoomMessageText,
        _requester_user_id: str,
        *,
        media_events: list[object] | None = None,
        handled_turn: TurnRecord | None = None,
        **_metadata: object,
    ) -> None:
        _ = handled_turn
        calls.append((dispatched_event.body, _handled_turn_source_event_ids(handled_turn), len(media_events or [])))

    with patch.object(bot._turn_controller, "_dispatch_text_message", new=AsyncMock(side_effect=record_dispatch)):
        await _enqueue_for_dispatch(
            bot,
            image_event,
            room,
            source_kind="image",
            requester_user_id="@user:localhost",
        )
        await _enqueue_for_dispatch(
            bot,
            text_event,
            room,
            source_kind="message",
            requester_user_id="@user:localhost",
        )
        await bot._coalescing_gate.drain_all()

    assert calls == [
        (
            "The user sent the following messages in quick succession. "
            "Treat them as one turn and respond once:\n\n"
            "<messages>\n"
            '<msg event_id="$img1" from="@user:localhost" ts="1970-01-01 00:00 UTC"><![CDATA[[Attached image]]]></msg>\n'
            '<msg event_id="$m2" from="@user:localhost" ts="1970-01-01 00:00 UTC"><![CDATA[describe it]]></msg>\n'
            "</messages>",
            ["$img1", "$m2"],
            1,
        ),
    ]


@pytest.mark.asyncio
async def test_room_root_image_and_caption_coalesce_into_single_dispatch(tmp_path: Path) -> None:
    """Root media and the trailing caption share the room coalescing scope into one dispatch."""
    bot = _make_bot(tmp_path, debounce_ms=200)
    room = _make_room()
    image_event = _image_event(event_id="$img", server_timestamp=1000)
    text_event = _text_event(event_id="$text", body="describe this", server_timestamp=1001)
    calls: list[tuple[str, list[str], int]] = []

    async def record_dispatch(
        _room: nio.MatrixRoom,
        dispatched_event: nio.RoomMessageText,
        _requester_user_id: str,
        *,
        media_events: list[object] | None = None,
        handled_turn: TurnRecord | None = None,
        **_metadata: object,
    ) -> None:
        calls.append((dispatched_event.body, _handled_turn_source_event_ids(handled_turn), len(media_events or [])))

    with patch.object(bot._turn_controller, "_dispatch_text_message", new=AsyncMock(side_effect=record_dispatch)):
        await bot._turn_controller.handle_media_event(room, image_event)
        await asyncio.sleep(0.005)
        await bot._turn_controller.handle_text_event(room, text_event)
        await _wait_for(lambda: len(calls) == 1)

    assert calls == [
        (
            "The user sent the following messages in quick succession. "
            "Treat them as one turn and respond once:\n\n"
            "<messages>\n"
            '<msg event_id="$img" from="@user:localhost" ts="1970-01-01 00:00 UTC"><![CDATA[[Attached image]]]></msg>\n'
            '<msg event_id="$text" from="@user:localhost" ts="1970-01-01 00:00 UTC"><![CDATA[describe this]]></msg>\n'
            "</messages>",
            ["$img", "$text"],
            1,
        ),
    ]


@pytest.mark.asyncio
async def test_images_then_caption_coalesce_into_single_dispatch(tmp_path: Path) -> None:
    """Uploads arrive first and the trailing caption closes the batch into one dispatch."""
    bot = _make_bot(tmp_path, debounce_ms=60_000)
    room = _make_room()
    first_image = _image_event(event_id="$img1", server_timestamp=1000, thread_id="$thread")
    second_image = _image_event(event_id="$img2", server_timestamp=1001, thread_id="$thread")
    caption = _text_event(event_id="$m1", body="summarize these", server_timestamp=1002, thread_id="$thread")
    calls: list[tuple[list[str], int]] = []

    async def record_dispatch(
        _room: nio.MatrixRoom,
        _dispatched_event: nio.RoomMessageText,
        _requester_user_id: str,
        *,
        media_events: list[object] | None = None,
        handled_turn: TurnRecord | None = None,
        **_metadata: object,
    ) -> None:
        calls.append((_handled_turn_source_event_ids(handled_turn), len(media_events or [])))

    with patch.object(bot._turn_controller, "_dispatch_text_message", new=AsyncMock(side_effect=record_dispatch)):
        await _enqueue_for_dispatch(
            bot,
            first_image,
            room,
            source_kind="image",
            requester_user_id="@user:localhost",
        )
        await _enqueue_for_dispatch(
            bot,
            second_image,
            room,
            source_kind="image",
            requester_user_id="@user:localhost",
        )
        await asyncio.sleep(0.02)
        assert calls == []

        await _enqueue_for_dispatch(
            bot,
            caption,
            room,
            source_kind="message",
            requester_user_id="@user:localhost",
        )
        await _wait_for(lambda: len(calls) == 1)

    assert calls == [(["$img1", "$img2", "$m1"], 2)]
    assert _coalescing_gate_is_idle(bot._coalescing_gate)


@pytest.mark.asyncio
async def test_each_thread_text_dispatches_immediately_as_own_turn(tmp_path: Path) -> None:
    """Thread text never waits for the debounce window; later text becomes its own turn."""
    bot = _make_bot(tmp_path, debounce_ms=60_000)
    room = _make_room()
    first = _text_event(event_id="$m1", body="first turn", server_timestamp=1000, thread_id="$thread")
    second = _text_event(event_id="$m2", body="second turn", server_timestamp=1001, thread_id="$thread")
    calls: list[tuple[str, list[str]]] = []

    async def record_dispatch(
        _room: nio.MatrixRoom,
        dispatched_event: nio.RoomMessageText,
        _requester_user_id: str,
        *,
        media_events: list[object] | None = None,
        handled_turn: TurnRecord | None = None,
        **_metadata: object,
    ) -> None:
        _ = media_events, handled_turn
        calls.append((dispatched_event.body, _handled_turn_source_event_ids(handled_turn)))

    with patch.object(bot._turn_controller, "_dispatch_text_message", new=AsyncMock(side_effect=record_dispatch)):
        await _enqueue_for_dispatch(
            bot,
            first,
            room,
            source_kind="message",
            requester_user_id="@user:localhost",
        )
        await _wait_for(lambda: calls == [("first turn", ["$m1"])])

        await _enqueue_for_dispatch(
            bot,
            second,
            room,
            source_kind="message",
            requester_user_id="@user:localhost",
        )
        await _wait_for(lambda: len(calls) == 2)

    assert calls == [
        ("first turn", ["$m1"]),
        ("second turn", ["$m2"]),
    ]


@pytest.mark.asyncio
async def test_image_after_text_dispatch_is_a_second_batch(tmp_path: Path) -> None:
    """An upload that arrives after its text already dispatched remains a later turn."""
    bot = _make_bot(tmp_path, debounce_ms=10)
    room = _make_room()
    text_event = _text_event(event_id="$m1", body="first turn", server_timestamp=1000)
    image_event = _image_event(event_id="$img1", server_timestamp=1001)
    calls: list[tuple[list[str], int]] = []

    async def record_dispatch(
        _room: nio.MatrixRoom,
        _dispatched_event: nio.RoomMessageText,
        _requester_user_id: str,
        *,
        media_events: list[object] | None = None,
        handled_turn: TurnRecord | None = None,
        **_metadata: object,
    ) -> None:
        _ = handled_turn
        calls.append((_handled_turn_source_event_ids(handled_turn), len(media_events or [])))

    with patch.object(bot._turn_controller, "_dispatch_text_message", new=AsyncMock(side_effect=record_dispatch)):
        await _enqueue_for_dispatch(
            bot,
            text_event,
            room,
            source_kind="message",
            requester_user_id="@user:localhost",
        )
        await _wait_for(lambda: len(calls) == 1)

        await _enqueue_for_dispatch(
            bot,
            image_event,
            room,
            source_kind="image",
            requester_user_id="@user:localhost",
        )
        await _wait_for(lambda: len(calls) == 2)

    assert calls == [
        (["$m1"], 0),
        (["$img1"], 1),
    ]


@pytest.mark.asyncio
async def test_different_senders_dispatch_separately(tmp_path: Path) -> None:
    """Keep coalescing isolated per sending Matrix user."""
    bot = _make_bot(tmp_path)
    room = _make_room()
    alice = _text_event(event_id="$m1", body="hi", sender="@alice:localhost")
    bob = _text_event(event_id="$m2", body="hello", sender="@bob:localhost")
    calls: list[list[str]] = []

    async def record_dispatch(
        _room: nio.MatrixRoom,
        _dispatched_event: nio.RoomMessageText,
        _requester_user_id: str,
        *,
        media_events: list[object] | None = None,
        handled_turn: TurnRecord | None = None,
        **_metadata: object,
    ) -> None:
        _ = media_events, handled_turn
        calls.append(_handled_turn_source_event_ids(handled_turn))

    with patch.object(bot._turn_controller, "_dispatch_text_message", new=AsyncMock(side_effect=record_dispatch)):
        await _enqueue_for_dispatch(
            bot,
            alice,
            room,
            source_kind="message",
            requester_user_id="@alice:localhost",
        )
        await _enqueue_for_dispatch(
            bot,
            bob,
            room,
            source_kind="message",
            requester_user_id="@bob:localhost",
        )
        await bot._coalescing_gate.drain_all()

    assert sorted(calls) == [["$m1"], ["$m2"]]


def test_build_coalesced_batch_keeps_normalized_voice_out_of_media_events() -> None:
    """Voice messages should enter coalescing as synthetic text, not raw media."""
    room = _make_room()
    voice_event = PreparedTextEvent(
        sender="@user:localhost",
        event_id="$voice1",
        body="transcribed voice",
        source={"content": {"body": "transcribed voice", SOURCE_KIND_KEY: "voice"}},
    )

    batch = build_coalesced_batch(
        CoalescingKey("!room:localhost", None, "@user:localhost"),
        [PendingEvent(event=voice_event, room=room, source_kind="voice")],
    )

    assert batch.prompt == "transcribed voice"
    assert batch.source_event_ids == ["$voice1"]
    assert batch.media_events == []


def test_build_coalesced_batch_preserves_fifo_order_with_synthetic_events() -> None:
    """Preserve queue order even when Matrix timestamps disagree."""
    room = _make_room()
    real_event = _text_event(event_id="$real", body="real", server_timestamp=1_712_350_002_000)
    synthetic_event = PreparedTextEvent(
        sender="@user:localhost",
        event_id="$synthetic",
        body="synthetic",
        source={"content": {"body": "synthetic", SOURCE_KIND_KEY: "voice"}},
        server_timestamp=1_712_350_003_000,
    )

    batch = build_coalesced_batch(
        CoalescingKey("!room:localhost", None, "@user:localhost"),
        [
            PendingEvent(event=synthetic_event, room=room, source_kind="voice", enqueue_time=50_000.0),
            PendingEvent(event=real_event, room=room, source_kind="message"),
        ],
    )

    assert batch.source_event_ids == ["$synthetic", "$real"]
    assert batch.prompt.endswith("synthetic\nreal")


def test_build_coalesced_batch_prefers_media_source_kind_over_text_primary() -> None:
    """Mixed batches should keep media source_kind even when text is the primary event."""
    room = _make_room()
    image_event = _image_event(event_id="$img1", server_timestamp=1000)
    text_event = _text_event(event_id="$m2", body="describe it", server_timestamp=1001)

    batch = build_coalesced_batch(
        CoalescingKey("!room:localhost", None, "@user:localhost"),
        [
            PendingEvent(event=image_event, room=room, source_kind="image"),
            PendingEvent(event=text_event, room=room, source_kind="message"),
        ],
    )

    assert batch.primary_event is text_event
    assert batch.source_kind == "image"


def test_build_coalesced_batch_prefers_voice_source_kind_over_media_and_text() -> None:
    """Voice should win batch source_kind precedence even when a text event is primary."""
    room = _make_room()
    voice_event = PreparedTextEvent(
        sender="@user:localhost",
        event_id="$voice1",
        body="voice prompt",
        source={"content": {"body": "voice prompt", SOURCE_KIND_KEY: "voice"}},
    )
    image_event = _image_event(event_id="$img1", server_timestamp=1000)
    text_event = _text_event(event_id="$m2", body="follow-up", server_timestamp=1001)

    batch = build_coalesced_batch(
        CoalescingKey("!room:localhost", None, "@user:localhost"),
        [
            PendingEvent(event=voice_event, room=room, source_kind="voice", enqueue_time=0.5),
            PendingEvent(event=image_event, room=room, source_kind="image"),
            PendingEvent(event=text_event, room=room, source_kind="message"),
        ],
    )

    assert batch.primary_event is text_event
    assert batch.source_kind == "voice"


@pytest.mark.asyncio
async def test_same_sender_different_threads_dispatch_separately(tmp_path: Path) -> None:
    """Keep coalescing isolated per thread for the same sender."""
    bot = _make_bot(tmp_path)
    room = _make_room()
    thread_a = _text_event(event_id="$m1", body="a", thread_id="$thread-a")
    thread_b = _text_event(event_id="$m2", body="b", thread_id="$thread-b")
    calls: list[list[str]] = []

    async def record_dispatch(
        _room: nio.MatrixRoom,
        _dispatched_event: nio.RoomMessageText,
        _requester_user_id: str,
        *,
        media_events: list[object] | None = None,
        handled_turn: TurnRecord | None = None,
        **_metadata: object,
    ) -> None:
        _ = media_events, handled_turn
        calls.append(_handled_turn_source_event_ids(handled_turn))

    with patch.object(bot._turn_controller, "_dispatch_text_message", new=AsyncMock(side_effect=record_dispatch)):
        await _enqueue_for_dispatch(
            bot,
            thread_a,
            room,
            source_kind="message",
            requester_user_id="@user:localhost",
        )
        await _enqueue_for_dispatch(
            bot,
            thread_b,
            room,
            source_kind="message",
            requester_user_id="@user:localhost",
        )
        await _wait_for(lambda: len(calls) == 2)

    assert sorted(calls) == [["$m1"], ["$m2"]]


@pytest.mark.asyncio
async def test_room_message_and_plain_reply_to_known_thread_do_not_coalesce_together(tmp_path: Path) -> None:
    """Inherited-thread plain replies must not batch with unrelated room-level messages."""
    bot = _make_bot(tmp_path)
    room = _make_room()
    room_message = _text_event(event_id="$roommsg", body="room message", server_timestamp=1000)
    threaded_plain_reply = _reply_event(
        event_id="$reply",
        body="bridged follow-up",
        reply_to_event_id="$thread-seed",
        server_timestamp=1001,
    )
    bot._turn_controller.deps.resolver.deps.conversation_cache.get_thread_id_for_event = AsyncMock(
        side_effect=lambda _room_id, event_id: "$thread-root" if event_id == "$thread-seed" else None,
    )
    calls: list[list[str]] = []

    async def record_dispatch(
        _room: nio.MatrixRoom,
        _dispatched_event: nio.RoomMessageText,
        _requester_user_id: str,
        *,
        media_events: list[object] | None = None,
        handled_turn: TurnRecord | None = None,
        **_metadata: object,
    ) -> None:
        _ = media_events, handled_turn
        calls.append(_handled_turn_source_event_ids(handled_turn))

    with patch.object(bot._turn_controller, "_dispatch_text_message", new=AsyncMock(side_effect=record_dispatch)):
        await _enqueue_for_dispatch(
            bot,
            room_message,
            room,
            source_kind="message",
            requester_user_id="@user:localhost",
        )
        await _enqueue_for_dispatch(
            bot,
            threaded_plain_reply,
            room,
            source_kind="message",
            requester_user_id="@user:localhost",
        )
        await _wait_for(lambda: len(calls) == 2)

    assert sorted(calls) == [["$reply"], ["$roommsg"]]


@pytest.mark.asyncio
async def test_plain_reply_with_unproven_root_is_not_admitted_under_guessed_key(tmp_path: Path) -> None:
    """Unproven roots should not be admitted as canonical room-level coalescing keys."""
    bot = _make_bot(tmp_path)
    room = _make_room()
    reply_a = _reply_event(
        event_id="$reply-a",
        body="follow up a",
        reply_to_event_id="$root-a",
        server_timestamp=1000,
    )

    def root_response(event_id: str) -> nio.RoomGetEventResponse:
        return nio.RoomGetEventResponse.from_dict(
            {
                "content": {"body": event_id, "msgtype": "m.text"},
                "event_id": event_id,
                "sender": "@user:localhost",
                "origin_server_ts": 999,
                "room_id": room.room_id,
                "type": "m.room.message",
            },
        )

    async def get_event(_room_id: str, event_id: str) -> nio.RoomGetEventResponse:
        return root_response(event_id)

    bot._turn_controller.deps.resolver.deps.conversation_cache.get_thread_id_for_event = AsyncMock(return_value=None)
    bot._turn_controller.deps.resolver.deps.conversation_cache.get_event = AsyncMock(side_effect=get_event)
    bot._turn_controller.deps.resolver.deps.conversation_cache.get_dispatch_thread_snapshot = AsyncMock(
        side_effect=TimeoutError("dispatch read timed out"),
    )
    calls: list[list[str]] = []

    async def record_dispatch(
        _room: nio.MatrixRoom,
        _dispatched_event: nio.RoomMessageText,
        _requester_user_id: str,
        *,
        media_events: list[object] | None = None,
        handled_turn: TurnRecord | None = None,
        **_metadata: object,
    ) -> None:
        _ = media_events, handled_turn
        calls.append(_handled_turn_source_event_ids(handled_turn))

    with (
        patch.object(bot._turn_controller, "_dispatch_text_message", new=AsyncMock(side_effect=record_dispatch)),
        pytest.raises(RuntimeError, match="Could not resolve canonical coalescing thread"),
    ):
        await _enqueue_for_dispatch(
            bot,
            reply_a,
            room,
            source_kind="message",
            requester_user_id="@user:localhost",
        )

    assert calls == []


@pytest.mark.asyncio
async def test_command_executes_immediately_while_text_batch_debounces(tmp_path: Path) -> None:
    """A command is a control input: it dispatches before batching and never waits on debounce."""
    bot = _make_bot(tmp_path, debounce_ms=60_000)
    room = _make_room()
    first = _text_event(event_id="$m1", body="tell me more", server_timestamp=1000)
    command = _text_event(event_id="$m2", body="!help", server_timestamp=1001)
    calls: list[tuple[str, list[str]]] = []

    async def record_dispatch(
        _room: nio.MatrixRoom,
        dispatched_event: nio.RoomMessageText,
        _requester_user_id: str,
        *,
        media_events: list[object] | None = None,
        handled_turn: TurnRecord | None = None,
        **_metadata: object,
    ) -> None:
        _ = media_events, handled_turn
        calls.append((dispatched_event.body, _handled_turn_source_event_ids(handled_turn)))

    with patch.object(bot._turn_controller, "_dispatch_text_message", new=AsyncMock(side_effect=record_dispatch)):
        await bot._turn_controller.handle_text_event(room, first)
        await bot._turn_controller.handle_text_event(room, command)

        assert calls == [("!help", ["$m2"])]

        await bot._coalescing_gate.drain_all()

    assert calls == [
        ("!help", ["$m2"]),
        ("tell me more", ["$m1"]),
    ]


@pytest.mark.asyncio
async def test_command_during_media_debounce_executes_immediately(tmp_path: Path) -> None:
    """A command never waits behind a held media batch, which flushes on its own."""
    bot = _make_bot(tmp_path, debounce_ms=60_000)
    room = _make_room()
    image_event = _image_event(event_id="$img1", server_timestamp=1000)
    command_event = _text_event(event_id="$m2", body="!help", server_timestamp=1001)
    calls: list[list[str]] = []

    async def record_dispatch(
        _room: nio.MatrixRoom,
        _dispatched_event: nio.RoomMessageText,
        _requester_user_id: str,
        *,
        media_events: list[object] | None = None,
        handled_turn: TurnRecord | None = None,
        **_metadata: object,
    ) -> None:
        _ = media_events
        calls.append(_handled_turn_source_event_ids(handled_turn))

    with patch.object(bot._turn_controller, "_dispatch_text_message", new=AsyncMock(side_effect=record_dispatch)):
        await bot._turn_controller.handle_media_event(room, image_event)
        await bot._turn_controller.handle_text_event(room, command_event)

        assert calls == [["$m2"]]

        await bot._coalescing_gate.drain_all()

    assert calls == [
        ["$m2"],
        ["$img1"],
    ]


@pytest.mark.asyncio
async def test_messages_during_active_response_wait_and_batch_after_completion(tmp_path: Path) -> None:
    """Hold all threaded follow-ups while the first-turn response is in flight, then batch them."""
    bot = _make_bot(tmp_path, debounce_ms=10)
    room = _make_room()
    first = _text_event(event_id="$m1", body="first", server_timestamp=1000)
    second = _text_event(event_id="$m2", body="second", server_timestamp=1001, thread_id="$m1")
    third = _text_event(event_id="$m3", body="third", server_timestamp=1002, thread_id="$m1")
    entered_first_dispatch = asyncio.Event()
    release_first_dispatch = asyncio.Event()
    calls: list[list[str]] = []

    active_threads: set[str | None] = set()

    async def wait_for_thread_response_idle(room_id: str, thread_id: str | None) -> None:
        assert room_id == room.room_id
        assert thread_id == "$m1"
        await release_first_dispatch.wait()

    async def record_dispatch(
        _room: nio.MatrixRoom,
        _dispatched_event: nio.RoomMessageText,
        _requester_user_id: str,
        *,
        media_events: list[object] | None = None,
        handled_turn: TurnRecord | None = None,
        **_metadata: object,
    ) -> None:
        _ = media_events, handled_turn
        calls.append(_handled_turn_source_event_ids(handled_turn))
        if _handled_turn_source_event_ids(handled_turn) == ["$m1"]:
            entered_first_dispatch.set()
            await release_first_dispatch.wait()

    with (
        patch.object(bot._turn_controller, "_dispatch_text_message", new=AsyncMock(side_effect=record_dispatch)),
        patch.object(
            bot._response_runner,
            "wait_for_thread_response_idle",
            new=AsyncMock(side_effect=wait_for_thread_response_idle),
        ),
        patch.object(
            bot._response_runner,
            "active_thread_ids_for_room",
            new=lambda _room_id: frozenset(active_threads),
        ),
    ):
        await _enqueue_for_dispatch(
            bot,
            first,
            room,
            source_kind="message",
            requester_user_id="@user:localhost",
        )
        await asyncio.sleep(0.03)
        await entered_first_dispatch.wait()
        active_threads.add("$m1")

        for followup in (second, third):
            await _enqueue_for_dispatch(
                bot,
                followup,
                room,
                source_kind="message",
                requester_user_id="@user:localhost",
                coalescing_key=CoalescingKey(room.room_id, "$m1", "@user:localhost"),
            )
            await asyncio.sleep(0.03)

        assert calls == [["$m1"]]

        active_threads.clear()
        release_first_dispatch.set()
        await _wait_for(lambda: calls == [["$m1"], ["$m2", "$m3"]])

    assert calls == [["$m1"], ["$m2", "$m3"]]


@pytest.mark.asyncio
async def test_active_follow_ups_share_target_gate_across_requesters(tmp_path: Path) -> None:
    """Active-response follow-ups should queue by conversation while preserving each requester."""
    bot = _make_bot(tmp_path, debounce_ms=0)
    room = _make_room()
    first = _text_event(event_id="$a", body="first", sender="@alice:localhost", thread_id="$thread")
    second = _text_event(event_id="$b", body="second", sender="@bob:localhost", thread_id="$thread")
    active_threads: set[str | None] = {"$thread"}
    idle = asyncio.Event()
    calls: list[CoalescedBatch] = []

    async def record_dispatch(batch: CoalescedBatch) -> None:
        calls.append(batch)

    async def wait_for_thread_response_idle(_room_id: str, _thread_id: str | None) -> None:
        await idle.wait()

    with (
        patch.object(bot._turn_controller, "handle_coalesced_batch", new=AsyncMock(side_effect=record_dispatch)),
        patch.object(
            bot._response_runner,
            "wait_for_thread_response_idle",
            new=AsyncMock(side_effect=wait_for_thread_response_idle),
        ),
        patch.object(
            bot._response_runner,
            "active_thread_ids_for_room",
            new=lambda _room_id: frozenset(active_threads),
        ),
        patch.object(bot._response_runner, "reserve_waiting_human_message", return_value=MagicMock()),
    ):
        for event, requester_user_id in ((first, "@alice:localhost"), (second, "@bob:localhost")):
            await _enqueue_for_dispatch(
                bot,
                event,
                room,
                source_kind="message",
                requester_user_id=requester_user_id,
                coalescing_key=CoalescingKey(room.room_id, "$thread", requester_user_id),
            )
        await asyncio.sleep(0.01)
        assert calls == []

        active_threads.clear()
        idle.set()
        await _wait_for(lambda: [list(batch.source_event_ids) for batch in calls] == [["$a", "$b"]])

    assert calls[0].dispatch_policy_source_kind == ACTIVE_THREAD_FOLLOW_UP_SOURCE_KIND
    assert [pending_event.requester_user_id for pending_event in calls[0].pending_events] == [
        "@alice:localhost",
        "@bob:localhost",
    ]


@pytest.mark.asyncio
async def test_slow_thread_lookup_active_follow_up_stays_before_later_follow_up(tmp_path: Path) -> None:
    """A slow older lookup must stay before a later same-sender follow-up in the backlog."""
    bot = _make_bot(tmp_path, debounce_ms=0)
    room = _make_room()
    second = _text_event(
        event_id="$m2",
        body="second",
        sender="@alice:localhost",
        server_timestamp=1001,
        thread_id="$thread",
    )
    third = _text_event(
        event_id="$m3",
        body="third",
        sender="@alice:localhost",
        server_timestamp=1002,
        thread_id="$thread",
    )
    response_runner = unwrap_extracted_collaborator(bot._response_runner)
    lifecycle = response_runner._lifecycle_coordinator
    target = MessageTarget.resolve(room.room_id, "$thread", "$response")
    lifecycle_lock = lifecycle._response_lifecycle_lock(target)
    queued_signal = lifecycle._get_or_create_queued_signal(target)
    second_lookup_started = asyncio.Event()
    release_second_lookup = asyncio.Event()
    calls: list[list[str]] = []

    async def coalescing_thread_id(_room: nio.MatrixRoom, event: nio.Event | PreparedTextEvent) -> str | None:
        if event.event_id == "$m2":
            second_lookup_started.set()
            await release_second_lookup.wait()
        return "$thread"

    async def record_dispatch(
        _room: nio.MatrixRoom,
        _dispatched_event: nio.RoomMessageText,
        _requester_user_id: str,
        *,
        media_events: list[object] | None = None,
        handled_turn: TurnRecord | None = None,
        **_metadata: object,
    ) -> None:
        _ = media_events
        calls.append(_handled_turn_source_event_ids(handled_turn))

    await lifecycle_lock.acquire()
    queued_signal.begin_response_turn()
    second_task: asyncio.Task[None] | None = None
    third_task: asyncio.Task[None] | None = None
    try:
        with (
            patch.object(
                bot._conversation_resolver,
                "coalescing_thread_id",
                new=AsyncMock(side_effect=coalescing_thread_id),
            ),
            patch.object(bot._turn_controller, "_dispatch_text_message", new=AsyncMock(side_effect=record_dispatch)),
            patch("mindroom.turn_controller.interactive.handle_text_response", new=AsyncMock(return_value=None)),
        ):
            second_task = asyncio.create_task(bot._turn_controller.handle_text_event(room, second))
            await asyncio.wait_for(second_lookup_started.wait(), timeout=0.5)

            third_task = asyncio.create_task(bot._turn_controller.handle_text_event(room, third))
            await asyncio.wait_for(third_task, timeout=0.5)

            queued_signal.finish_response_turn()
            lifecycle_lock.release()
            await asyncio.sleep(0.05)
            assert calls == []

            release_second_lookup.set()
            await asyncio.wait_for(second_task, timeout=0.5)
            await _wait_for(lambda: calls == [["$m2", "$m3"]])
    finally:
        release_second_lookup.set()
        queued_signal.finish_response_turn()
        if lifecycle_lock.locked():
            lifecycle_lock.release()
        pending_tasks = [task for task in (second_task, third_task) if task is not None and not task.done()]
        for task in pending_tasks:
            task.cancel()
        if pending_tasks:
            await asyncio.gather(*pending_tasks, return_exceptions=True)

    assert calls == [["$m2", "$m3"]]


@pytest.mark.asyncio
async def test_later_slow_thread_lookup_active_follow_up_lands_as_own_turn(tmp_path: Path) -> None:
    """A follow-up still resolving when the backlog flushes lands as its own turn, never lost."""
    bot = _make_bot(tmp_path, debounce_ms=0)
    room = _make_room()
    first = _text_event(
        event_id="$m1",
        body="first",
        sender="@alice:localhost",
        server_timestamp=1001,
        thread_id="$thread",
    )
    second = _text_event(
        event_id="$m2",
        body="second",
        sender="@bob:localhost",
        server_timestamp=1002,
        thread_id="$thread",
    )
    response_runner = unwrap_extracted_collaborator(bot._response_runner)
    lifecycle = response_runner._lifecycle_coordinator
    target = MessageTarget.resolve(room.room_id, "$thread", "$response")
    lifecycle_lock = lifecycle._response_lifecycle_lock(target)
    queued_signal = lifecycle._get_or_create_queued_signal(target)
    second_lookup_started = asyncio.Event()
    release_second_lookup = asyncio.Event()
    calls: list[list[str]] = []

    async def coalescing_thread_id(_room: nio.MatrixRoom, event: nio.Event | PreparedTextEvent) -> str | None:
        if event.event_id == "$m2":
            second_lookup_started.set()
            await release_second_lookup.wait()
        return "$thread"

    async def record_dispatch(
        _room: nio.MatrixRoom,
        _dispatched_event: nio.RoomMessageText,
        _requester_user_id: str,
        *,
        media_events: list[object] | None = None,
        handled_turn: TurnRecord | None = None,
        **_metadata: object,
    ) -> None:
        _ = media_events
        calls.append(_handled_turn_source_event_ids(handled_turn))

    await lifecycle_lock.acquire()
    queued_signal.begin_response_turn()
    first_task: asyncio.Task[None] | None = None
    second_task: asyncio.Task[None] | None = None
    try:
        with (
            patch.object(
                bot._conversation_resolver,
                "coalescing_thread_id",
                new=AsyncMock(side_effect=coalescing_thread_id),
            ),
            patch.object(bot._turn_controller, "_dispatch_text_message", new=AsyncMock(side_effect=record_dispatch)),
            patch("mindroom.turn_controller.interactive.handle_text_response", new=AsyncMock(return_value=None)),
        ):
            first_task = asyncio.create_task(bot._turn_controller.handle_text_event(room, first))
            await asyncio.wait_for(first_task, timeout=0.5)

            second_task = asyncio.create_task(bot._turn_controller.handle_text_event(room, second))
            await asyncio.wait_for(second_lookup_started.wait(), timeout=0.5)

            queued_signal.finish_response_turn()
            lifecycle_lock.release()
            await _wait_for(lambda: calls == [["$m1"]])

            release_second_lookup.set()
            await asyncio.wait_for(second_task, timeout=0.5)
            await _wait_for(lambda: calls == [["$m1"], ["$m2"]])
    finally:
        release_second_lookup.set()
        queued_signal.finish_response_turn()
        if lifecycle_lock.locked():
            lifecycle_lock.release()
        pending_tasks = [task for task in (first_task, second_task) if task is not None and not task.done()]
        for task in pending_tasks:
            task.cancel()
        if pending_tasks:
            await asyncio.gather(*pending_tasks, return_exceptions=True)

    assert calls == [["$m1"], ["$m2"]]


@pytest.mark.asyncio
async def test_active_follow_up_owner_includes_later_media_payload(tmp_path: Path) -> None:
    """The first queued follow-up owner should carry later media before skipping its duplicate run."""
    bot = _make_bot(tmp_path, debounce_ms=0)
    room = _make_room()
    text_event = _text_event(
        event_id="$text",
        body="text follow-up",
        sender="@alice:localhost",
        server_timestamp=1000,
        thread_id="$thread",
    )
    image_event = _image_event(
        event_id="$img",
        sender="@bob:localhost",
        server_timestamp=1001,
        thread_id="$thread",
    )
    target = MessageTarget.resolve(room.room_id, "$thread", "$text")
    response_runner = unwrap_extracted_collaborator(bot._response_runner)
    lifecycle = response_runner._lifecycle_coordinator
    lifecycle_lock = lifecycle._response_lifecycle_lock(target)
    queued_signal = lifecycle._get_or_create_queued_signal(target)
    captured_requests = []
    payload_requests = []
    registered_media_event_ids = []

    async def fake_prepare_dispatch(
        _room: nio.MatrixRoom,
        event: DispatchEvent,
        requester_user_id: str,
        **_kwargs: object,
    ) -> object:
        dispatch = _prepared_dispatch(
            event_id=event.event_id,
            requester_user_id=requester_user_id,
            body=event.body,
            thread_id="$thread",
            dispatch_policy_source_kind=ACTIVE_THREAD_FOLLOW_UP_SOURCE_KIND,
        )
        dispatch.context.am_i_mentioned = True
        return prepared_dispatch_result(dispatch)

    async def fake_register_batch_media_attachments(
        request: BatchMediaAttachmentRequest,
    ) -> _BatchMediaAttachmentResult:
        registered_media_event_ids.append([media_event.event_id for media_event in request.media_events])
        return _BatchMediaAttachmentResult(attachment_ids=["img-attachment"])

    async def fake_build_payload(request: DispatchPayloadWithAttachmentsRequest) -> DispatchPayload:
        payload_requests.append(request)
        return DispatchPayload(
            prompt=request.prompt,
            model_prompt="attachment guidance",
            attachment_ids=list(request.current_attachment_ids),
        )

    async def fake_generate_response_locked(_self: object, request: object, **_kwargs: object) -> str:
        prepared_request = await response_runner._prepare_request_after_lock(request)
        captured_requests.append(prepared_request)
        return "$response"

    await lifecycle_lock.acquire()
    queued_signal.begin_response_turn()
    try:
        with (
            patch.object(bot._turn_controller, "_prepare_dispatch", new=fake_prepare_dispatch),
            patch.object(bot._turn_policy, "plan_turn", new=AsyncMock(return_value=_respond_dispatch_plan())),
            patch.object(
                bot._inbound_turn_normalizer,
                "register_batch_media_attachments",
                new=fake_register_batch_media_attachments,
            ),
            patch.object(
                bot._inbound_turn_normalizer,
                "build_dispatch_payload_with_attachments",
                new=fake_build_payload,
            ),
            patch.object(bot._turn_controller, "_has_newer_unresponded_in_thread", return_value=False),
            patch.object(ResponsePayloadPreparer, "_log_dispatch_latency"),
            patch(
                "mindroom.response_runner.ResponseRunner._generate_response_locked",
                new=fake_generate_response_locked,
            ),
        ):
            for event, requester_user_id in (
                (text_event, "@alice:localhost"),
                (image_event, "@bob:localhost"),
            ):
                await _enqueue_for_dispatch(
                    bot,
                    event,
                    room,
                    source_kind=IMAGE_SOURCE_KIND if event.event_id == "$img" else MESSAGE_SOURCE_KIND,
                    requester_user_id=requester_user_id,
                    coalescing_key=CoalescingKey(room.room_id, "$thread", requester_user_id),
                )

            await _wait_for(
                lambda: sum(len(entry.queue) for entry in bot._coalescing_gate._gates.values()) == 2,
            )
            queued_signal.finish_response_turn()
            lifecycle_lock.release()
            await bot._coalescing_gate.drain_all()
    finally:
        queued_signal.finish_response_turn()
        if lifecycle_lock.locked():
            lifecycle_lock.release()

    assert registered_media_event_ids == [["$img"]]
    assert payload_requests[0].current_attachment_ids == ["img-attachment"]
    assert captured_requests[0].attachment_ids == ("img-attachment",)
    assert captured_requests[0].model_prompt == "attachment guidance"
    assert captured_requests[0].prompt == (
        "Messages arrived while the previous response was still running. "
        "They are in chat timeline order. Respond once to the combined context:\n\n"
        "<queued_messages>\n"
        '<msg event_id="$text" from="@alice:localhost" ts="1970-01-01 00:00 UTC"><![CDATA[text follow-up]]></msg>\n'
        '<msg event_id="$img" from="@bob:localhost" ts="1970-01-01 00:00 UTC"><![CDATA[[Attached image]]]></msg>\n'
        "</queued_messages>"
    )


@pytest.mark.asyncio
async def test_room_scope_text_then_voice_live_debounce_coalesces_receive_time() -> None:
    """Room-scoped text should join a following voice event that arrives during debounce."""
    room = _make_room()
    text = _text_event(event_id="$text", body="typed first", server_timestamp=1000)
    voice = _text_event(
        event_id="$voice",
        body="voice transcript",
        server_timestamp=1001,
        source_kind=VOICE_SOURCE_KIND,
    )
    calls: list[list[str]] = []

    async def dispatch_batch(batch: CoalescedBatch) -> None:
        calls.append(list(batch.source_event_ids))

    gate = CoalescingGate(
        dispatch_batch=dispatch_batch,
        debounce_seconds=lambda: 5.0,
        is_shutting_down=lambda: False,
    )
    key = CoalescingKey("!room:localhost", None, "@user:localhost")

    await _admit_ready(gate, key, PendingEvent(event=text, room=room, source_kind="message"))
    await _admit_ready(gate, key, PendingEvent(event=voice, room=room, source_kind=VOICE_SOURCE_KIND))
    await gate.drain_all()

    assert calls == [["$text", "$voice"]]
    assert _coalescing_gate_is_idle(gate)


@pytest.mark.asyncio
async def test_room_scope_text_then_pending_voice_waits_for_voice_class_admission() -> None:
    """Room-scoped text should wait for pending raw voice admitted during debounce."""
    room = _make_room()
    text = _text_event(event_id="$text", body="typed first", server_timestamp=1000)
    voice = _text_event(
        event_id="$voice",
        body="voice transcript",
        server_timestamp=1001,
        source_kind=VOICE_SOURCE_KIND,
    )
    release_voice = asyncio.Event()
    calls: list[list[str]] = []

    async def dispatch_batch(batch: CoalescedBatch) -> None:
        calls.append(list(batch.source_event_ids))

    async def ready_voice() -> ReadyPendingEvent:
        await release_voice.wait()
        return ReadyPendingEvent(
            pending_event=PendingEvent(event=voice, room=room, source_kind=VOICE_SOURCE_KIND),
        )

    gate = CoalescingGate(
        dispatch_batch=dispatch_batch,
        debounce_seconds=lambda: 0.05,
        is_shutting_down=lambda: False,
    )
    key = CoalescingKey("!room:localhost", None, "@user:localhost")

    await _admit_ready(gate, key, PendingEvent(event=text, room=room, source_kind="message"))
    voice_slot = gate.enter_lane(room_id=room.room_id, sender_id="@user:localhost")
    gate.submit_lane_slot(
        voice_slot,
        key=key,
        source_event_id="$voice",
        source_kind=VOICE_SOURCE_KIND,
        ready_task=asyncio.create_task(ready_voice()),
    )
    await asyncio.sleep(0.07)

    assert calls == []

    release_voice.set()
    await _wait_for(lambda: calls == [["$text", "$voice"]], deadline_seconds=0.2)
    assert _coalescing_gate_is_idle(gate)


@pytest.mark.asyncio
async def test_voice_ready_release_combines_same_thread_backlog_in_receipt_order() -> None:
    """Voice resolving late releases the whole same-thread backlog as one receipt-ordered turn."""
    room = _make_room()
    first = _text_event(event_id="$typed1", body="first", server_timestamp=1000, thread_id="$thread-a")
    second = _text_event(event_id="$typed2", body="second", server_timestamp=1002, thread_id="$thread-a")
    voice = _text_event(
        event_id="$voice",
        body="voice transcript",
        server_timestamp=1001,
        thread_id="$thread-a",
        source_kind=VOICE_SOURCE_KIND,
    )
    text_key = CoalescingKey(room.room_id, "$thread-a", "@user:localhost")
    release_voice = asyncio.Event()
    calls: list[list[str]] = []

    async def ready_voice() -> ReadyPendingEvent:
        await release_voice.wait()
        return ReadyPendingEvent(
            pending_event=PendingEvent(event=voice, room=room, source_kind=VOICE_SOURCE_KIND),
        )

    async def dispatch_batch(batch: CoalescedBatch) -> None:
        calls.append(list(batch.source_event_ids))

    gate = CoalescingGate(
        dispatch_batch=dispatch_batch,
        debounce_seconds=lambda: 0.02,
        is_shutting_down=lambda: False,
    )

    start = time.monotonic() - 1.0
    await gate.admit(
        text_key,
        ready_result=ReadyPendingEvent(pending_event=PendingEvent(event=first, room=room, source_kind="message")),
        receipt_time=start,
        source_event_id="$typed1",
        source_kind="message",
    )
    voice_slot = gate.enter_lane(room_id=room.room_id, sender_id="@user:localhost", receipt_time=start + 0.005)
    gate.submit_lane_slot(
        voice_slot,
        key=text_key,
        source_event_id="$voice",
        source_kind=VOICE_SOURCE_KIND,
        ready_task=asyncio.create_task(ready_voice()),
    )
    await gate.admit(
        text_key,
        ready_result=ReadyPendingEvent(pending_event=PendingEvent(event=second, room=room, source_kind="message")),
        receipt_time=start + 0.5,
        source_event_id="$typed2",
        source_kind="message",
    )
    await asyncio.sleep(0.05)

    assert calls == []

    release_voice.set()
    await _wait_for(lambda: calls == [["$typed1", "$voice", "$typed2"]], deadline_seconds=0.5)
    assert _coalescing_gate_is_idle(gate)


@pytest.mark.asyncio
async def test_command_executes_immediately_despite_unresolved_ingress(tmp_path: Path) -> None:
    """A command never waits behind unresolved late-ready ingress from the same sender."""
    bot = _make_bot(tmp_path)
    room = _make_room()
    command_event = _text_event(event_id="$cmd", body="!help", server_timestamp=1001)
    calls: list[tuple[str, list[str]]] = []

    async def record_dispatch(
        _room: nio.MatrixRoom,
        dispatched_event: nio.RoomMessageText,
        _requester_user_id: str,
        *,
        media_events: list[object] | None = None,
        handled_turn: TurnRecord | None = None,
        **_metadata: object,
    ) -> None:
        _ = media_events, handled_turn
        calls.append((dispatched_event.body, _handled_turn_source_event_ids(handled_turn)))

    reservation_owner = bot._turn_controller.reserve_prompt_ingress_order(room, "@user:localhost")
    try:
        with patch.object(bot._turn_controller, "_dispatch_text_message", new=AsyncMock(side_effect=record_dispatch)):
            await bot._turn_controller.handle_text_event(room, command_event)

        assert calls == [("!help", ["$cmd"])]
    finally:
        await reservation_owner.release()


@pytest.mark.asyncio
async def test_interrupted_claimed_admission_is_retried_on_next_drain() -> None:
    """A cancelled drain should not lose queued or still-resolving admissions."""
    room = _make_room()
    key = CoalescingKey(room.room_id, "$thread", "@user:localhost")
    first = _text_event(event_id="$first", body="first")
    second = _text_event(event_id="$second", body="second")
    release_second = asyncio.Event()
    calls: list[list[str]] = []

    async def ready_second() -> ReadyPendingEvent:
        await release_second.wait()
        return ReadyPendingEvent(
            pending_event=PendingEvent(event=second, room=room, source_kind="message"),
        )

    async def dispatch_batch(batch: CoalescedBatch) -> None:
        calls.append(list(batch.source_event_ids))

    gate = CoalescingGate(
        dispatch_batch=dispatch_batch,
        debounce_seconds=lambda: 0.0,
        is_shutting_down=lambda: False,
    )

    await _admit_ready(gate, key, PendingEvent(event=first, room=room, source_kind="message"))
    second_slot = gate.enter_lane(room_id=room.room_id, sender_id="@user:localhost")
    gate.submit_lane_slot(
        second_slot,
        key=key,
        source_event_id="$second",
        source_kind="message",
        ready_task=asyncio.create_task(ready_second()),
    )
    [gate_entry] = gate._gates.values()
    drain_task = gate_entry.drain_task
    assert drain_task is not None
    await asyncio.sleep(0.02)

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

    release_second.set()
    await gate.drain_all()

    assert calls == [["$first", "$second"]]
    assert _coalescing_gate_is_idle(gate)


@pytest.mark.asyncio
async def test_voice_handoff_buffers_same_thread_followups_while_in_flight() -> None:
    """Follow-ups on the admitted voice thread should buffer behind the in-flight voice dispatch."""
    room = _make_room()
    voice = _text_event(
        event_id="$voice",
        body="voice transcript",
        server_timestamp=1000,
        source_kind=VOICE_SOURCE_KIND,
    )
    followup = _text_event(
        event_id="$followup",
        body="follow-up",
        server_timestamp=1001,
        thread_id="$voice_thread",
    )
    resolved_key = CoalescingKey(room.room_id, "$voice_thread", "@user:localhost")
    entered_voice_dispatch = asyncio.Event()
    release_voice_dispatch = asyncio.Event()
    calls: list[list[str]] = []

    async def ready_voice() -> ReadyPendingEvent:
        return ReadyPendingEvent(
            pending_event=PendingEvent(event=voice, room=room, source_kind=VOICE_SOURCE_KIND),
        )

    async def dispatch_batch(batch: CoalescedBatch) -> None:
        calls.append(list(batch.source_event_ids))
        if batch.source_event_ids == ["$voice"]:
            entered_voice_dispatch.set()
            await release_voice_dispatch.wait()

    gate = CoalescingGate(
        dispatch_batch=dispatch_batch,
        debounce_seconds=lambda: 0.0,
        is_shutting_down=lambda: False,
    )

    voice_slot = gate.enter_lane(room_id=room.room_id, sender_id="@user:localhost")
    gate.submit_lane_slot(
        voice_slot,
        key=resolved_key,
        source_event_id="$voice",
        source_kind=VOICE_SOURCE_KIND,
        ready_task=asyncio.create_task(ready_voice()),
    )
    await asyncio.wait_for(entered_voice_dispatch.wait(), timeout=0.2)
    await _admit_ready(gate, resolved_key, PendingEvent(event=followup, room=room, source_kind="message"))
    await asyncio.sleep(0.02)

    assert calls == [["$voice"]]

    release_voice_dispatch.set()
    await _wait_for(lambda: calls == [["$voice"], ["$followup"]], deadline_seconds=0.2)
    assert _coalescing_gate_is_idle(gate)


@pytest.mark.asyncio
async def test_voice_before_text_uses_stable_admission_key() -> None:
    """Voice admitted with a canonical thread key should wait for readiness during debounce."""
    room = _make_room()
    resolved_key = CoalescingKey(room.room_id, "$thread-root", "@user:localhost")
    voice = _text_event(
        event_id="$voice",
        body="voice transcript",
        server_timestamp=1000,
        source_kind=VOICE_SOURCE_KIND,
    )
    typed = _text_event(
        event_id="$typed",
        body="typed follow-up",
        server_timestamp=1001,
        thread_id="$thread-root",
    )
    release_voice = asyncio.Event()
    calls: list[tuple[CoalescingKey, list[str]]] = []

    async def ready_voice() -> ReadyPendingEvent:
        await release_voice.wait()
        return ReadyPendingEvent(
            pending_event=PendingEvent(event=voice, room=room, source_kind=VOICE_SOURCE_KIND),
        )

    async def dispatch_batch(batch: CoalescedBatch) -> None:
        calls.append((batch.coalescing_key, list(batch.source_event_ids)))

    gate = CoalescingGate(
        dispatch_batch=dispatch_batch,
        debounce_seconds=lambda: 0.05,
        is_shutting_down=lambda: False,
    )

    voice_slot = gate.enter_lane(room_id=room.room_id, sender_id="@user:localhost")
    gate.submit_lane_slot(
        voice_slot,
        key=resolved_key,
        source_event_id="$voice",
        source_kind=VOICE_SOURCE_KIND,
        ready_task=asyncio.create_task(ready_voice()),
    )
    await asyncio.sleep(0.005)
    await _admit_ready(gate, resolved_key, PendingEvent(event=typed, room=room, source_kind="message"))
    await asyncio.sleep(0.06)
    assert calls == []

    release_voice.set()
    await _wait_for(lambda: calls == [(resolved_key, ["$voice", "$typed"])], deadline_seconds=0.2)
    assert _coalescing_gate_is_idle(gate)


@pytest.mark.asyncio
async def test_text_before_voice_uses_stable_admission_key() -> None:
    """A later voice admitted with the same canonical key should join queued text."""
    room = _make_room()
    resolved_key = CoalescingKey(room.room_id, "$thread-root", "@user:localhost")
    typed = _text_event(
        event_id="$typed",
        body="typed follow-up",
        server_timestamp=1000,
        thread_id="$thread-root",
    )
    voice = _text_event(
        event_id="$voice",
        body="voice transcript",
        server_timestamp=1001,
        source_kind=VOICE_SOURCE_KIND,
    )
    release_voice = asyncio.Event()
    calls: list[tuple[CoalescingKey, list[str]]] = []

    async def ready_voice() -> ReadyPendingEvent:
        await release_voice.wait()
        return ReadyPendingEvent(
            pending_event=PendingEvent(event=voice, room=room, source_kind=VOICE_SOURCE_KIND),
        )

    async def dispatch_batch(batch: CoalescedBatch) -> None:
        calls.append((batch.coalescing_key, list(batch.source_event_ids)))

    gate = CoalescingGate(
        dispatch_batch=dispatch_batch,
        debounce_seconds=lambda: 0.05,
        is_shutting_down=lambda: False,
    )

    await _admit_ready(gate, resolved_key, PendingEvent(event=typed, room=room, source_kind="message"))
    voice_slot = gate.enter_lane(room_id=room.room_id, sender_id="@user:localhost")
    gate.submit_lane_slot(
        voice_slot,
        key=resolved_key,
        source_event_id="$voice",
        source_kind=VOICE_SOURCE_KIND,
        ready_task=asyncio.create_task(ready_voice()),
    )
    await asyncio.sleep(0.06)
    assert calls == []

    release_voice.set()
    await _wait_for(lambda: calls == [(resolved_key, ["$typed", "$voice"])], deadline_seconds=0.2)
    assert _coalescing_gate_is_idle(gate)


@pytest.mark.asyncio
async def test_plain_reply_voice_resolution_batches_related_text() -> None:
    """A resolved voice should merge related text that waited behind audio readiness."""
    room = _make_room()
    root_key = CoalescingKey(room.room_id, "$thread-root", "@user:localhost")
    voice = _text_event(
        event_id="$voice",
        body="voice transcript",
        server_timestamp=1000,
        source_kind=VOICE_SOURCE_KIND,
    )
    typed = _text_event(
        event_id="$typed",
        body="typed follow-up",
        server_timestamp=1001,
        thread_id="$thread-root",
    )
    release_voice = asyncio.Event()
    calls: list[list[str]] = []

    async def ready_voice() -> ReadyPendingEvent:
        await release_voice.wait()
        return ReadyPendingEvent(
            pending_event=PendingEvent(event=voice, room=room, source_kind=VOICE_SOURCE_KIND),
        )

    async def dispatch_batch(batch: CoalescedBatch) -> None:
        calls.append(list(batch.source_event_ids))

    gate = CoalescingGate(
        dispatch_batch=dispatch_batch,
        debounce_seconds=lambda: 0.0,
        is_shutting_down=lambda: False,
    )

    voice_slot = gate.enter_lane(room_id=room.room_id, sender_id="@user:localhost")
    gate.submit_lane_slot(
        voice_slot,
        key=root_key,
        source_event_id="$voice",
        source_kind=VOICE_SOURCE_KIND,
        ready_task=asyncio.create_task(ready_voice()),
    )
    await _admit_ready(gate, root_key, PendingEvent(event=typed, room=room, source_kind="message"))
    await asyncio.sleep(0.02)
    assert calls == []

    release_voice.set()
    await _wait_for(lambda: calls == [["$voice", "$typed"]], deadline_seconds=0.2)
    assert _coalescing_gate_is_idle(gate)


@pytest.mark.asyncio
async def test_text_first_waits_for_plain_reply_voice_ready_during_debounce() -> None:
    """A later voice reply may still belong to a text gate that is debouncing."""
    room = _make_room()
    root_key = CoalescingKey(room.room_id, "$thread-root", "@user:localhost")
    typed = _text_event(
        event_id="$typed",
        body="typed follow-up",
        server_timestamp=1000,
        thread_id="$thread-root",
    )
    voice = _text_event(
        event_id="$voice",
        body="voice transcript",
        server_timestamp=1001,
        source_kind=VOICE_SOURCE_KIND,
    )
    release_voice = asyncio.Event()
    calls: list[list[str]] = []

    async def ready_voice() -> ReadyPendingEvent:
        await release_voice.wait()
        return ReadyPendingEvent(
            pending_event=PendingEvent(event=voice, room=room, source_kind=VOICE_SOURCE_KIND),
        )

    async def dispatch_batch(batch: CoalescedBatch) -> None:
        calls.append(list(batch.source_event_ids))

    gate = CoalescingGate(
        dispatch_batch=dispatch_batch,
        debounce_seconds=lambda: 0.03,
        is_shutting_down=lambda: False,
    )

    await _admit_ready(gate, root_key, PendingEvent(event=typed, room=room, source_kind="message"))
    voice_slot = gate.enter_lane(room_id=room.room_id, sender_id="@user:localhost")
    gate.submit_lane_slot(
        voice_slot,
        key=root_key,
        source_event_id="$voice",
        source_kind=VOICE_SOURCE_KIND,
        ready_task=asyncio.create_task(ready_voice()),
    )

    await asyncio.sleep(0.06)
    assert calls == []

    release_voice.set()
    await _wait_for(lambda: calls == [["$typed", "$voice"]], deadline_seconds=0.2)
    assert _coalescing_gate_is_idle(gate)


@pytest.mark.asyncio
async def test_later_different_thread_voice_does_not_hold_earlier_text() -> None:
    """A later voice from another sender's thread must not hold earlier text."""
    room = _make_room()
    text_key = CoalescingKey(room.room_id, "$thread-a-root", "@user:localhost")
    voice_root_key = CoalescingKey(room.room_id, "$thread-b-root", "@voice-sender:localhost")
    typed = _text_event(
        event_id="$typed",
        body="typed follow-up",
        server_timestamp=1000,
        thread_id="$thread-a-root",
    )
    voice = _text_event(
        event_id="$voice",
        body="voice transcript",
        sender="@voice-sender:localhost",
        server_timestamp=1001,
        source_kind=VOICE_SOURCE_KIND,
    )
    release_voice = asyncio.Event()
    calls: list[list[str]] = []

    async def ready_voice() -> ReadyPendingEvent:
        await release_voice.wait()
        return ReadyPendingEvent(
            pending_event=PendingEvent(
                event=voice,
                room=room,
                source_kind=VOICE_SOURCE_KIND,
                requester_user_id="@voice-sender:localhost",
            ),
        )

    async def dispatch_batch(batch: CoalescedBatch) -> None:
        calls.append(list(batch.source_event_ids))

    gate = CoalescingGate(
        dispatch_batch=dispatch_batch,
        debounce_seconds=lambda: 0.03,
        is_shutting_down=lambda: False,
    )

    await _admit_ready(gate, text_key, PendingEvent(event=typed, room=room, source_kind="message"))
    await asyncio.sleep(0.005)
    voice_slot = gate.enter_lane(room_id=room.room_id, sender_id="@voice-sender:localhost")
    gate.submit_lane_slot(
        voice_slot,
        key=voice_root_key,
        source_event_id="$voice",
        source_kind=VOICE_SOURCE_KIND,
        ready_task=asyncio.create_task(ready_voice()),
    )
    await asyncio.sleep(0.06)
    assert calls == [["$typed"]]

    release_voice.set()
    await _wait_for(lambda: calls == [["$typed"], ["$voice"]], deadline_seconds=0.2)
    assert _coalescing_gate_is_idle(gate)


@pytest.mark.asyncio
async def test_failed_room_voice_does_not_coalesce_surviving_room_roots() -> None:
    """A failed voice admission should not make unrelated room roots share a turn."""
    room = _make_room()
    key = CoalescingKey(room.room_id, None, "@user:localhost")
    first = _text_event(event_id="$first", body="first", server_timestamp=1000)
    second = _text_event(event_id="$second", body="second", server_timestamp=1002)
    calls: list[list[str]] = []

    async def failed_voice() -> ReadyPendingEvent:
        msg = "stt failed"
        raise RuntimeError(msg)

    async def dispatch_batch(batch: CoalescedBatch) -> None:
        calls.append(list(batch.source_event_ids))

    gate = CoalescingGate(
        dispatch_batch=dispatch_batch,
        debounce_seconds=lambda: 5.0,
        is_shutting_down=lambda: False,
    )

    await _admit_ready(gate, key, PendingEvent(event=first, room=room, source_kind="message"))
    voice_slot = gate.enter_lane(room_id=room.room_id, sender_id="@user:localhost")
    gate.submit_lane_slot(
        voice_slot,
        key=key,
        source_event_id="$voice",
        source_kind=VOICE_SOURCE_KIND,
        ready_task=asyncio.create_task(failed_voice()),
    )
    await _admit_ready(gate, key, PendingEvent(event=second, room=room, source_kind="message"))

    await gate.drain_all()

    assert calls == [["$first"], ["$second"]]
    assert _coalescing_gate_is_idle(gate)


@pytest.mark.asyncio
async def test_voice_admissions_resolving_to_different_threads_do_not_coalesce() -> None:
    """Voice admissions in different resolved threads must stay separate turns."""
    room = _make_room()
    first_key = CoalescingKey(room.room_id, "$post_one", "@user:localhost")
    second_key = CoalescingKey(room.room_id, "$post_two", "@user:localhost")
    first_voice = _text_event(
        event_id="$voice1",
        body="first voice",
        server_timestamp=1000,
        thread_id="$post_one",
        source_kind=VOICE_SOURCE_KIND,
    )
    second_voice = _text_event(
        event_id="$voice2",
        body="second voice",
        server_timestamp=1001,
        thread_id="$post_two",
        source_kind=VOICE_SOURCE_KIND,
    )
    calls: list[tuple[CoalescingKey, list[str]]] = []

    async def ready_voice(event: nio.RoomMessageText) -> ReadyPendingEvent:
        return ReadyPendingEvent(
            pending_event=PendingEvent(event=event, room=room, source_kind=VOICE_SOURCE_KIND),
        )

    async def dispatch_batch(batch: CoalescedBatch) -> None:
        calls.append((batch.coalescing_key, list(batch.source_event_ids)))

    gate = CoalescingGate(
        dispatch_batch=dispatch_batch,
        debounce_seconds=lambda: 5.0,
        is_shutting_down=lambda: False,
    )

    first_slot = gate.enter_lane(room_id=room.room_id, sender_id="@user:localhost")
    second_slot = gate.enter_lane(room_id=room.room_id, sender_id="@user:localhost")
    gate.submit_lane_slot(
        first_slot,
        key=first_key,
        source_event_id="$voice1",
        source_kind=VOICE_SOURCE_KIND,
        ready_task=asyncio.create_task(ready_voice(first_voice)),
    )
    gate.submit_lane_slot(
        second_slot,
        key=second_key,
        source_event_id="$voice2",
        source_kind=VOICE_SOURCE_KIND,
        ready_task=asyncio.create_task(ready_voice(second_voice)),
    )

    await gate.drain_all()

    assert calls == [
        (first_key, ["$voice1"]),
        (second_key, ["$voice2"]),
    ]
    assert _coalescing_gate_is_idle(gate)


@pytest.mark.asyncio
async def test_pending_thread_voice_does_not_capture_unrelated_thread_text() -> None:
    """A pending voice admission must not steal another sender's thread turn."""
    room = _make_room()
    voice_key = CoalescingKey(room.room_id, "$thread-a-child", "@user:localhost")
    unrelated_thread_key = CoalescingKey(room.room_id, "$thread-b-root", "@other:localhost")
    voice = _text_event(
        event_id="$voice",
        body="voice transcript",
        server_timestamp=1000,
        source_kind=VOICE_SOURCE_KIND,
    )
    typed = _text_event(
        event_id="$typed",
        body="typed follow-up",
        sender="@other:localhost",
        server_timestamp=1001,
        thread_id="$thread-b-root",
    )
    release_voice = asyncio.Event()
    calls: list[tuple[CoalescingKey, list[str]]] = []

    async def ready_voice() -> ReadyPendingEvent:
        await release_voice.wait()
        return ReadyPendingEvent(
            pending_event=PendingEvent(event=voice, room=room, source_kind=VOICE_SOURCE_KIND),
        )

    async def dispatch_batch(batch: CoalescedBatch) -> None:
        calls.append((batch.coalescing_key, list(batch.source_event_ids)))

    gate = CoalescingGate(
        dispatch_batch=dispatch_batch,
        debounce_seconds=lambda: 0.0,
        is_shutting_down=lambda: False,
    )

    voice_slot = gate.enter_lane(room_id=room.room_id, sender_id="@user:localhost")
    gate.submit_lane_slot(
        voice_slot,
        key=voice_key,
        source_event_id="$voice",
        source_kind=VOICE_SOURCE_KIND,
        ready_task=asyncio.create_task(ready_voice()),
    )
    await _admit_ready(
        gate,
        unrelated_thread_key,
        PendingEvent(event=typed, room=room, source_kind="message", requester_user_id="@other:localhost"),
    )
    await _wait_for(lambda: calls == [(unrelated_thread_key, ["$typed"])])

    release_voice.set()
    await _wait_for(
        lambda: (
            calls
            == [
                (unrelated_thread_key, ["$typed"]),
                (voice_key, ["$voice"]),
            ]
        ),
        deadline_seconds=0.2,
    )
    assert _coalescing_gate_is_idle(gate)


@pytest.mark.asyncio
async def test_room_scope_voice_burst_coalesces_under_null_thread_key() -> None:
    """Multiple room-scoped voice events should still be one user turn."""
    room = _make_room()
    first_voice = _text_event(
        event_id="$voice1",
        body="first voice transcript",
        server_timestamp=1000,
        source_kind=VOICE_SOURCE_KIND,
    )
    second_voice = _text_event(
        event_id="$voice2",
        body="second voice transcript",
        server_timestamp=1001,
        source_kind=VOICE_SOURCE_KIND,
    )
    calls: list[list[str]] = []

    async def dispatch_batch(batch: CoalescedBatch) -> None:
        calls.append(list(batch.source_event_ids))

    gate = CoalescingGate(
        dispatch_batch=dispatch_batch,
        debounce_seconds=lambda: 0.05,
        is_shutting_down=lambda: False,
    )
    key = CoalescingKey("!room:localhost", None, "@user:localhost")

    await _admit_ready(gate, key, PendingEvent(event=first_voice, room=room, source_kind=VOICE_SOURCE_KIND))
    await _admit_ready(gate, key, PendingEvent(event=second_voice, room=room, source_kind=VOICE_SOURCE_KIND))
    await gate.drain_all()

    assert calls == [["$voice1", "$voice2"]]
    assert _coalescing_gate_is_idle(gate)


@pytest.mark.asyncio
async def test_deferred_room_scope_voice_burst_stays_one_turn_under_null_thread_key() -> None:
    """Deferred room-scoped voice events should still be one user turn."""
    room = _make_room()
    first_voice = _text_event(
        event_id="$voice1",
        body="first voice transcript",
        server_timestamp=1000,
        source_kind=VOICE_SOURCE_KIND,
    )
    second_voice = _text_event(
        event_id="$voice2",
        body="second voice transcript",
        server_timestamp=1001,
        source_kind=VOICE_SOURCE_KIND,
    )
    key = CoalescingKey(room.room_id, None, "@user:localhost")
    calls: list[tuple[tuple[str, str | None, str], list[str]]] = []

    async def ready_voice(event: nio.RoomMessageText) -> ReadyPendingEvent:
        return ReadyPendingEvent(
            pending_event=PendingEvent(event=event, room=room, source_kind=VOICE_SOURCE_KIND),
        )

    async def dispatch_batch(batch: CoalescedBatch) -> None:
        calls.append((batch.coalescing_key, list(batch.source_event_ids)))

    gate = CoalescingGate(
        dispatch_batch=dispatch_batch,
        debounce_seconds=lambda: 0.0,
        is_shutting_down=lambda: False,
    )

    first_slot = gate.enter_lane(room_id=room.room_id, sender_id="@user:localhost")
    second_slot = gate.enter_lane(room_id=room.room_id, sender_id="@user:localhost")
    gate.submit_lane_slot(
        first_slot,
        key=key,
        source_event_id="$voice1",
        source_kind=VOICE_SOURCE_KIND,
        ready_task=asyncio.create_task(ready_voice(first_voice)),
    )
    gate.submit_lane_slot(
        second_slot,
        key=key,
        source_event_id="$voice2",
        source_kind=VOICE_SOURCE_KIND,
        ready_task=asyncio.create_task(ready_voice(second_voice)),
    )
    await gate.drain_all()

    assert calls == [(key, ["$voice1", "$voice2"])]
    assert _coalescing_gate_is_idle(gate)


@pytest.mark.asyncio
async def test_enqueue_for_dispatch_returns_while_drain_dispatch_blocks(tmp_path: Path) -> None:
    """A blocked coalescing drain must not hold later Matrix ingress callbacks."""
    bot = _make_bot(tmp_path, debounce_ms=0)
    room = _make_room()
    first = _text_event(event_id="$m1", body="first", server_timestamp=1000)
    second = _text_event(event_id="$m2", body="second", server_timestamp=1001, thread_id="$m1")
    entered_first_dispatch = asyncio.Event()
    release_first_dispatch = asyncio.Event()
    calls: list[list[str]] = []

    async def record_dispatch(
        _room: nio.MatrixRoom,
        _dispatched_event: nio.RoomMessageText,
        _requester_user_id: str,
        *,
        media_events: list[object] | None = None,
        handled_turn: TurnRecord | None = None,
        **_metadata: object,
    ) -> None:
        _ = media_events
        source_event_ids = _handled_turn_source_event_ids(handled_turn)
        calls.append(source_event_ids)
        if source_event_ids == ["$m1"]:
            entered_first_dispatch.set()
            await release_first_dispatch.wait()

    with patch.object(bot._turn_controller, "_dispatch_text_message", new=AsyncMock(side_effect=record_dispatch)):
        await asyncio.wait_for(
            _enqueue_for_dispatch(
                bot,
                first,
                room,
                source_kind="message",
                requester_user_id="@user:localhost",
            ),
            timeout=0.05,
        )
        await entered_first_dispatch.wait()

        await asyncio.wait_for(
            _enqueue_for_dispatch(
                bot,
                second,
                room,
                source_kind="message",
                requester_user_id="@user:localhost",
            ),
            timeout=0.05,
        )

        await _wait_for(lambda: calls == [["$m1"], ["$m2"]])
        assert not release_first_dispatch.is_set()

        release_first_dispatch.set()
        await bot._coalescing_gate.drain_all()

    assert calls == [["$m1"], ["$m2"]]
    assert _coalescing_gate_is_idle(bot._coalescing_gate)


def test_automation_source_kinds_are_coalescing_exempt() -> None:
    """Dispatch automation source kinds as FIFO barriers."""
    scheduled = _text_event(event_id="$scheduled", body="scheduled", source_kind="scheduled")
    hook = _text_event(event_id="$hook", body="hook", source_kind="hook")
    hook_dispatch = _text_event(event_id="$hook_dispatch", body="hook dispatch", source_kind="hook_dispatch")

    assert is_coalescing_exempt_source_kind(scheduled, "scheduled") is True
    assert is_coalescing_exempt_source_kind(hook, "hook") is True
    assert is_coalescing_exempt_source_kind(hook_dispatch, "hook_dispatch") is True


@pytest.mark.asyncio
@pytest.mark.parametrize("source_kind", ["hook", "hook_dispatch"])
async def test_coalescing_exempt_source_kinds_bypass_gate(tmp_path: Path, source_kind: str) -> None:
    """Bypass the gate only for hook-originated synthetic events."""
    bot = _make_bot(tmp_path)
    room = _make_room()
    event = _text_event(event_id=f"${source_kind}", body=f"{source_kind} task", source_kind=source_kind)
    calls: list[str] = []

    async def record_dispatch(
        _room: nio.MatrixRoom,
        dispatched_event: nio.RoomMessageText,
        _requester_user_id: str,
        *,
        media_events: list[object] | None = None,
        handled_turn: TurnRecord | None = None,
        **_metadata: object,
    ) -> None:
        _ = media_events, handled_turn
        calls.append(dispatched_event.body)

    with patch.object(bot._turn_controller, "_dispatch_text_message", new=AsyncMock(side_effect=record_dispatch)):
        await _enqueue_for_dispatch(
            bot,
            event,
            room,
            source_kind=source_kind,
            requester_user_id="@user:localhost",
        )
        await _wait_for(lambda: len(calls) == 1)

    assert calls == [f"{source_kind} task"]


@pytest.mark.asyncio
async def test_pending_dispatch_policy_preserves_active_followup_without_bypassing_modality() -> None:
    """Active follow-up policy should stay metadata while voice remains coalescible."""
    room = _make_room()
    event = PreparedTextEvent(
        sender="@user:localhost",
        event_id="$voice_followup",
        body="voice follow-up",
        source={"content": {"body": "voice follow-up", SOURCE_KIND_KEY: "voice"}},
        server_timestamp=1000,
        source_kind_override="voice",
    )
    calls: list[CoalescedBatch] = []

    async def dispatch_batch(batch: CoalescedBatch) -> None:
        calls.append(batch)

    gate = CoalescingGate(
        dispatch_batch=dispatch_batch,
        debounce_seconds=lambda: 60.0,
        is_shutting_down=lambda: False,
    )

    await _admit_ready(
        gate,
        CoalescingKey("!room:localhost", "$thread", "@user:localhost"),
        PendingEvent(
            event=event,
            room=room,
            source_kind="voice",
            dispatch_policy_source_kind=ACTIVE_THREAD_FOLLOW_UP_SOURCE_KIND,
        ),
    )
    await gate.drain_all()

    assert calls[0].source_kind == "voice"
    assert calls[0].dispatch_policy_source_kind == ACTIVE_THREAD_FOLLOW_UP_SOURCE_KIND
    dispatch_event = _build_batch_dispatch_event(calls[0])
    assert isinstance(dispatch_event, PreparedTextEvent)
    assert dispatch_event.source_kind_override == "voice"


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "spoofed_source_kind",
    [ACTIVE_THREAD_FOLLOW_UP_SOURCE_KIND, "hook", "hook_dispatch", "voice"],
)
async def test_untrusted_source_kind_content_does_not_bypass_or_promote(
    tmp_path: Path,
    spoofed_source_kind: str,
) -> None:
    """User-controlled source_kind content should not become trusted dispatch policy."""
    bot = _make_bot(tmp_path, debounce_ms=1000)
    room = _make_room()
    event = _image_event(
        event_id=f"$spoof_{spoofed_source_kind}",
        source_kind=spoofed_source_kind,
    )
    calls: list[nio.RoomMessageImage | PreparedTextEvent] = []

    async def record_dispatch(
        _room: nio.MatrixRoom,
        dispatched_event: nio.RoomMessageImage | PreparedTextEvent,
        _requester_user_id: str,
        *,
        media_events: list[object] | None = None,
        handled_turn: TurnRecord | None = None,
        queued_notice_reservation: object | None = None,
        **_metadata: object,
    ) -> None:
        _ = media_events, handled_turn, queued_notice_reservation
        calls.append(dispatched_event)

    with patch.object(bot._turn_controller, "_dispatch_text_message", new=AsyncMock(side_effect=record_dispatch)):
        await _enqueue_for_dispatch(
            bot,
            event,
            room,
            source_kind="image",
            requester_user_id="@user:localhost",
        )
        await asyncio.sleep(0.01)
        assert calls == []

        await bot.prepare_for_sync_shutdown()

    assert len(calls) == 1
    dispatched = calls[0]
    assert isinstance(dispatched, PreparedTextEvent)
    assert dispatched.event_id == f"$spoof_{spoofed_source_kind}"
    assert dispatched.source_kind_override == IMAGE_SOURCE_KIND


@pytest.mark.asyncio
async def test_bypass_preserves_fifo_order_behind_existing_normal_work() -> None:
    """Hook-originated bypass events dispatch solo without jumping ahead of queued user work."""
    room = _make_room()
    first = _text_event(event_id="$m1", body="first", server_timestamp=1000)
    hook = _text_event(event_id="$hook", body="hook", server_timestamp=1001, source_kind="hook")
    second = _text_event(event_id="$m2", body="second", server_timestamp=1002)
    calls: list[list[str]] = []

    async def dispatch_batch(batch: CoalescedBatch) -> None:
        calls.append(list(batch.source_event_ids))

    gate = CoalescingGate(
        dispatch_batch=dispatch_batch,
        debounce_seconds=lambda: 0.02,
        is_shutting_down=lambda: False,
    )
    key = CoalescingKey("!room:localhost", None, "@user:localhost")

    await _admit_ready(gate, key, PendingEvent(event=first, room=room, source_kind="message"))
    await _admit_ready(gate, key, PendingEvent(event=hook, room=room, source_kind="hook"))
    await _admit_ready(gate, key, PendingEvent(event=second, room=room, source_kind="message"))

    await _wait_for(lambda: calls == [["$m1"], ["$hook"], ["$m2"]])

    assert _coalescing_gate_is_idle(gate)


@pytest.mark.asyncio
async def test_room_mode_voice_queued_notice_is_solo_barrier_before_nearby_normal_message() -> None:
    """Room-scoped voice notices should dispatch solo instead of joining normal debounce batches."""
    room = _make_room()
    voice = _text_event(event_id="$voice-room", body="voice transcript", server_timestamp=1000, source_kind="voice")
    normal = _text_event(event_id="$normal", body="nearby text", server_timestamp=1001)
    reservation = MagicMock()
    metadata = (
        PendingDispatchMetadata(
            kind="queued_notice_reservation",
            payload=reservation,
            close=reservation.cancel,
            requires_solo_batch=True,
        ),
    )
    calls: list[list[str]] = []

    async def dispatch_batch(batch: CoalescedBatch) -> None:
        calls.append(list(batch.source_event_ids))
        if batch.source_event_ids == ["$voice-room"]:
            assert batch.dispatch_metadata == metadata
            return
        assert batch.dispatch_metadata == ()

    gate = CoalescingGate(
        dispatch_batch=dispatch_batch,
        debounce_seconds=lambda: 5.0,
        is_shutting_down=lambda: False,
    )
    key = CoalescingKey("!room:localhost", None, "@user:localhost")

    await _admit_ready(
        gate,
        key,
        PendingEvent(event=voice, room=room, source_kind="voice", dispatch_metadata=metadata),
    )
    await _admit_ready(gate, key, PendingEvent(event=normal, room=room, source_kind="message"))

    await _wait_for(lambda: calls == [["$voice-room"], ["$normal"]], deadline_seconds=0.2)
    reservation.cancel.assert_not_called()

    await gate.drain_all()

    assert calls == [["$voice-room"], ["$normal"]]
    reservation.cancel.assert_not_called()
    assert _coalescing_gate_is_idle(gate)


@pytest.mark.asyncio
async def test_overlapping_scheduled_checkins_coalesce(tmp_path: Path) -> None:
    """Scheduled turns should buffer behind an in-flight dispatch instead of bypassing the gate."""
    bot = _make_bot(tmp_path)
    room = _make_room()
    first = _text_event(
        event_id="$m1",
        body="first scheduled",
        sender="@mindroom_test_agent:localhost",
        server_timestamp=1000,
        thread_id="$thread_root",
        source_kind="scheduled",
        original_sender="@user:localhost",
    )
    second = _text_event(
        event_id="$m2",
        body="second scheduled",
        sender="@mindroom_test_agent:localhost",
        server_timestamp=1001,
        thread_id="$thread_root",
        source_kind="scheduled",
        original_sender="@user:localhost",
    )
    entered_first_dispatch = asyncio.Event()
    release_first_dispatch = asyncio.Event()
    calls: list[list[str]] = []

    async def record_dispatch(
        _room: nio.MatrixRoom,
        _dispatched_event: nio.RoomMessageText,
        _requester_user_id: str,
        *,
        media_events: list[object] | None = None,
        handled_turn: TurnRecord | None = None,
        **_metadata: object,
    ) -> None:
        _ = media_events, handled_turn
        calls.append(_handled_turn_source_event_ids(handled_turn))
        if _handled_turn_source_event_ids(handled_turn) == ["$m1"]:
            entered_first_dispatch.set()
            await release_first_dispatch.wait()

    with patch.object(bot._turn_controller, "_dispatch_text_message", new=AsyncMock(side_effect=record_dispatch)):
        await _enqueue_for_dispatch(
            bot,
            first,
            room,
            source_kind="scheduled",
            requester_user_id="@user:localhost",
        )
        await asyncio.sleep(0.03)
        await entered_first_dispatch.wait()

        await _enqueue_for_dispatch(
            bot,
            second,
            room,
            source_kind="scheduled",
            requester_user_id="@user:localhost",
        )
        await asyncio.sleep(0.03)

        assert calls == [["$m1"]]

        release_first_dispatch.set()
        await _wait_for(lambda: calls == [["$m1"], ["$m2"]])

    assert calls == [["$m1"], ["$m2"]]


@pytest.mark.asyncio
async def test_prepare_for_sync_shutdown_waits_for_active_flush_task(tmp_path: Path) -> None:
    """Wait for an active flush task before finishing sync shutdown."""
    bot = _make_bot(tmp_path)
    bot.client.next_batch = "s_test_token"
    room = _make_room()
    event = _text_event(event_id="$m1", body="hello")
    entered_dispatch = asyncio.Event()
    release_dispatch = asyncio.Event()
    calls: list[list[str]] = []

    async def record_dispatch(
        _room: nio.MatrixRoom,
        _dispatched_event: nio.RoomMessageText,
        _requester_user_id: str,
        *,
        media_events: list[object] | None = None,
        handled_turn: TurnRecord | None = None,
        **_metadata: object,
    ) -> None:
        _ = media_events, handled_turn
        calls.append(_handled_turn_source_event_ids(handled_turn))
        entered_dispatch.set()
        await release_dispatch.wait()

    with patch.object(bot._turn_controller, "_dispatch_text_message", new=AsyncMock(side_effect=record_dispatch)):
        await _enqueue_for_dispatch(
            bot,
            event,
            room,
            source_kind="message",
            requester_user_id="@user:localhost",
        )
        await asyncio.sleep(0.03)
        await entered_dispatch.wait()

        shutdown_task = asyncio.create_task(bot.prepare_for_sync_shutdown())
        await asyncio.sleep(0.01)
        assert shutdown_task.done() is False

        release_dispatch.set()
        await shutdown_task

    assert calls == [["$m1"]]
    assert _coalescing_gate_is_idle(bot._coalescing_gate)


@pytest.mark.asyncio
async def test_prepare_for_sync_shutdown_drains_pending_debounced_messages(tmp_path: Path) -> None:
    """Flush any queued debounced messages during sync shutdown."""
    bot = _make_bot(tmp_path, debounce_ms=1000)
    room = _make_room()
    event = _text_event(event_id="$m1", body="hello")
    calls: list[tuple[str, list[str]]] = []

    async def record_dispatch(
        _room: nio.MatrixRoom,
        dispatched_event: nio.RoomMessageText,
        _requester_user_id: str,
        *,
        media_events: list[object] | None = None,
        handled_turn: TurnRecord | None = None,
        **_metadata: object,
    ) -> None:
        _ = media_events, handled_turn
        calls.append((dispatched_event.body, _handled_turn_source_event_ids(handled_turn)))

    with patch.object(bot._turn_controller, "_dispatch_text_message", new=AsyncMock(side_effect=record_dispatch)):
        await _enqueue_for_dispatch(
            bot,
            event,
            room,
            source_kind="message",
            requester_user_id="@user:localhost",
        )
        await bot.prepare_for_sync_shutdown()

    assert calls == [("hello", ["$m1"])]
    assert _coalescing_gate_is_idle(bot._coalescing_gate)


@pytest.mark.asyncio
async def test_prepare_for_sync_shutdown_drains_pending_media_debounce(tmp_path: Path) -> None:
    """Flush a held media batch immediately when shutdown interrupts the debounce window."""
    bot = _make_bot(tmp_path, debounce_ms=60_000)
    room = _make_room()
    event = _image_event(event_id="$img1", server_timestamp=1000, thread_id="$thread")
    calls: list[list[str]] = []

    async def record_dispatch(
        _room: nio.MatrixRoom,
        _dispatched_event: nio.RoomMessageText,
        _requester_user_id: str,
        *,
        media_events: list[object] | None = None,
        handled_turn: TurnRecord | None = None,
        **_metadata: object,
    ) -> None:
        _ = media_events
        calls.append(_handled_turn_source_event_ids(handled_turn))

    with patch.object(bot._turn_controller, "_dispatch_text_message", new=AsyncMock(side_effect=record_dispatch)):
        await _enqueue_for_dispatch(
            bot,
            event,
            room,
            source_kind="image",
            requester_user_id="@user:localhost",
        )
        await asyncio.sleep(0.02)
        assert calls == []

        await bot.prepare_for_sync_shutdown()

    assert calls == [["$img1"]]
    assert _coalescing_gate_is_idle(bot._coalescing_gate)


@pytest.mark.asyncio
async def test_shutdown_during_in_flight_dispatch_flushes_remaining_without_wait(tmp_path: Path) -> None:
    """Shutdown during an in-flight dispatch flushes the queued remainder immediately."""
    bot = _make_bot(tmp_path, debounce_ms=10)
    room = _make_room()
    first = _text_event(event_id="$m1", body="first", server_timestamp=1000)
    second = _text_event(event_id="$m2", body="second", server_timestamp=1001)
    entered_dispatch = asyncio.Event()
    release_dispatch = asyncio.Event()
    calls: list[tuple[str, list[str]]] = []

    async def record_dispatch(
        _room: nio.MatrixRoom,
        dispatched_event: nio.RoomMessageText,
        _requester_user_id: str,
        *,
        media_events: list[object] | None = None,
        handled_turn: TurnRecord | None = None,
        **_metadata: object,
    ) -> None:
        _ = media_events, handled_turn
        calls.append((dispatched_event.body, _handled_turn_source_event_ids(handled_turn)))
        if _handled_turn_source_event_ids(handled_turn) == ["$m1"]:
            entered_dispatch.set()
            await release_dispatch.wait()

    with patch.object(bot._turn_controller, "_dispatch_text_message", new=AsyncMock(side_effect=record_dispatch)):
        await _enqueue_for_dispatch(
            bot,
            first,
            room,
            source_kind="message",
            requester_user_id="@user:localhost",
        )
        await asyncio.sleep(0.03)
        await entered_dispatch.wait()

        # Enqueue another message while first is in-flight
        await _enqueue_for_dispatch(
            bot,
            second,
            room,
            source_kind="message",
            requester_user_id="@user:localhost",
        )

        # Start shutdown — should wait for in-flight, then flush remaining without waiting
        shutdown_task = asyncio.create_task(bot.prepare_for_sync_shutdown())
        await asyncio.sleep(0.01)
        assert shutdown_task.done() is False

        release_dispatch.set()
        await shutdown_task

    assert calls == [("first", ["$m1"]), ("second", ["$m2"])]
    assert _coalescing_gate_is_idle(bot._coalescing_gate)


@pytest.mark.asyncio
async def test_thread_followups_dispatch_while_first_turn_root_in_flight(tmp_path: Path) -> None:
    """Thread replies dispatch as their own turn without waiting on the root's in-flight dispatch."""
    bot = _make_bot(tmp_path, debounce_ms=0)
    room = _make_room()
    first = _text_event(event_id="$m1", body="first")
    followups = [
        _text_event(event_id="$m2", body="second", thread_id="$m1"),
        _text_event(event_id="$m3", body="third", thread_id="$m1"),
    ]
    entered_dispatch = asyncio.Event()
    release_first_dispatch = asyncio.Event()
    calls: list[list[str]] = []

    async def record_dispatch(
        _room: nio.MatrixRoom,
        _dispatched_event: nio.RoomMessageText,
        _requester_user_id: str,
        *,
        media_events: list[object] | None = None,
        handled_turn: TurnRecord | None = None,
        **_metadata: object,
    ) -> None:
        _ = media_events, handled_turn
        calls.append(_handled_turn_source_event_ids(handled_turn))
        if len(calls) == 1:
            entered_dispatch.set()
            await release_first_dispatch.wait()

    with patch.object(bot._turn_controller, "_dispatch_text_message", new=AsyncMock(side_effect=record_dispatch)):
        first_task = asyncio.create_task(
            _enqueue_for_dispatch(
                bot,
                first,
                room,
                source_kind="message",
                requester_user_id="@user:localhost",
            ),
        )
        await entered_dispatch.wait()

        for followup in followups:
            await _enqueue_for_dispatch(
                bot,
                followup,
                room,
                source_kind="message",
                requester_user_id="@user:localhost",
            )

        await _wait_for(lambda: ["$m2", "$m3"] in calls)
        assert not release_first_dispatch.is_set()
        assert calls == [["$m1"], ["$m2", "$m3"]]

        release_first_dispatch.set()
        await first_task
        await bot._coalescing_gate.drain_all()

    assert calls == [["$m1"], ["$m2", "$m3"]]
    assert _coalescing_gate_is_idle(bot._coalescing_gate)


@pytest.mark.asyncio
async def test_active_approval_fallthrough_reserves_before_async_approval_lookup(tmp_path: Path) -> None:
    """An approval reply that falls through to normal text must keep receive order."""
    bot = _make_bot(tmp_path, debounce_ms=0)
    room = _make_room()
    approval_lookup_started = asyncio.Event()
    release_approval_lookup = asyncio.Event()
    batches: list[list[str]] = []

    async def dispatch_batch(batch: CoalescedBatch) -> None:
        batches.append(batch.source_event_ids)

    async def maybe_approval(
        *,
        event: nio.RoomMessageText,
        **_kwargs: object,
    ) -> bool:
        if event.event_id == "$first:localhost":
            approval_lookup_started.set()
            await release_approval_lookup.wait()
        return False

    bot._coalescing_gate._dispatch_batch = dispatch_batch
    first = _reply_event(
        event_id="$first:localhost",
        body="not my approval",
        reply_to_event_id="$approval-card:localhost",
        server_timestamp=1_000_000,
    )
    later = _text_event(event_id="$later:localhost", body="later", server_timestamp=1_000_001)

    with (
        patch("mindroom.bot.is_process_active_approval_card", return_value=True),
        patch("mindroom.bot.maybe_handle_tool_approval_reply", side_effect=maybe_approval),
    ):
        first_task = asyncio.create_task(bot._on_message(room, first))
        await _wait_for(approval_lookup_started.is_set)
        await bot._on_message(room, later)
        await asyncio.sleep(0.05)

        assert batches == []

        release_approval_lookup.set()
        await first_task
        await bot._coalescing_gate.drain_all()

    assert batches == [["$first:localhost"], ["$later:localhost"]]


@pytest.mark.asyncio
async def test_trusted_relay_approval_fallthrough_reserves_effective_requester(tmp_path: Path) -> None:
    """Approval fallthrough reservations must use original human requester for trusted relays."""
    bot = _make_bot(tmp_path, debounce_ms=0)
    room = _make_room()
    approval_lookup_started = asyncio.Event()
    release_approval_lookup = asyncio.Event()
    batches: list[list[str]] = []

    async def dispatch_batch(batch: CoalescedBatch) -> None:
        batches.append(batch.source_event_ids)

    async def maybe_approval(
        *,
        event: nio.RoomMessageText,
        **_kwargs: object,
    ) -> bool:
        if event.event_id == "$relay-first:localhost":
            approval_lookup_started.set()
            await release_approval_lookup.wait()
        return False

    bot._coalescing_gate._dispatch_batch = dispatch_batch
    first = _reply_event(
        event_id="$relay-first:localhost",
        body="not my approval",
        reply_to_event_id="$approval-card:localhost",
        sender="@mindroom_test_agent:localhost",
        server_timestamp=1_000_000,
    )
    first.source["content"][SOURCE_KIND_KEY] = TRUSTED_INTERNAL_RELAY_SOURCE_KIND
    first.source["content"][ORIGINAL_SENDER_KEY] = "@external:example.org"
    later = _text_event(
        event_id="$later:localhost",
        body="later",
        sender="@external:example.org",
        server_timestamp=1_000_001,
    )

    with (
        patch("mindroom.bot.is_process_active_approval_card", return_value=True),
        patch("mindroom.bot.maybe_handle_tool_approval_reply", side_effect=maybe_approval),
    ):
        first_task = asyncio.create_task(bot._on_message(room, first))
        await _wait_for(approval_lookup_started.is_set)
        await bot._on_message(room, later)
        await asyncio.sleep(0.05)

        assert batches == []

        release_approval_lookup.set()
        await first_task
        await bot._coalescing_gate.drain_all()

    assert batches == [["$relay-first:localhost"], ["$later:localhost"]]


@pytest.mark.asyncio
async def test_zero_debounce_immediate_flush_logs_pending_count_before_clearing() -> None:
    """Immediate-flush telemetry should report the batch size before _flush clears pending."""
    room = _make_room()
    gate = CoalescingGate(
        dispatch_batch=AsyncMock(),
        debounce_seconds=lambda: 0.0,
        is_shutting_down=lambda: False,
    )

    with patch("mindroom.coalescing.emit_elapsed_timing") as mock_emit:
        await _admit_ready(
            gate,
            CoalescingKey("!room:localhost", None, "@user:localhost"),
            PendingEvent(
                event=_text_event(event_id="$m1", body="first"),
                room=room,
                source_kind="message",
            ),
        )

    immediate_flush_calls = [
        call
        for call in mock_emit.call_args_list
        if call.args and call.args[0] == "coalescing_gate.enqueue" and call.kwargs.get("path") == "zero_debounce"
    ]
    assert len(immediate_flush_calls) == 1
    assert immediate_flush_calls[0].kwargs["pending_count"] == 1
    assert immediate_flush_calls[0].kwargs["flush_outcome"] == "scheduled_drain"


@pytest.mark.asyncio
async def test_enqueue_for_dispatch_timing_events_include_explicit_scope(tmp_path: Path) -> None:
    """Pre-dispatch handoff telemetry should carry the source-event timing scope explicitly."""
    bot = _make_bot(tmp_path)
    room = _make_room()
    event = _text_event(event_id="$m1", body="hello")

    with (
        patch.object(bot._coalescing_gate, "admit", new=AsyncMock()),
        patch("mindroom.turn_controller.emit_elapsed_timing") as mock_emit,
    ):
        await _enqueue_for_dispatch(
            bot,
            event,
            room,
            source_kind="message",
            requester_user_id="@user:localhost",
            coalescing_key=CoalescingKey(room.room_id, None, "@user:localhost"),
        )

    handoff_calls = [
        call
        for call in mock_emit.call_args_list
        if call.args
        and isinstance(call.args[0], str)
        and call.args[0].startswith("ingress_handoff.enqueue_for_dispatch")
    ]
    assert handoff_calls
    assert all(call.kwargs["timing_scope"] == "$m1" for call in handoff_calls)


@pytest.mark.asyncio
async def test_matrix_ingress_logging_includes_receive_lag(tmp_path: Path) -> None:
    """Matrix callback logs should include receive lag when origin_server_ts is present."""
    bot = _make_bot(tmp_path)
    bot.logger = MagicMock()
    bot._turn_controller.handle_text_event = AsyncMock()
    room = _make_room()
    event = _text_event(event_id="$m1", body="hello", server_timestamp=1000, thread_id="$thread")

    with patch("mindroom.bot.time.time", return_value=2.5):
        await bot._on_message(room, event)

    bot.logger.info.assert_any_call(
        "matrix_event_callback_started",
        callback="message",
        event_id="$m1",
        room_id="!room:localhost",
        agent_name="test_agent",
        receive_timestamp_ms=2500,
        origin_server_ts_ms=1000,
        matrix_event_receive_lag_ms=1500.0,
    )
    bot._turn_controller.handle_text_event.assert_awaited_once()
    assert bot._turn_controller.handle_text_event.await_args.args == (room, event)
    assert isinstance(bot._turn_controller.handle_text_event.await_args.kwargs["receipt_time"], float)


@pytest.mark.asyncio
async def test_matrix_ingress_logging_handles_missing_origin_timestamp(tmp_path: Path) -> None:
    """Matrix callback logs should tolerate events without origin_server_ts."""
    bot = _make_bot(tmp_path)
    bot.logger = MagicMock()
    bot._turn_controller.handle_text_event = AsyncMock()
    room = _make_room()
    event = MagicMock(spec=nio.RoomMessageText)
    event.event_id = "$m1"
    event.sender = "@user:localhost"
    event.source = {"content": {"msgtype": "m.text", "body": "hello"}}

    with patch("mindroom.bot.time.time", return_value=2.5):
        await bot._on_message(room, event)

    log_call = next(call for call in bot.logger.info.call_args_list if call.args == ("matrix_event_callback_started",))
    assert log_call.kwargs == {
        "callback": "message",
        "event_id": "$m1",
        "room_id": "!room:localhost",
        "agent_name": "test_agent",
        "receive_timestamp_ms": 2500,
    }
    bot._turn_controller.handle_text_event.assert_awaited_once()
    assert bot._turn_controller.handle_text_event.await_args.args == (room, event)
    assert isinstance(bot._turn_controller.handle_text_event.await_args.kwargs["receipt_time"], float)


@pytest.mark.asyncio
async def test_handle_coalesced_batch_timing_events_include_dispatch_scope(tmp_path: Path) -> None:
    """Coalesced-batch telemetry emitted before dispatch should carry the batch event scope."""
    bot = _make_bot(tmp_path)
    room = _make_room()
    batch = build_coalesced_batch(
        CoalescingKey(room.room_id, None, "@user:localhost"),
        [
            PendingEvent(
                event=_text_event(event_id="$m1", body="hello"),
                room=room,
                source_kind="message",
            ),
        ],
    )

    with (
        patch.object(bot._turn_controller, "_dispatch_text_message", new=AsyncMock()),
        patch("mindroom.turn_controller.emit_elapsed_timing") as mock_emit,
    ):
        await bot._turn_controller.handle_coalesced_batch(batch)

    batch_calls = [
        call
        for call in mock_emit.call_args_list
        if call.args and isinstance(call.args[0], str) and call.args[0].startswith("coalescing.handle_batch.")
    ]
    assert batch_calls
    assert all(call.kwargs["timing_scope"] == "$m1" for call in batch_calls)


@pytest.mark.asyncio
async def test_handle_coalesced_batch_uses_batch_key_for_text_primary(tmp_path: Path) -> None:
    """A mixed batch should dispatch on its single coalescing key even when text is primary."""
    bot = _make_bot(tmp_path)
    room = _make_room()
    voice = _text_event(
        event_id="$voice",
        body="voice transcript",
        server_timestamp=1000,
        source_kind=VOICE_SOURCE_KIND,
        thread_id="$voice_thread",
    )
    typed = _text_event(event_id="$typed", body="typed follow-up", server_timestamp=1001)
    batch = build_coalesced_batch(
        CoalescingKey(room.room_id, "$voice_thread", "@user:localhost"),
        [
            PendingEvent(event=voice, room=room, source_kind=VOICE_SOURCE_KIND),
            PendingEvent(event=typed, room=room, source_kind="message"),
        ],
    )

    with patch.object(bot._turn_controller, "_dispatch_text_message", new=AsyncMock()) as mock_dispatch:
        await bot._turn_controller.handle_coalesced_batch(batch)

    dispatched_event = mock_dispatch.await_args.args[1]
    assert isinstance(dispatched_event, PreparedTextEvent)
    content = dispatched_event.source["content"]
    assert content["m.relates_to"] == {"rel_type": "m.thread", "event_id": "$voice_thread"}


def test_room_resolved_voice_batch_clears_stale_primary_thread_relation() -> None:
    """A room-resolved voice batch must not dispatch through a typed reply's stale thread relation."""
    room = _make_room()
    room_key = CoalescingKey(room.room_id, None, "@user:localhost")
    voice = _text_event(
        event_id="$voice",
        body="voice transcript",
        server_timestamp=1000,
        source_kind=VOICE_SOURCE_KIND,
    )
    typed = _text_event(
        event_id="$typed",
        body="typed reply to pending voice",
        server_timestamp=1001,
        thread_id="$voice",
    )

    batch = build_coalesced_batch(
        room_key,
        [
            PendingEvent(event=voice, room=room, source_kind=VOICE_SOURCE_KIND),
            PendingEvent(event=typed, room=room, source_kind="message"),
        ],
    )

    handoff = build_dispatch_handoff(batch)

    assert isinstance(handoff.event, PreparedTextEvent)
    assert "m.relates_to" not in handoff.event.source["content"]


def test_room_level_batch_preserves_plain_reply_relation_without_thread_target() -> None:
    """Room-level batches should preserve plain reply shape without adding thread targeting."""
    room = _make_room()
    typed_reply = _reply_event(
        event_id="$typed",
        reply_to_event_id="$voice",
        body="typed follow-up",
        server_timestamp=1001,
    )
    batch = build_coalesced_batch(
        CoalescingKey(room.room_id, None, "@user:localhost"),
        [PendingEvent(event=typed_reply, room=room, source_kind=MESSAGE_SOURCE_KIND)],
    )

    handoff = build_dispatch_handoff(batch)

    assert isinstance(handoff.event, PreparedTextEvent)
    assert handoff.event.source["content"]["m.relates_to"] == {"m.in_reply_to": {"event_id": "$voice"}}
    assert not EventInfo.from_event(handoff.event.source).can_be_thread_root


def test_room_level_batch_preserves_mentions_while_removing_stale_thread_relation() -> None:
    """Mention metadata must survive, but explicit stale threads come from the batch key."""
    room = _make_room()
    typed_reply = _text_event(
        event_id="$typed",
        body="@agent follow-up",
        server_timestamp=1001,
        thread_id="$stale-thread",
    )
    typed_reply.source["content"]["m.mentions"] = {"user_ids": ["@agent:localhost"]}
    batch = build_coalesced_batch(
        CoalescingKey(room.room_id, None, "@user:localhost"),
        [PendingEvent(event=typed_reply, room=room, source_kind=MESSAGE_SOURCE_KIND)],
    )

    handoff = build_dispatch_handoff(batch)

    assert isinstance(handoff.event, PreparedTextEvent)
    content = handoff.event.source["content"]
    assert "m.relates_to" not in content
    assert content["m.mentions"] == {"user_ids": ["@agent:localhost"]}


def test_room_level_mention_batch_preserves_plain_reply_relation() -> None:
    """Mention metadata must preserve plain reply shape without reintroducing thread targeting."""
    room = _make_room()
    typed_reply = _reply_event(
        event_id="$typed",
        reply_to_event_id="$old-reply",
        body="@agent follow-up",
        server_timestamp=1001,
    )
    typed_reply.source["content"]["m.mentions"] = {"user_ids": ["@agent:localhost"]}
    batch = build_coalesced_batch(
        CoalescingKey(room.room_id, None, "@user:localhost"),
        [PendingEvent(event=typed_reply, room=room, source_kind=MESSAGE_SOURCE_KIND)],
    )

    handoff = build_dispatch_handoff(batch)

    assert isinstance(handoff.event, PreparedTextEvent)
    content = handoff.event.source["content"]
    assert content["m.relates_to"] == {"m.in_reply_to": {"event_id": "$old-reply"}}
    assert content["m.mentions"] == {"user_ids": ["@agent:localhost"]}
    assert not EventInfo.from_event(handoff.event.source).can_be_thread_root


@pytest.mark.asyncio
async def test_coalesced_room_plain_reply_target_uses_prompt_thread_not_reply_thread(tmp_path: Path) -> None:
    """Room-level handoff targets the prompt event, not a stale preserved plain reply."""
    bot = _make_bot(tmp_path, debounce_ms=0)
    room = _make_room()
    typed_reply = _reply_event(
        event_id="$typed",
        reply_to_event_id="$old-reply",
        body="room-level follow-up",
        server_timestamp=1001,
    )
    bot._conversation_cache.get_thread_id_for_event = AsyncMock(
        side_effect=lambda _room_id, event_id: "$thread-root" if event_id == "$old-reply" else None,
    )
    batch = build_coalesced_batch(
        CoalescingKey(room.room_id, None, "@user:localhost"),
        [PendingEvent(event=typed_reply, room=room, source_kind=MESSAGE_SOURCE_KIND)],
    )
    dispatches: list[PreparedDispatch] = []

    async def record_response(*args: object, **_kwargs: object) -> None:
        dispatches.append(cast("PreparedDispatch", args[2]))

    with (
        patch.object(bot._turn_policy, "plan_turn", new=AsyncMock(return_value=_respond_dispatch_plan())),
        patch.object(bot._turn_controller, "_execute_response_action", new=AsyncMock(side_effect=record_response)),
    ):
        await bot._turn_controller.handle_coalesced_batch(batch)

    assert len(dispatches) == 1
    assert dispatches[0].target.resolved_thread_id == "$typed"
    assert dispatches[0].context.thread_id is None


def test_single_mentioned_followup_batch_uses_coalescing_thread_relation() -> None:
    """Mentions must not preserve an explicit stale thread relation."""
    room = _make_room()
    typed = _text_event(
        event_id="$typed",
        body="@agent follow-up",
        server_timestamp=1001,
        thread_id="$old-thread",
    )
    typed.source["content"]["m.mentions"] = {"user_ids": ["@agent:localhost"]}
    batch = build_coalesced_batch(
        CoalescingKey(room.room_id, "$new-thread", "@user:localhost"),
        [PendingEvent(event=typed, room=room, source_kind=MESSAGE_SOURCE_KIND)],
    )

    handoff = build_dispatch_handoff(batch)

    assert isinstance(handoff.event, PreparedTextEvent)
    content = handoff.event.source["content"]
    assert content["m.relates_to"] == {"rel_type": "m.thread", "event_id": "$new-thread"}
    assert content["m.mentions"] == {"user_ids": ["@agent:localhost"]}


def test_single_followup_batch_uses_coalescing_thread_relation() -> None:
    """A single follow-up batch dispatches on its coalescing key."""
    room = _make_room()
    post_key = CoalescingKey(room.room_id, "$post-stt-thread", "@user:localhost")
    typed = _text_event(
        event_id="$typed",
        body="typed follow-up",
        server_timestamp=1001,
        thread_id="$voice",
    )

    batch = build_coalesced_batch(
        post_key,
        [PendingEvent(event=typed, room=room, source_kind="message")],
    )

    handoff = build_dispatch_handoff(batch)

    assert isinstance(handoff.event, PreparedTextEvent)
    assert handoff.event.source["content"]["m.relates_to"] == {
        "rel_type": "m.thread",
        "event_id": "$post-stt-thread",
    }


@pytest.mark.asyncio
async def test_register_batch_media_attachments_emits_payload_timing_for_empty_batch(tmp_path: Path) -> None:
    """Batch media registration timing should report empty payload batches."""
    bot = _make_bot(tmp_path)

    with patch("mindroom.inbound_turn_normalizer.emit_elapsed_timing") as mock_emit:
        result = await bot._inbound_turn_normalizer.register_batch_media_attachments(
            BatchMediaAttachmentRequest(
                room_id="!room:localhost",
                thread_id="$thread",
                media_events=[],
            ),
        )

    assert result.attachment_ids == []
    assert result.fallback_images is None
    mock_emit.assert_called_once()
    assert mock_emit.call_args.args[0] == "response_payload.register_batch_media_attachments"
    assert isinstance(mock_emit.call_args.args[1], float)
    assert mock_emit.call_args.kwargs == {
        "room_id": "!room:localhost",
        "thread_id": "$thread",
        "outcome": "success",
        "media_event_count": 0,
        "image_event_count": 0,
        "file_or_video_event_count": 0,
        "attachment_count": 0,
        "fallback_image_count": 0,
    }


@pytest.mark.asyncio
async def test_register_batch_media_attachments_emits_payload_timing_on_failure(tmp_path: Path) -> None:
    """Batch media registration timing should still emit when attachment hydration fails."""
    bot = _make_bot(tmp_path)

    with (
        patch("mindroom.inbound_turn_normalizer.download_image", new=AsyncMock(return_value=None)),
        patch("mindroom.inbound_turn_normalizer.emit_elapsed_timing") as mock_emit,
        pytest.raises(RuntimeError, match="Failed to download image"),
    ):
        await bot._inbound_turn_normalizer.register_batch_media_attachments(
            BatchMediaAttachmentRequest(
                room_id="!room:localhost",
                thread_id="$thread",
                media_events=[_image_event(event_id="$img1")],
            ),
        )

    mock_emit.assert_called_once()
    assert mock_emit.call_args.args[0] == "response_payload.register_batch_media_attachments"
    assert isinstance(mock_emit.call_args.args[1], float)
    assert mock_emit.call_args.kwargs == {
        "room_id": "!room:localhost",
        "thread_id": "$thread",
        "outcome": "failed",
        "media_event_count": 1,
        "image_event_count": 1,
        "file_or_video_event_count": 0,
        "attachment_count": 0,
        "fallback_image_count": 0,
    }


@pytest.mark.asyncio
async def test_dispatch_payload_registers_unregistered_image_from_thread_history(tmp_path: Path) -> None:
    """Prior thread images without MindRoom metadata should become context attachments."""
    bot = _make_bot(tmp_path)
    image_event = _image_event(event_id="$img-history", thread_id="$thread")
    image_content = image_event.source["content"]
    assert isinstance(image_content, dict)
    history_image = ResolvedVisibleMessage(
        sender=image_event.sender,
        body=image_event.body,
        timestamp=1000,
        event_id=image_event.event_id,
        content=image_content,
        thread_id="$thread",
        latest_event_id=image_event.event_id,
    )
    download_response = MagicMock(spec=nio.DownloadResponse)
    download_response.body = b"\x89PNG\r\n\x1a\npayload"
    bot.client.download = AsyncMock(return_value=download_response)

    with patch("mindroom.inbound_turn_normalizer.resolve_thread_attachment_ids", new=AsyncMock(return_value=[])):
        payload = await bot._inbound_turn_normalizer.build_dispatch_payload_with_attachments(
            DispatchPayloadWithAttachmentsRequest(
                room_id="!room:localhost",
                prompt="@test_agent see above",
                current_attachment_ids=[],
                thread_id="$thread",
                media_thread_id="$thread",
                thread_history=[history_image],
            ),
        )

    attachment_id = _attachment_id_for_event("$img-history")
    assert payload.attachment_ids == [attachment_id]
    # History media is pinned to its thread-history message at execution
    # preparation time, so the current-turn payload carries neither the
    # attachment prompt nor inline media for it.
    assert payload.model_prompt is None
    assert payload.media.images == ()
    assert payload.media.audio == ()
    assert payload.media.files == ()
    assert payload.media.videos == ()
    record = load_attachment(tmp_path, attachment_id)
    assert record is not None
    assert record.source_event_id == "$img-history"
    assert record.thread_id == "$thread"


@pytest.mark.asyncio
async def test_trusted_current_attachment_ids_bypass_final_thread_filter(tmp_path: Path) -> None:
    """Current-turn trusted attachments should survive when final dispatch root differs."""
    bot = _make_bot(tmp_path)
    media_path = tmp_path / "voice.ogg"
    media_path.write_bytes(b"voice-bytes")
    attachment = register_local_attachment(
        tmp_path,
        media_path,
        kind="audio",
        attachment_id="voice-attachment",
        mime_type="audio/ogg",
        room_id="!room:localhost",
        thread_id="$voice-root",
        source_event_id="$voice-root",
        sender="@user:localhost",
    )
    assert attachment is not None

    payload = await bot._inbound_turn_normalizer.build_dispatch_payload_with_attachments(
        DispatchPayloadWithAttachmentsRequest(
            room_id="!room:localhost",
            prompt="voice plus later text",
            current_attachment_ids=["voice-attachment"],
            trusted_current_attachment_ids=["voice-attachment"],
            thread_id="$typed-root",
            media_thread_id="$typed-root",
            thread_history=[],
        ),
    )

    assert payload.attachment_ids == ["voice-attachment"]
    assert len(payload.media.audio) == 1


@pytest.mark.asyncio
async def test_flush_logs_failed_outcome_when_dispatch_batch_raises() -> None:
    """Flush telemetry should not report success when dispatch_batch raises."""
    room = _make_room()

    async def failing_dispatch_batch(_batch: object) -> None:
        msg = "boom"
        raise RuntimeError(msg)

    gate = CoalescingGate(
        dispatch_batch=failing_dispatch_batch,
        debounce_seconds=lambda: 0.0,
        is_shutting_down=lambda: False,
    )

    with patch("mindroom.coalescing.emit_elapsed_timing") as mock_emit:
        await _admit_ready(
            gate,
            CoalescingKey("!room:localhost", None, "@user:localhost"),
            PendingEvent(
                event=_text_event(event_id="$m1", body="first"),
                room=room,
                source_kind="message",
            ),
        )
        await gate.drain_all()
        assert _coalescing_gate_is_idle(gate)

    flush_calls = [call for call in mock_emit.call_args_list if call.args and call.args[0] == "coalescing_gate.flush"]
    assert len(flush_calls) == 1
    assert flush_calls[0].kwargs["outcome"] == "failed"


@pytest.mark.asyncio
async def test_coalescing_enqueue_logs_pending_count() -> None:
    """Coalescing enqueue diagnostics should identify the pending scope."""
    room = _make_room()
    gate = CoalescingGate(
        dispatch_batch=AsyncMock(),
        debounce_seconds=lambda: 10.0,
        is_shutting_down=lambda: False,
    )

    with patch("mindroom.coalescing.logger.info") as mock_info:
        await _admit_ready(
            gate,
            CoalescingKey("!room:localhost", "$thread", "@user:localhost"),
            PendingEvent(
                event=_text_event(event_id="$m1", body="first"),
                room=room,
                source_kind="message",
            ),
        )

    await gate.drain_all()
    enqueue_call = next(call for call in mock_info.call_args_list if call.args == ("coalescing_gate_message_enqueued",))
    assert enqueue_call.kwargs["pending_count"] == 1
    assert enqueue_call.kwargs["event_id"] == "$m1"
    assert enqueue_call.kwargs["room_id"] == "!room:localhost"
    assert enqueue_call.kwargs["thread_id"] == "$thread"


@pytest.mark.asyncio
async def test_slow_coalescing_flush_warns_with_correlation_metadata() -> None:
    """Slow flush diagnostics should carry event, room, and thread identifiers."""
    room = _make_room()
    gate = CoalescingGate(
        dispatch_batch=AsyncMock(),
        debounce_seconds=lambda: 0.0,
        is_shutting_down=lambda: False,
    )
    with (
        patch("mindroom.coalescing._COALESCING_FLUSH_WARNING_SECONDS", 0.0),
        patch("mindroom.coalescing.logger.warning") as mock_warning,
    ):
        await _admit_ready(
            gate,
            CoalescingKey("!room:localhost", "$thread", "@user:localhost"),
            PendingEvent(
                event=_text_event(event_id="$m1", body="first"),
                room=room,
                source_kind="message",
            ),
        )
        await _wait_for(lambda: _coalescing_gate_is_idle(gate))

    mock_warning.assert_called_once()
    assert mock_warning.call_args.args == ("coalescing_gate_flush_slow",)
    assert mock_warning.call_args.kwargs["duration_ms"] >= 0.0
    assert mock_warning.call_args.kwargs["source_event_ids"] == ["$m1"]
    assert mock_warning.call_args.kwargs["room_id"] == "!room:localhost"
    assert mock_warning.call_args.kwargs["thread_id"] == "$thread"
    assert mock_warning.call_args.kwargs["outcome"] == "dispatched"


@pytest.mark.asyncio
async def test_timer_flush_logs_dispatch_failure_without_unhandled_task() -> None:
    """Timer callbacks should consume dispatch failures instead of leaking task exceptions."""
    room = _make_room()
    loop = asyncio.get_running_loop()
    loop_exceptions: list[dict[str, object]] = []
    previous_exception_handler = loop.get_exception_handler()

    async def failing_dispatch_batch(_batch: object) -> None:
        msg = "boom"
        raise RuntimeError(msg)

    gate = CoalescingGate(
        dispatch_batch=failing_dispatch_batch,
        debounce_seconds=lambda: 0.01,
        is_shutting_down=lambda: False,
    )

    def capture_loop_exception(_loop: asyncio.AbstractEventLoop, context: dict[str, object]) -> None:
        loop_exceptions.append(context)

    loop.set_exception_handler(capture_loop_exception)
    try:
        with patch("mindroom.coalescing.logger.exception") as mock_exception:
            await _admit_ready(
                gate,
                CoalescingKey("!room:localhost", None, "@user:localhost"),
                PendingEvent(
                    event=_text_event(event_id="$m1", body="first"),
                    room=room,
                    source_kind="message",
                ),
            )
            await _wait_for(lambda: mock_exception.called)
    finally:
        loop.set_exception_handler(previous_exception_handler)

    assert loop_exceptions == []
    mock_exception.assert_called_once()
    assert mock_exception.call_args.args == ("Coalescing drain failed",)
    assert mock_exception.call_args.kwargs["exception_type"] == "RuntimeError"
    assert mock_exception.call_args.kwargs["error_message"] == "Coalesced dispatch failed."
    assert "pending_count" in mock_exception.call_args.kwargs
    assert "oldest_pending_age_ms" in mock_exception.call_args.kwargs
    assert _coalescing_gate_is_idle(gate)


@pytest.mark.asyncio
async def test_dispatch_failure_handoff_runs_after_gate_releases_exact_sources() -> None:
    """Failed deferred sources must reach durable retry ownership after gate cleanup."""
    room = _make_room()
    failure_handoffs: list[tuple[PendingEvent, ...]] = []
    source_owned_during_handoff: list[bool] = []
    handoff_complete = asyncio.Event()

    async def failing_dispatch_batch(_batch: object) -> None:
        msg = "boom"
        raise RuntimeError(msg)

    def on_dispatch_failure(pending_events: tuple[PendingEvent, ...]) -> None:
        failure_handoffs.append(pending_events)
        source_owned_during_handoff.extend(
            gate.has_pending_source_event(pending_event.event.event_id) for pending_event in pending_events
        )
        handoff_complete.set()

    gate = CoalescingGate(
        dispatch_batch=failing_dispatch_batch,
        debounce_seconds=lambda: 0.01,
        is_shutting_down=lambda: False,
        on_dispatch_failure=on_dispatch_failure,
    )

    await _admit_ready(
        gate,
        CoalescingKey("!room:localhost", None, "@user:localhost"),
        PendingEvent(
            event=_text_event(event_id="$m1", body="first"),
            room=room,
            source_kind="message",
        ),
    )
    await asyncio.wait_for(handoff_complete.wait(), timeout=1)

    assert [[pending.event.event_id for pending in handoff] for handoff in failure_handoffs] == [["$m1"]]
    assert source_owned_during_handoff == [False]


@pytest.mark.asyncio
async def test_failed_drain_does_not_poison_future_ingress() -> None:
    """A failed drain should log, clean up, and allow later events to dispatch."""
    room = _make_room()
    dispatched_source_event_ids: list[list[str]] = []

    async def dispatch_batch(batch: CoalescedBatch) -> None:
        source_event_ids = list(batch.source_event_ids)
        if source_event_ids == ["$m1"]:
            msg = "boom"
            raise RuntimeError(msg)
        dispatched_source_event_ids.append(source_event_ids)

    gate = CoalescingGate(
        dispatch_batch=dispatch_batch,
        debounce_seconds=lambda: 0.0,
        is_shutting_down=lambda: False,
    )

    with patch("mindroom.coalescing.logger.exception") as mock_exception:
        await _admit_ready(
            gate,
            CoalescingKey("!room:localhost", None, "@user:localhost"),
            PendingEvent(
                event=_text_event(event_id="$m1", body="first"),
                room=room,
                source_kind="message",
            ),
        )
        await _wait_for(lambda: mock_exception.called)

    assert _coalescing_gate_is_idle(gate)

    await _admit_ready(
        gate,
        CoalescingKey("!room:localhost", None, "@user:localhost"),
        PendingEvent(
            event=_text_event(event_id="$m2", body="second"),
            room=room,
            source_kind="message",
        ),
    )
    await _wait_for(lambda: dispatched_source_event_ids == [["$m2"]])

    assert _coalescing_gate_is_idle(gate)


@pytest.mark.asyncio
async def test_failed_drain_dispatches_buffered_ingress_without_waiting_for_another_event() -> None:
    """Ingress buffered behind a failed dispatch should get its own follow-up drain."""
    room = _make_room()
    entered_first_dispatch = asyncio.Event()
    release_first_dispatch = asyncio.Event()
    dispatched_source_event_ids: list[list[str]] = []

    async def dispatch_batch(batch: CoalescedBatch) -> None:
        source_event_ids = list(batch.source_event_ids)
        if source_event_ids == ["$m1"]:
            entered_first_dispatch.set()
            await release_first_dispatch.wait()
            msg = "boom"
            raise RuntimeError(msg)
        dispatched_source_event_ids.append(source_event_ids)

    gate = CoalescingGate(
        dispatch_batch=dispatch_batch,
        debounce_seconds=lambda: 0.0,
        is_shutting_down=lambda: False,
    )

    with patch("mindroom.coalescing.logger.exception") as mock_exception:
        await _admit_ready(
            gate,
            CoalescingKey("!room:localhost", None, "@user:localhost"),
            PendingEvent(
                event=_text_event(event_id="$m1", body="first"),
                room=room,
                source_kind="message",
            ),
        )
        await entered_first_dispatch.wait()
        await _admit_ready(
            gate,
            CoalescingKey("!room:localhost", None, "@user:localhost"),
            PendingEvent(
                event=_text_event(event_id="$m2", body="second"),
                room=room,
                source_kind="message",
            ),
        )

        release_first_dispatch.set()
        await _wait_for(lambda: mock_exception.called)
        await _wait_for(lambda: dispatched_source_event_ids == [["$m2"]])

    assert _coalescing_gate_is_idle(gate)


@pytest.mark.asyncio
async def test_cancelled_drain_cleans_state_for_later_message() -> None:
    """A cancelled in-flight dispatch should not prevent a fresh later drain."""
    room = _make_room()
    entered_first_dispatch = asyncio.Event()
    never_release = asyncio.Event()
    dispatched_source_event_ids: list[list[str]] = []

    async def dispatch_batch(batch: CoalescedBatch) -> None:
        source_event_ids = list(batch.source_event_ids)
        dispatched_source_event_ids.append(source_event_ids)
        if source_event_ids == ["$m1"]:
            entered_first_dispatch.set()
            await never_release.wait()

    gate = CoalescingGate(
        dispatch_batch=dispatch_batch,
        debounce_seconds=lambda: 0.0,
        is_shutting_down=lambda: False,
    )

    await _admit_ready(
        gate,
        CoalescingKey("!room:localhost", None, "@user:localhost"),
        PendingEvent(
            event=_text_event(event_id="$m1", body="first"),
            room=room,
            source_kind="message",
        ),
    )
    await entered_first_dispatch.wait()

    drain_task = next(iter(gate._gates.values())).drain_task
    assert drain_task is not None
    drain_task.cancel()
    await asyncio.gather(drain_task, return_exceptions=True)

    assert _coalescing_gate_is_idle(gate)

    await _admit_ready(
        gate,
        CoalescingKey("!room:localhost", None, "@user:localhost"),
        PendingEvent(
            event=_text_event(event_id="$m2", body="second"),
            room=room,
            source_kind="message",
        ),
    )
    await _wait_for(lambda: dispatched_source_event_ids == [["$m1"], ["$m2"]])

    assert _coalescing_gate_is_idle(gate)


@pytest.mark.asyncio
async def test_cancelled_drain_dispatches_buffered_ingress_without_waiting_for_another_event() -> None:
    """Ingress buffered behind a cancelled dispatch should get its own follow-up drain."""
    room = _make_room()
    entered_first_dispatch = asyncio.Event()
    never_release = asyncio.Event()
    dispatched_source_event_ids: list[list[str]] = []

    async def dispatch_batch(batch: CoalescedBatch) -> None:
        source_event_ids = list(batch.source_event_ids)
        dispatched_source_event_ids.append(source_event_ids)
        if source_event_ids == ["$m1"]:
            entered_first_dispatch.set()
            await never_release.wait()

    gate = CoalescingGate(
        dispatch_batch=dispatch_batch,
        debounce_seconds=lambda: 0.0,
        is_shutting_down=lambda: False,
    )

    await _admit_ready(
        gate,
        CoalescingKey("!room:localhost", None, "@user:localhost"),
        PendingEvent(
            event=_text_event(event_id="$m1", body="first"),
            room=room,
            source_kind="message",
        ),
    )
    await entered_first_dispatch.wait()
    await _admit_ready(
        gate,
        CoalescingKey("!room:localhost", None, "@user:localhost"),
        PendingEvent(
            event=_text_event(event_id="$m2", body="second"),
            room=room,
            source_kind="message",
        ),
    )

    drain_task = next(iter(gate._gates.values())).drain_task
    assert drain_task is not None
    drain_task.cancel()
    await asyncio.gather(drain_task, return_exceptions=True)
    await _wait_for(lambda: dispatched_source_event_ids == [["$m1"], ["$m2"]])

    assert _coalescing_gate_is_idle(gate)


@pytest.mark.asyncio
async def test_coalescing_drain_logs_lifecycle_metadata() -> None:
    """Drain diagnostics should include enqueue, start, finish, count, and age fields."""
    room = _make_room()
    gate = CoalescingGate(
        dispatch_batch=AsyncMock(),
        debounce_seconds=lambda: 0.0,
        is_shutting_down=lambda: False,
    )

    with patch("mindroom.coalescing.logger.debug") as mock_debug:
        await _admit_ready(
            gate,
            CoalescingKey("!room:localhost", None, "@user:localhost"),
            PendingEvent(
                event=_text_event(event_id="$m1", body="first"),
                room=room,
                source_kind="message",
            ),
        )
        await _wait_for(lambda: _coalescing_gate_is_idle(gate))

    debug_events = [call.args[0] for call in mock_debug.call_args_list if call.args]
    assert "coalescing_gate_enqueue" in debug_events
    assert "coalescing_drain_start" in debug_events
    assert "coalescing_drain_finish" in debug_events
    for event_name in ("coalescing_gate_enqueue", "coalescing_drain_start", "coalescing_drain_finish"):
        event_call = next(call for call in mock_debug.call_args_list if call.args and call.args[0] == event_name)
        assert "pending_count" in event_call.kwargs
        assert "oldest_pending_age_ms" in event_call.kwargs
    enqueue_call = next(
        call for call in mock_debug.call_args_list if call.args and call.args[0] == "coalescing_gate_enqueue"
    )
    assert "duration_ms" in enqueue_call.kwargs


@pytest.mark.asyncio
async def test_cleanup_drains_pending_debounce_tasks(tmp_path: Path) -> None:
    """Drain pending debounce tasks when a bot is cleaned up."""
    bot = _make_bot(tmp_path, debounce_ms=1000)
    bot.client = AsyncMock()
    bot._emit_agent_lifecycle_event = AsyncMock()
    room = _make_room()
    event = _image_event(event_id="$m1")

    with (
        patch.object(bot._turn_controller, "_dispatch_text_message", new=AsyncMock()) as mock_dispatch,
        patch("mindroom.bot.get_joined_rooms", new=AsyncMock(return_value=[])),
        patch("mindroom.bot.wait_for_background_tasks", new=AsyncMock()),
    ):
        await _enqueue_for_dispatch(
            bot,
            event,
            room,
            source_kind="image",
            requester_user_id="@user:localhost",
        )
        await _wait_for(lambda: not _coalescing_gate_is_idle(bot._coalescing_gate))

        await bot.cleanup()

    mock_dispatch.assert_awaited_once()
    assert _coalescing_gate_is_idle(bot._coalescing_gate)


@pytest.mark.asyncio
async def test_turn_store_marks_all_batch_event_ids(tmp_path: Path) -> None:
    """Mark every source event ID from a coalesced batch as responded."""
    bot = _make_bot(tmp_path, debounce_ms=250)
    room = _make_room()
    first = _text_event(event_id="$m1", body="first", server_timestamp=1000, thread_id="$thread")
    second = _text_event(event_id="$m2", body="second", server_timestamp=1001, thread_id="$thread")
    dispatch = _prepared_dispatch(event_id="$m2")
    send_response = AsyncMock(return_value="$placeholder")
    generate_response = AsyncMock(return_value="$response")
    install_send_response_mock(bot, send_response)
    install_generate_response_mock(bot, generate_response)

    with (
        patch.object(
            bot._turn_controller,
            "_prepare_dispatch",
            new=AsyncMock(return_value=prepared_dispatch_result(dispatch)),
        ),
        patch.object(bot._turn_policy, "plan_turn", new=AsyncMock(return_value=_respond_dispatch_plan())),
        patch.object(
            bot._inbound_turn_normalizer,
            "build_dispatch_payload_with_attachments",
            new=AsyncMock(return_value=DispatchPayload(prompt="combined")),
        ),
        patch.object(ResponsePayloadPreparer, "_log_dispatch_latency"),
    ):
        await _enqueue_for_dispatch(
            bot,
            first,
            room,
            source_kind="message",
            requester_user_id="@user:localhost",
        )
        await _enqueue_for_dispatch(
            bot,
            second,
            room,
            source_kind="message",
            requester_user_id="@user:localhost",
        )
        await _wait_for(lambda: bot._turn_store.is_handled("$m1"), deadline_seconds=1.0)

    assert bot._turn_store.is_handled("$m1")
    assert bot._turn_store.is_handled("$m2")
    turn_record = bot._turn_store.get_turn_record("$m1")
    assert turn_record is not None
    assert turn_record.response_event_id == "$response"
    assert turn_record.source_event_ids == ("$m1", "$m2")
    assert turn_record.anchor_event_id == "$m2"


@pytest.mark.asyncio
async def test_zero_debounce_dispatches_immediately(tmp_path: Path) -> None:
    """A zero debounce window should dispatch each message without delay."""
    bot = _make_bot(tmp_path, debounce_ms=0)
    room = _make_room()
    event = _text_event(event_id="$m1", body="immediate")
    calls: list[tuple[str, list[str]]] = []

    async def record_dispatch(
        _room: nio.MatrixRoom,
        dispatched_event: nio.RoomMessageText,
        _requester_user_id: str,
        *,
        media_events: list[object] | None = None,
        handled_turn: TurnRecord | None = None,
        **_metadata: object,
    ) -> None:
        _ = media_events, handled_turn
        calls.append((dispatched_event.body, _handled_turn_source_event_ids(handled_turn)))

    with patch.object(bot._turn_controller, "_dispatch_text_message", new=AsyncMock(side_effect=record_dispatch)):
        await _enqueue_for_dispatch(
            bot,
            event,
            room,
            source_kind="message",
            requester_user_id="@user:localhost",
        )
        await _wait_for(lambda: len(calls) == 1)

    assert calls == [("immediate", ["$m1"])]
    assert _coalescing_gate_is_idle(bot._coalescing_gate)


@pytest.mark.asyncio
async def test_multiple_commands_each_dispatch_independently(tmp_path: Path) -> None:
    """Each command dispatches as its own solo control input even when sent rapidly."""
    bot = _make_bot(tmp_path, debounce_ms=60_000)
    room = _make_room()
    first_cmd = _text_event(event_id="$c1", body="!help", server_timestamp=1000)
    second_cmd = _text_event(event_id="$c2", body="!schedule list", server_timestamp=1001)
    calls: list[tuple[str, list[str]]] = []

    async def record_dispatch(
        _room: nio.MatrixRoom,
        dispatched_event: nio.RoomMessageText,
        _requester_user_id: str,
        *,
        media_events: list[object] | None = None,
        handled_turn: TurnRecord | None = None,
        **_metadata: object,
    ) -> None:
        _ = media_events, handled_turn
        calls.append((dispatched_event.body, _handled_turn_source_event_ids(handled_turn)))

    with patch.object(bot._turn_controller, "_dispatch_text_message", new=AsyncMock(side_effect=record_dispatch)):
        await bot._turn_controller.handle_text_event(room, first_cmd)
        await bot._turn_controller.handle_text_event(room, second_cmd)

    assert calls == [
        ("!help", ["$c1"]),
        ("!schedule list", ["$c2"]),
    ]


@pytest.mark.asyncio
async def test_gate_entry_removed_after_dispatch_with_no_pending(tmp_path: Path) -> None:
    """A gate entry should be cleaned up once dispatch completes with no new pending."""
    bot = _make_bot(tmp_path)
    room = _make_room()
    event = _text_event(event_id="$m1", body="hello")

    with patch.object(bot._turn_controller, "_dispatch_text_message", new=AsyncMock()):
        assert _coalescing_gate_is_idle(bot._coalescing_gate)
        await _enqueue_for_dispatch(
            bot,
            event,
            room,
            source_kind="message",
            requester_user_id="@user:localhost",
        )
        await _wait_for(lambda: _coalescing_gate_is_idle(bot._coalescing_gate))

    assert _coalescing_gate_is_idle(bot._coalescing_gate)


# ---------------------------------------------------------------------------
# BLOCKER 1 regression: thread-history guard for backlog/replay scenarios
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_backlog_replay_skips_older_message_when_newer_exists(tmp_path: Path) -> None:
    """Skip an older message during backlog replay when a newer unresponded message exists."""
    bot = _make_bot(tmp_path)
    room = _make_room()
    older_event = PreparedTextEvent(
        sender="@user:localhost",
        event_id="$m1",
        body="old",
        source={"content": {"msgtype": "m.text", "body": "old"}},
        server_timestamp=1000,
    )
    dispatch = _prepared_dispatch(event_id="$m1", body="old")

    # Thread history already contains a newer unresponded message from the same sender
    newer_msg = ResolvedVisibleMessage(
        sender="@user:localhost",
        body="newer",
        timestamp=2000,
        event_id="$m2",
        content={"body": "newer"},
        thread_id=None,
        latest_event_id="$m2",
    )
    _set_context_histories(dispatch, [newer_msg])

    action_mock = AsyncMock()
    with (
        patch.object(
            bot._turn_controller,
            "_prepare_dispatch",
            new=AsyncMock(return_value=prepared_dispatch_result(dispatch)),
        ),
        patch.object(bot._turn_policy, "plan_turn", new=action_mock),
    ):
        await bot._turn_controller._dispatch_text_message(room, older_event, "@user:localhost")

    # Older message should be skipped — resolve_dispatch_action never called
    action_mock.assert_not_awaited()
    assert not bot._turn_store.is_handled("$m1")


_SourceOwnership = Literal["single", "mixed", "absent", "partial", "redacted"]


def _coalesced_ownership_record(ownership: _SourceOwnership) -> TurnRecord:
    """Build the persisted coalesced record for one source-ownership shape.

    ``absent`` models a record written before ``source_event_metadata`` existed, which still
    decodes because the ledger schema version never changed. ``partial`` models a map missing one
    source's entry, and ``redacted`` models the map normalization prunes for a redacted source.
    """
    if ownership == "absent":
        record = TurnRecord.create(["$alice", "$bob"])
        assert record.source_event_metadata is None
        return record
    alice_metadata = SourceEventMetadata(
        sender="@alice:localhost" if ownership == "mixed" else "@bob:localhost",
        timestamp_ms=1000,
    )
    bob_metadata = SourceEventMetadata(sender="@bob:localhost", timestamp_ms=2000)
    record = TurnRecord.create(
        ["$alice", "$bob"],
        redacted_source_event_ids=["$alice"] if ownership == "redacted" else [],
        source_event_metadata=(
            {"$bob": bob_metadata} if ownership == "partial" else {"$alice": alice_metadata, "$bob": bob_metadata}
        ),
    )
    assert record.source_event_metadata is not None
    assert set(record.source_event_metadata) == (
        {"$bob"} if ownership in {"partial", "redacted"} else {"$alice", "$bob"}
    )
    return record


@pytest.mark.asyncio
@pytest.mark.parametrize("degraded", [False, True])
@pytest.mark.parametrize("ownership", get_args(_SourceOwnership))
async def test_backlog_replay_respects_coalesced_source_ownership(
    tmp_path: Path,
    *,
    degraded: bool,
    ownership: _SourceOwnership,
) -> None:
    """A newer requester turn supersedes a coalesced batch only when every replayable source is theirs.

    Sources the record cannot attribute — metadata absent entirely, or missing for one source — fail
    closed, because whole-turn suppression would settle another requester's message without a reply.
    """
    bot = _make_bot(tmp_path)
    room = _make_room()
    primary_event = PreparedTextEvent(
        sender="@bob:localhost",
        event_id="$bob",
        body="bob",
        source={"content": {"msgtype": "m.text", "body": "bob"}},
        server_timestamp=2000,
    )
    dispatch = _prepared_dispatch(
        event_id="$bob",
        requester_user_id="@bob:localhost",
        body="bob",
        thread_id="$thread",
    )
    newer_bob_message = ResolvedVisibleMessage(
        sender="@bob:localhost",
        body="newer bob",
        timestamp=3000,
        event_id="$newer-bob",
        content={"body": "newer bob"},
        thread_id="$thread",
        latest_event_id="$newer-bob",
    )
    if degraded:
        degraded_history = ThreadHistoryResult(
            [newer_bob_message],
            is_full_history=False,
            diagnostics={
                THREAD_HISTORY_SOURCE_DIAGNOSTIC: THREAD_HISTORY_SOURCE_DEGRADED,
                THREAD_HISTORY_DEGRADED_DIAGNOSTIC: True,
            },
        )
        dispatch.context.thread_history = degraded_history
        dispatch.context.replay_guard_history = degraded_history
        bot.event_cache.get_recent_room_events.return_value = [
            _text_event(
                event_id="$newer-bob",
                body="newer bob",
                sender="@bob:localhost",
                server_timestamp=3000,
                thread_id="$thread",
            ).source,
        ]
    else:
        _set_context_histories(dispatch, [newer_bob_message])
    handled_turn = _coalesced_ownership_record(ownership)
    superseded = ownership in {"single", "redacted"}

    plan_calls: list[PreparedDispatch] = []

    async def prepare_dispatch(*_args: object, **_kwargs: object) -> object:
        return prepared_dispatch_result(dispatch)

    async def plan_turn(
        _room: object,
        _event: object,
        prepared_dispatch: PreparedDispatch,
        **_kwargs: object,
    ) -> _DispatchPlan:
        plan_calls.append(prepared_dispatch)
        return _DispatchPlan(kind="ignore")

    with (
        capture_logs() as captured_logs,
        patch.object(
            bot._turn_controller,
            "_prepare_dispatch",
            new=prepare_dispatch,
        ),
        patch.object(bot._turn_policy, "plan_turn", new=plan_turn),
    ):
        await bot._turn_controller._dispatch_text_message(
            room,
            primary_event,
            "@bob:localhost",
            handled_turn=handled_turn,
        )

    assert plan_calls == ([] if superseded else [dispatch])
    assert not any(
        log.get("event") == "Thread replay guard degraded; proceeding without negative newer-message proof"
        for log in captured_logs
    )
    if degraded and superseded:
        bot.event_cache.get_recent_room_events.assert_awaited_once_with(
            room.room_id,
            event_type="m.room.message",
            since_ts_ms=2000,
        )
    else:
        bot.event_cache.get_recent_room_events.assert_not_awaited()
    assert not bot._turn_store.is_handled("$alice")
    assert not bot._turn_store.is_handled("$bob")


@pytest.mark.asyncio
@pytest.mark.parametrize("include_physical_metadata", [False, True])
async def test_backlog_replay_fails_closed_when_physical_source_collides_with_alias(
    tmp_path: Path,
    *,
    include_physical_metadata: bool,
) -> None:
    """A relay alias cannot hide a differently owned physical source during whole-turn suppression."""
    bot = _make_bot(tmp_path)
    room = _make_room()
    relay_event_id = "$relay"
    human_event_id = "$human"
    primary_event = PreparedTextEvent(
        sender="@bob:localhost",
        event_id=relay_event_id,
        body="relay",
        source={"content": {"msgtype": "m.text", "body": "relay"}},
        server_timestamp=1000,
    )
    dispatch = _prepared_dispatch(
        event_id=relay_event_id,
        requester_user_id="@bob:localhost",
        body="relay",
        thread_id="$thread",
    )
    _set_context_histories(
        dispatch,
        [
            ResolvedVisibleMessage(
                sender="@bob:localhost",
                body="newer bob",
                timestamp=3000,
                event_id="$newer-bob",
                content={"body": "newer bob"},
                thread_id="$thread",
                latest_event_id="$newer-bob",
            ),
        ],
    )
    source_event_metadata = {
        relay_event_id: SourceEventMetadata(
            sender="@bob:localhost",
            discovery_event_id=human_event_id,
        ),
    }
    if include_physical_metadata:
        source_event_metadata[human_event_id] = SourceEventMetadata(sender="@alice:localhost")
    handled_turn = TurnRecord.create(
        [relay_event_id, human_event_id],
        source_event_metadata=source_event_metadata,
        requester_id="@bob:localhost",
    )
    plan_turn = AsyncMock(return_value=_DispatchPlan(kind="ignore"))

    with (
        patch.object(
            bot._turn_controller,
            "_prepare_dispatch",
            new=AsyncMock(return_value=prepared_dispatch_result(dispatch)),
        ),
        patch.object(bot._turn_policy, "plan_turn", new=plan_turn),
    ):
        await bot._turn_controller._dispatch_text_message(
            room,
            primary_event,
            "@bob:localhost",
            handled_turn=handled_turn,
        )

    plan_turn.assert_awaited_once()
    assert not bot._turn_store.is_handled(relay_event_id)
    assert not bot._turn_store.is_handled(human_event_id)


@pytest.mark.asyncio
async def test_backlog_replay_fails_closed_after_legacy_coalesced_projection(tmp_path: Path) -> None:
    """Projecting a legacy coalesced redaction marker must not invent singleton ownership."""
    bot = _make_bot(tmp_path)
    room = _make_room()
    conflicting_event_id = "$already-owned"
    retained_event_id = "$retained"
    bot._turn_store.record_turn(
        TurnRecord.create([conflicting_event_id], response_event_id="$existing-response"),
    )
    projected = bot._turn_store.record_pending_turn(
        TurnRecord.create(
            [conflicting_event_id, retained_event_id],
            redacted_source_event_ids=[conflicting_event_id],
            requester_id="@bob:localhost",
            completed=False,
        ),
    )
    assert projected is not None
    assert projected.source_event_ids == (retained_event_id,)
    assert projected.source_event_metadata == {}

    primary_event = PreparedTextEvent(
        sender="@bob:localhost",
        event_id=retained_event_id,
        body="retained",
        source={"content": {"msgtype": "m.text", "body": "retained"}},
        server_timestamp=1000,
    )
    dispatch = _prepared_dispatch(
        event_id=retained_event_id,
        requester_user_id="@bob:localhost",
        body="retained",
        thread_id="$thread",
    )
    _set_context_histories(
        dispatch,
        [
            ResolvedVisibleMessage(
                sender="@bob:localhost",
                body="newer bob",
                timestamp=3000,
                event_id="$newer-bob",
                content={"body": "newer bob"},
                thread_id="$thread",
                latest_event_id="$newer-bob",
            ),
        ],
    )
    plan_turn = AsyncMock(return_value=_DispatchPlan(kind="ignore"))

    with (
        patch.object(
            bot._turn_controller,
            "_prepare_dispatch",
            new=AsyncMock(return_value=prepared_dispatch_result(dispatch)),
        ),
        patch.object(bot._turn_policy, "plan_turn", new=plan_turn),
    ):
        await bot._turn_controller._dispatch_text_message(
            room,
            primary_event,
            "@bob:localhost",
            handled_turn=projected,
        )

    plan_turn.assert_awaited_once()
    assert not bot._turn_store.is_handled(retained_event_id)


@pytest.mark.asyncio
async def test_backlog_replay_degraded_thread_history_uses_cached_room_event_positive_proof(tmp_path: Path) -> None:
    """Degraded empty thread history must not prove that no newer thread message exists."""
    bot = _make_bot(tmp_path)
    room = _make_room()
    older_event = PreparedTextEvent(
        sender="@user:localhost",
        event_id="$m1",
        body="old",
        source={"content": {"msgtype": "m.text", "body": "old"}},
        server_timestamp=1000,
    )
    dispatch = _prepared_dispatch(event_id="$m1", body="old", thread_id="$thread")
    degraded_history = ThreadHistoryResult(
        [],
        is_full_history=False,
        diagnostics={
            THREAD_HISTORY_SOURCE_DIAGNOSTIC: THREAD_HISTORY_SOURCE_DEGRADED,
            THREAD_HISTORY_DEGRADED_DIAGNOSTIC: True,
            THREAD_HISTORY_ERROR_DIAGNOSTIC: "cache_coordinator_timeout",
        },
    )
    dispatch.context.am_i_mentioned = False
    dispatch.context.thread_history = degraded_history
    dispatch.context.replay_guard_history = degraded_history
    dispatch.context.requires_model_history_refresh = True
    newer_event_source = {
        "event_id": "$m2",
        "sender": "@user:localhost",
        "origin_server_ts": 2000,
        "room_id": room.room_id,
        "type": "m.room.message",
        "content": {
            "msgtype": "m.text",
            "body": "newer",
            "m.relates_to": {"rel_type": "m.thread", "event_id": "$thread"},
        },
    }
    bot.event_cache.get_recent_room_events.return_value = [newer_event_source]

    action_mock = AsyncMock()
    history_guard = MagicMock(wraps=bot._turn_controller._has_newer_unresponded_in_thread)
    with (
        patch.object(
            bot._turn_controller,
            "_prepare_dispatch",
            new=AsyncMock(return_value=prepared_dispatch_result(dispatch)),
        ),
        patch.object(bot._turn_controller, "_has_newer_unresponded_in_thread", new=history_guard),
        patch.object(bot._turn_policy, "plan_turn", new=action_mock),
    ):
        await bot._turn_controller._dispatch_text_message(room, older_event, "@user:localhost")

    history_guard.assert_not_called()
    bot.event_cache.get_recent_room_events.assert_awaited_once_with(
        room.room_id,
        event_type="m.room.message",
        since_ts_ms=1000,
    )
    action_mock.assert_not_awaited()
    assert not bot._turn_store.is_handled("$m1")


@pytest.mark.asyncio
async def test_backlog_replay_degraded_thread_history_ignores_equal_timestamp_cached_event(tmp_path: Path) -> None:
    """Cached replay proof must be strictly newer, matching the full-history guard."""
    bot = _make_bot(tmp_path)
    room = _make_room()
    older_event = PreparedTextEvent(
        sender="@user:localhost",
        event_id="$m1",
        body="old",
        source={"content": {"msgtype": "m.text", "body": "old"}},
        server_timestamp=1000,
    )
    dispatch = _prepared_dispatch(event_id="$m1", body="old", thread_id="$thread")
    degraded_history = ThreadHistoryResult(
        [],
        is_full_history=False,
        diagnostics={
            THREAD_HISTORY_SOURCE_DIAGNOSTIC: THREAD_HISTORY_SOURCE_DEGRADED,
            THREAD_HISTORY_DEGRADED_DIAGNOSTIC: True,
            THREAD_HISTORY_ERROR_DIAGNOSTIC: "cache_coordinator_timeout",
        },
    )
    dispatch.context.am_i_mentioned = False
    dispatch.context.thread_history = degraded_history
    dispatch.context.replay_guard_history = degraded_history
    dispatch.context.requires_model_history_refresh = True
    same_timestamp_event_source = {
        "event_id": "$m2",
        "sender": "@user:localhost",
        "origin_server_ts": 1000,
        "room_id": room.room_id,
        "type": "m.room.message",
        "content": {
            "msgtype": "m.text",
            "body": "same millisecond",
            "m.relates_to": {"rel_type": "m.thread", "event_id": "$thread"},
        },
    }
    bot.event_cache.get_recent_room_events.return_value = [same_timestamp_event_source]

    action_mock = AsyncMock(return_value=_DispatchPlan(kind="ignore"))
    history_guard = MagicMock(wraps=bot._turn_controller._has_newer_unresponded_in_thread)
    with (
        patch.object(
            bot._turn_controller,
            "_prepare_dispatch",
            new=AsyncMock(return_value=prepared_dispatch_result(dispatch)),
        ),
        patch.object(bot._turn_controller, "_has_newer_unresponded_in_thread", new=history_guard),
        patch.object(bot._turn_policy, "plan_turn", new=action_mock),
    ):
        await bot._turn_controller._dispatch_text_message(room, older_event, "@user:localhost")

    history_guard.assert_not_called()
    bot.event_cache.get_recent_room_events.assert_awaited_once_with(
        room.room_id,
        event_type="m.room.message",
        since_ts_ms=1000,
    )
    action_mock.assert_awaited_once()
    assert not bot._turn_store.is_handled("$m1")


@pytest.mark.asyncio
async def test_backlog_replay_degraded_thread_history_counts_trusted_voice_command_body(tmp_path: Path) -> None:
    """Cached voice transcripts that parse like commands should still count as requester turns."""
    bot = _make_bot(tmp_path)
    room = _make_room()
    older_event = PreparedTextEvent(
        sender="@user:localhost",
        event_id="$m1",
        body="old",
        source={"content": {"msgtype": "m.text", "body": "old"}},
        server_timestamp=1000,
    )
    dispatch = _prepared_dispatch(event_id="$m1", body="old", thread_id="$thread")
    degraded_history = ThreadHistoryResult(
        [],
        is_full_history=False,
        diagnostics={
            THREAD_HISTORY_SOURCE_DIAGNOSTIC: THREAD_HISTORY_SOURCE_DEGRADED,
            THREAD_HISTORY_DEGRADED_DIAGNOSTIC: True,
            THREAD_HISTORY_ERROR_DIAGNOSTIC: "cache_coordinator_timeout",
        },
    )
    dispatch.context.am_i_mentioned = False
    dispatch.context.thread_history = degraded_history
    dispatch.context.replay_guard_history = degraded_history
    dispatch.context.requires_model_history_refresh = True
    newer_voice_event_source = {
        "event_id": "$m2",
        "sender": bot.agent_user.user_id,
        "origin_server_ts": 2000,
        "room_id": room.room_id,
        "type": "m.room.message",
        "content": {
            "msgtype": "m.text",
            "body": "!help",
            SOURCE_KIND_KEY: "voice",
            ORIGINAL_SENDER_KEY: "@user:localhost",
            "m.relates_to": {"rel_type": "m.thread", "event_id": "$thread"},
        },
    }
    bot.event_cache.get_recent_room_events.return_value = [newer_voice_event_source]

    action_mock = AsyncMock()
    with (
        patch.object(
            bot._turn_controller,
            "_prepare_dispatch",
            new=AsyncMock(return_value=prepared_dispatch_result(dispatch)),
        ),
        patch.object(bot._turn_policy, "plan_turn", new=action_mock),
    ):
        await bot._turn_controller._dispatch_text_message(room, older_event, "@user:localhost")

    action_mock.assert_not_awaited()
    assert not bot._turn_store.is_handled("$m1")


def test_replay_guard_does_not_supersede_non_interactive_origin_turns() -> None:
    """Replay suppression is gated by turn-origin policy, not just sender/timestamp matching."""
    older_event = PreparedTextEvent(
        sender="@mindroom_general:localhost",
        event_id="$managed",
        body="internal status",
        source={"content": {"msgtype": "m.text", "body": "internal status"}},
        server_timestamp=1000,
    )
    newer_msg = ResolvedVisibleMessage(
        sender="@mindroom_general:localhost",
        body="newer internal status",
        timestamp=2000,
        event_id="$managed-newer",
        content={"body": "newer internal status"},
        thread_id="$thread",
        latest_event_id="$managed-newer",
    )

    should_skip = has_newer_unresponded_in_thread(
        older_event,
        "@mindroom_general:localhost",
        [newer_msg],
        may_be_superseded_by_newer_requester_turn=False,
        requester_user_id_for_event=lambda sender, _source: sender,
        is_visible_router_voice_echo=lambda _sender, _content: False,
        sender_is_trusted_for_ingress_metadata=lambda _sender: True,
        is_handled=lambda _event_id: False,
        logger=MagicMock(),
    )

    assert not should_skip


def test_full_history_replay_guard_ignores_visible_router_voice_echo() -> None:
    """Visible router voice echoes must not suppress the canonical voice turn in full history."""
    older_event = PreparedTextEvent(
        sender="@user:localhost",
        event_id="$voice",
        body="check my calendar",
        source={"content": {"msgtype": "m.text", "body": "check my calendar", SOURCE_KIND_KEY: "voice"}},
        server_timestamp=1000,
    )
    visible_echo = ResolvedVisibleMessage(
        sender="@mindroom_router:localhost",
        body="check my calendar",
        timestamp=2000,
        event_id="$visible_echo",
        content={
            "msgtype": "m.text",
            "body": "check my calendar",
            SOURCE_KIND_KEY: TRUSTED_INTERNAL_RELAY_SOURCE_KIND,
            ORIGINAL_SENDER_KEY: "@user:localhost",
            VISIBLE_ROUTER_VOICE_ECHO_KEY: True,
        },
        thread_id="$thread",
        latest_event_id="$visible_echo",
    )

    should_skip = has_newer_unresponded_in_thread(
        older_event,
        "@user:localhost",
        [visible_echo],
        may_be_superseded_by_newer_requester_turn=True,
        requester_user_id_for_event=lambda sender, source: (
            source["content"].get(ORIGINAL_SENDER_KEY) if sender == "@mindroom_router:localhost" else sender
        ),
        is_visible_router_voice_echo=lambda sender, content: (
            sender == "@mindroom_router:localhost"
            and isinstance(content, dict)
            and content.get(VISIBLE_ROUTER_VOICE_ECHO_KEY) is True
        ),
        sender_is_trusted_for_ingress_metadata=lambda sender: sender == "@mindroom_router:localhost",
        is_handled=lambda _event_id: False,
        logger=MagicMock(),
    )

    assert not should_skip


def test_full_history_replay_guard_counts_non_router_visible_echo_marker() -> None:
    """Only router voice echoes are display-only; other trusted relay turns still suppress older turns."""
    older_event = PreparedTextEvent(
        sender="@user:localhost",
        event_id="$voice",
        body="check my calendar",
        source={"content": {"msgtype": "m.text", "body": "check my calendar", SOURCE_KIND_KEY: "voice"}},
        server_timestamp=1000,
    )
    marked_helper_relay = ResolvedVisibleMessage(
        sender="@mindroom_helper:localhost",
        body="check my calendar",
        timestamp=2000,
        event_id="$helper_relay",
        content={
            "msgtype": "m.text",
            "body": "check my calendar",
            SOURCE_KIND_KEY: TRUSTED_INTERNAL_RELAY_SOURCE_KIND,
            ORIGINAL_SENDER_KEY: "@user:localhost",
            VISIBLE_ROUTER_VOICE_ECHO_KEY: True,
        },
        thread_id="$thread",
        latest_event_id="$helper_relay",
    )

    should_skip = has_newer_unresponded_in_thread(
        older_event,
        "@user:localhost",
        [marked_helper_relay],
        may_be_superseded_by_newer_requester_turn=True,
        requester_user_id_for_event=lambda sender, _source: (
            "@user:localhost" if sender == "@mindroom_helper:localhost" else sender
        ),
        is_visible_router_voice_echo=lambda sender, _content: sender == "@mindroom_router:localhost",
        sender_is_trusted_for_ingress_metadata=lambda sender: (
            sender in {"@mindroom_helper:localhost", "@mindroom_router:localhost"}
        ),
        is_handled=lambda _event_id: False,
        logger=MagicMock(),
    )

    assert should_skip


@pytest.mark.asyncio
async def test_backlog_replay_degraded_thread_history_ignores_visible_router_voice_echo(tmp_path: Path) -> None:
    """Router transcript echoes must not suppress the canonical voice turn they represent."""
    bot = _make_bot(tmp_path)
    room = _make_room()
    older_event = PreparedTextEvent(
        sender="@user:localhost",
        event_id="$voice",
        body="check my calendar",
        source={"content": {"msgtype": "m.text", "body": "check my calendar", SOURCE_KIND_KEY: "voice"}},
        server_timestamp=1000,
    )
    dispatch = _prepared_dispatch(event_id="$voice", body="check my calendar", thread_id="$thread")
    degraded_history = ThreadHistoryResult(
        [],
        is_full_history=False,
        diagnostics={
            THREAD_HISTORY_SOURCE_DIAGNOSTIC: THREAD_HISTORY_SOURCE_DEGRADED,
            THREAD_HISTORY_DEGRADED_DIAGNOSTIC: True,
            THREAD_HISTORY_ERROR_DIAGNOSTIC: "cache_coordinator_timeout",
        },
    )
    dispatch.context.am_i_mentioned = False
    dispatch.context.thread_history = degraded_history
    dispatch.context.replay_guard_history = degraded_history
    dispatch.context.requires_model_history_refresh = True
    visible_echo_source = {
        "event_id": "$visible_echo",
        "sender": "@mindroom_router:localhost",
        "origin_server_ts": 2000,
        "room_id": room.room_id,
        "type": "m.room.message",
        "content": {
            "msgtype": "m.text",
            "body": "check my calendar",
            SOURCE_KIND_KEY: TRUSTED_INTERNAL_RELAY_SOURCE_KIND,
            ORIGINAL_SENDER_KEY: "@user:localhost",
            VISIBLE_ROUTER_VOICE_ECHO_KEY: True,
            "m.relates_to": {"rel_type": "m.thread", "event_id": "$thread"},
        },
    }
    bot.event_cache.get_recent_room_events.return_value = [visible_echo_source]

    action_mock = AsyncMock(return_value=_DispatchPlan(kind="ignore"))
    history_guard = MagicMock(wraps=bot._turn_controller._has_newer_unresponded_in_thread)
    with (
        patch.object(
            bot._turn_controller,
            "_prepare_dispatch",
            new=AsyncMock(return_value=prepared_dispatch_result(dispatch)),
        ),
        patch.object(bot._turn_controller, "_has_newer_unresponded_in_thread", new=history_guard),
        patch.object(bot._turn_policy, "plan_turn", new=action_mock),
    ):
        await bot._turn_controller._dispatch_text_message(room, older_event, "@user:localhost")

    history_guard.assert_not_called()
    bot.event_cache.get_recent_room_events.assert_awaited_once_with(
        room.room_id,
        event_type="m.room.message",
        since_ts_ms=1000,
    )
    action_mock.assert_awaited_once()
    assert not bot._turn_store.is_handled("$voice")


@pytest.mark.asyncio
async def test_backlog_replay_degraded_thread_history_counts_non_router_visible_echo_marker(tmp_path: Path) -> None:
    """Only router transcript echoes are replay noise; other marked relays still count as newer turns."""
    bot = _make_bot(tmp_path)
    room = _make_room()
    older_event = PreparedTextEvent(
        sender="@user:localhost",
        event_id="$voice",
        body="check my calendar",
        source={"content": {"msgtype": "m.text", "body": "check my calendar", SOURCE_KIND_KEY: "voice"}},
        server_timestamp=1000,
    )
    dispatch = _prepared_dispatch(event_id="$voice", body="check my calendar", thread_id="$thread")
    degraded_history = ThreadHistoryResult(
        [],
        is_full_history=False,
        diagnostics={
            THREAD_HISTORY_SOURCE_DIAGNOSTIC: THREAD_HISTORY_SOURCE_DEGRADED,
            THREAD_HISTORY_DEGRADED_DIAGNOSTIC: True,
            THREAD_HISTORY_ERROR_DIAGNOSTIC: "cache_coordinator_timeout",
        },
    )
    dispatch.context.am_i_mentioned = False
    dispatch.context.thread_history = degraded_history
    dispatch.context.replay_guard_history = degraded_history
    dispatch.context.requires_model_history_refresh = True
    marked_helper_relay_source = {
        "event_id": "$helper_relay",
        "sender": bot.agent_user.user_id,
        "origin_server_ts": 2000,
        "room_id": room.room_id,
        "type": "m.room.message",
        "content": {
            "msgtype": "m.text",
            "body": "check my calendar",
            SOURCE_KIND_KEY: TRUSTED_INTERNAL_RELAY_SOURCE_KIND,
            ORIGINAL_SENDER_KEY: "@user:localhost",
            VISIBLE_ROUTER_VOICE_ECHO_KEY: True,
            "m.relates_to": {"rel_type": "m.thread", "event_id": "$thread"},
        },
    }
    bot.event_cache.get_recent_room_events.return_value = [marked_helper_relay_source]

    action_mock = AsyncMock()
    history_guard = MagicMock(wraps=bot._turn_controller._has_newer_unresponded_in_thread)
    with (
        patch.object(
            bot._turn_controller,
            "_prepare_dispatch",
            new=AsyncMock(return_value=prepared_dispatch_result(dispatch)),
        ),
        patch.object(bot._turn_controller, "_has_newer_unresponded_in_thread", new=history_guard),
        patch.object(bot._turn_policy, "plan_turn", new=action_mock),
    ):
        await bot._turn_controller._dispatch_text_message(room, older_event, "@user:localhost")

    history_guard.assert_not_called()
    bot.event_cache.get_recent_room_events.assert_awaited_once_with(
        room.room_id,
        event_type="m.room.message",
        since_ts_ms=1000,
    )
    action_mock.assert_not_awaited()
    assert not bot._turn_store.is_handled("$voice")


@pytest.mark.asyncio
async def test_backlog_replay_degraded_thread_history_uses_cache_indexed_plain_reply(tmp_path: Path) -> None:
    """Degraded replay guard should accept cache-indexed plain replies as same-thread proof."""
    bot = _make_bot(tmp_path)
    room = _make_room()
    older_event = PreparedTextEvent(
        sender="@user:localhost",
        event_id="$m1",
        body="old",
        source={"content": {"msgtype": "m.text", "body": "old"}},
        server_timestamp=1000,
    )
    dispatch = _prepared_dispatch(event_id="$m1", body="old", thread_id="$thread")
    degraded_history = ThreadHistoryResult(
        [],
        is_full_history=False,
        diagnostics={
            THREAD_HISTORY_SOURCE_DIAGNOSTIC: THREAD_HISTORY_SOURCE_DEGRADED,
            THREAD_HISTORY_DEGRADED_DIAGNOSTIC: True,
            THREAD_HISTORY_ERROR_DIAGNOSTIC: "cache_coordinator_timeout",
        },
    )
    dispatch.context.am_i_mentioned = False
    dispatch.context.thread_history = degraded_history
    dispatch.context.replay_guard_history = degraded_history
    dispatch.context.requires_model_history_refresh = True
    newer_event_source = {
        "event_id": "$m2",
        "sender": "@user:localhost",
        "origin_server_ts": 2000,
        "room_id": room.room_id,
        "type": "m.room.message",
        "content": {
            "msgtype": "m.text",
            "body": "newer",
            "m.relates_to": {"m.in_reply_to": {"event_id": "$root"}},
        },
    }
    bot.event_cache.get_recent_room_events.return_value = [newer_event_source]
    bot._conversation_cache.get_thread_id_for_event = AsyncMock(
        side_effect=lambda _room_id, event_id: "$thread" if event_id == "$m2" else None,
    )

    action_mock = AsyncMock()
    history_guard = MagicMock(wraps=bot._turn_controller._has_newer_unresponded_in_thread)
    with (
        patch.object(
            bot._turn_controller,
            "_prepare_dispatch",
            new=AsyncMock(return_value=prepared_dispatch_result(dispatch)),
        ),
        patch.object(bot._turn_controller, "_has_newer_unresponded_in_thread", new=history_guard),
        patch.object(bot._turn_policy, "plan_turn", new=action_mock),
    ):
        await bot._turn_controller._dispatch_text_message(room, older_event, "@user:localhost")

    history_guard.assert_not_called()
    bot._conversation_cache.get_thread_id_for_event.assert_awaited_once_with(room.room_id, "$m2")
    action_mock.assert_not_awaited()
    assert not bot._turn_store.is_handled("$m1")


@pytest.mark.asyncio
async def test_backlog_replay_degraded_thread_history_ignores_edit_events(tmp_path: Path) -> None:
    """Cached edits should not count as newer unresponded requester turns."""
    bot = _make_bot(tmp_path)
    room = _make_room()
    older_event = PreparedTextEvent(
        sender="@user:localhost",
        event_id="$m1",
        body="old",
        source={"content": {"msgtype": "m.text", "body": "old"}},
        server_timestamp=1000,
    )
    dispatch = _prepared_dispatch(event_id="$m1", body="old", thread_id="$thread")
    degraded_history = ThreadHistoryResult(
        [],
        is_full_history=False,
        diagnostics={
            THREAD_HISTORY_SOURCE_DIAGNOSTIC: THREAD_HISTORY_SOURCE_DEGRADED,
            THREAD_HISTORY_DEGRADED_DIAGNOSTIC: True,
            THREAD_HISTORY_ERROR_DIAGNOSTIC: "cache_coordinator_timeout",
        },
    )
    dispatch.context.am_i_mentioned = False
    dispatch.context.thread_history = degraded_history
    dispatch.context.replay_guard_history = degraded_history
    dispatch.context.requires_model_history_refresh = True
    edit_event_source = {
        "event_id": "$edit",
        "sender": "@user:localhost",
        "origin_server_ts": 2000,
        "room_id": room.room_id,
        "type": "m.room.message",
        "content": {
            "msgtype": "m.text",
            "body": "* newer",
            "m.new_content": {
                "msgtype": "m.text",
                "body": "newer",
                "m.relates_to": {"rel_type": "m.thread", "event_id": "$thread"},
            },
            "m.relates_to": {"rel_type": "m.replace", "event_id": "$original"},
        },
    }
    bot.event_cache.get_recent_room_events.return_value = [edit_event_source]

    action_mock = AsyncMock(return_value=_DispatchPlan(kind="ignore"))
    history_guard = MagicMock(wraps=bot._turn_controller._has_newer_unresponded_in_thread)
    with (
        patch.object(
            bot._turn_controller,
            "_prepare_dispatch",
            new=AsyncMock(return_value=prepared_dispatch_result(dispatch)),
        ),
        patch.object(bot._turn_controller, "_has_newer_unresponded_in_thread", new=history_guard),
        patch.object(bot._turn_policy, "plan_turn", new=action_mock),
    ):
        await bot._turn_controller._dispatch_text_message(room, older_event, "@user:localhost")

    history_guard.assert_not_called()
    action_mock.assert_awaited_once()
    assert not bot._turn_store.is_handled("$m1")


@pytest.mark.asyncio
async def test_backlog_replay_degraded_thread_history_fails_open_without_positive_cached_proof(
    tmp_path: Path,
) -> None:
    """Degraded replay guard should proceed unless raw cached events positively prove a newer same-thread turn."""
    bot = _make_bot(tmp_path)
    room = _make_room()
    older_event = PreparedTextEvent(
        sender="@user:localhost",
        event_id="$m1",
        body="old",
        source={"content": {"msgtype": "m.text", "body": "old"}},
        server_timestamp=1000,
    )
    dispatch = _prepared_dispatch(event_id="$m1", body="old", thread_id="$thread")
    degraded_history = ThreadHistoryResult(
        [],
        is_full_history=False,
        diagnostics={
            THREAD_HISTORY_SOURCE_DIAGNOSTIC: THREAD_HISTORY_SOURCE_DEGRADED,
            THREAD_HISTORY_DEGRADED_DIAGNOSTIC: True,
            THREAD_HISTORY_ERROR_DIAGNOSTIC: "cache_coordinator_timeout",
        },
    )
    dispatch.context.am_i_mentioned = False
    dispatch.context.thread_history = degraded_history
    dispatch.context.replay_guard_history = degraded_history
    dispatch.context.requires_model_history_refresh = True
    bot.event_cache.get_recent_room_events.return_value = []

    action_mock = AsyncMock(return_value=_DispatchPlan(kind="ignore"))
    history_guard = MagicMock(wraps=bot._turn_controller._has_newer_unresponded_in_thread)
    with (
        capture_logs() as captured_logs,
        patch.object(
            bot._turn_controller,
            "_prepare_dispatch",
            new=AsyncMock(return_value=prepared_dispatch_result(dispatch)),
        ),
        patch.object(bot._turn_controller, "_has_newer_unresponded_in_thread", new=history_guard),
        patch.object(bot._turn_policy, "plan_turn", new=action_mock),
    ):
        await bot._turn_controller._dispatch_text_message(room, older_event, "@user:localhost")

    history_guard.assert_not_called()
    bot.event_cache.get_recent_room_events.assert_awaited_once_with(
        room.room_id,
        event_type="m.room.message",
        since_ts_ms=1000,
    )
    action_mock.assert_awaited_once()
    assert not bot._turn_store.is_handled("$m1")
    assert any(
        log.get("event") == "Thread replay guard degraded; proceeding without negative newer-message proof"
        for log in captured_logs
    )


@pytest.mark.asyncio
async def test_media_dispatch_uses_replay_snapshot_instead_of_mutated_planning_history(tmp_path: Path) -> None:
    """Media-backed turns must use replay snapshot history instead of mutable planning history."""
    bot = _make_bot(tmp_path)
    room = _make_room()
    image_event = _image_event(event_id="$img1", server_timestamp=1000)
    dispatch_event = PreparedTextEvent(
        sender="@user:localhost",
        event_id="$img1",
        body="[Attached image]",
        source={"content": {"msgtype": "m.text", "body": "[Attached image]"}},
        server_timestamp=1000,
    )
    dispatch = _prepared_dispatch(event_id="$img1", body="[Attached image]")
    hydrated_msg = ResolvedVisibleMessage(
        sender="@user:localhost",
        body="hydrated newer message",
        timestamp=2000,
        event_id="$img2",
        content={"body": "hydrated newer message"},
        thread_id=None,
        latest_event_id="$img2",
    )

    action_mock = AsyncMock(return_value=_DispatchPlan(kind="ignore"))
    dispatch.context.thread_history = [hydrated_msg]
    dispatch.context.replay_guard_history = []

    newer_mock = MagicMock(return_value=False)
    with (
        patch.object(
            bot._turn_controller,
            "_prepare_dispatch",
            new=AsyncMock(return_value=prepared_dispatch_result(dispatch)),
        ),
        patch.object(bot._turn_policy, "plan_turn", new=action_mock),
        patch.object(bot._turn_controller, "_has_newer_unresponded_in_thread", new=newer_mock),
        patch.object(ResponsePayloadPreparer, "_log_dispatch_latency"),
    ):
        await bot._turn_controller._dispatch_text_message(
            room,
            dispatch_event,
            "@user:localhost",
            media_events=[image_event],
        )

    newer_mock.assert_called_once()
    assert list(newer_mock.call_args.args[2]) == []
    action_mock.assert_awaited_once()
    assert not bot._turn_store.is_handled("$img1")


@pytest.mark.asyncio
async def test_thread_history_guard_does_not_interfere_with_normal_dispatch(tmp_path: Path) -> None:
    """Normal live dispatch proceeds when no newer unresponded message exists."""
    bot = _make_bot(tmp_path)
    room = _make_room()
    event = PreparedTextEvent(
        sender="@user:localhost",
        event_id="$m1",
        body="hello",
        source={"content": {"msgtype": "m.text", "body": "hello"}},
        server_timestamp=2000,
    )
    dispatch = _prepared_dispatch(event_id="$m1", body="hello")
    dispatch.context.thread_history = []
    send_response = AsyncMock(return_value="$placeholder")
    generate_response = AsyncMock(return_value="$response")
    install_send_response_mock(bot, send_response)
    install_generate_response_mock(bot, generate_response)

    with (
        patch.object(
            bot._turn_controller,
            "_prepare_dispatch",
            new=AsyncMock(return_value=prepared_dispatch_result(dispatch)),
        ),
        patch.object(bot._turn_policy, "plan_turn", new=AsyncMock(return_value=_respond_dispatch_plan())),
        patch.object(
            bot._inbound_turn_normalizer,
            "build_dispatch_payload_with_attachments",
            new=AsyncMock(return_value=DispatchPayload(prompt="hello")),
        ),
        patch.object(ResponsePayloadPreparer, "_log_dispatch_latency"),
    ):
        await bot._turn_controller._dispatch_text_message(room, event, "@user:localhost")

    # Dispatch proceeded to completion
    assert bot._turn_store.is_handled("$m1")


# ---------------------------------------------------------------------------
# BLOCKER 2 regression: multi-event batch preserves merged metadata
# ---------------------------------------------------------------------------


def _mention_text_event(
    *,
    event_id: str,
    body: str,
    mentioned_user_ids: list[str],
    sender: str = "@user:localhost",
    server_timestamp: int = 1000,
    thread_id: str | None = None,
) -> nio.RoomMessageText:
    """Build a text event with m.mentions metadata."""
    content: dict[str, object] = {
        "msgtype": "m.text",
        "body": body,
        "m.mentions": {"user_ids": mentioned_user_ids},
    }
    if thread_id is not None:
        content["m.relates_to"] = {"rel_type": "m.thread", "event_id": thread_id}
    return cast(
        "nio.RoomMessageText",
        nio.RoomMessageText.from_dict(
            {
                "event_id": event_id,
                "sender": sender,
                "origin_server_ts": server_timestamp,
                "room_id": "!room:localhost",
                "type": "m.room.message",
                "content": content,
            },
        ),
    )


def test_batch_dispatch_event_merges_mentions_across_events() -> None:
    """A batch of '@agent first' + 'follow up' must preserve the mention."""
    room = _make_room()
    mention_event = _mention_text_event(
        event_id="$m1",
        body="@agent first",
        mentioned_user_ids=["@mindroom_test_agent:localhost"],
        server_timestamp=1000,
    )
    followup_event = _text_event(event_id="$m2", body="follow up", server_timestamp=1001)

    batch = build_coalesced_batch(
        CoalescingKey("!room:localhost", None, "@user:localhost"),
        [
            PendingEvent(event=mention_event, room=room, source_kind="message"),
            PendingEvent(event=followup_event, room=room, source_kind="message"),
        ],
    )
    dispatch_event = _build_batch_dispatch_event(batch)

    assert isinstance(dispatch_event, PreparedTextEvent)
    content = dispatch_event.source.get("content", {})
    mentions = content.get("m.mentions", {})
    assert "@mindroom_test_agent:localhost" in mentions.get("user_ids", [])


def test_batch_dispatch_event_preserves_voice_fallback_metadata() -> None:
    """A trusted voice + text batch must preserve system-owned fallback metadata."""
    room = _make_room()
    voice_event = PreparedTextEvent(
        sender="@user:localhost",
        event_id="$voice1",
        body="transcribed voice",
        source={
            "content": {
                "body": "transcribed voice",
                SOURCE_KIND_KEY: "voice",
                VOICE_RAW_AUDIO_FALLBACK_KEY: True,
            },
        },
        server_timestamp=1000,
    )
    text_event = _text_event(event_id="$m2", body="and this too", server_timestamp=1001)

    batch = build_coalesced_batch(
        CoalescingKey("!room:localhost", None, "@user:localhost"),
        [
            PendingEvent(
                event=voice_event,
                room=room,
                source_kind="voice",
                trust_internal_payload_metadata=True,
            ),
            PendingEvent(event=text_event, room=room, source_kind="message"),
        ],
    )
    dispatch_event = _build_batch_dispatch_event(batch)

    assert isinstance(dispatch_event, PreparedTextEvent)
    content = dispatch_event.source.get("content", {})
    assert content.get(VOICE_RAW_AUDIO_FALLBACK_KEY) is True


def test_single_prepared_batch_dispatch_event_preserves_source_kind() -> None:
    """Single prepared events should carry active policy separately from source kind."""
    room = _make_room()
    event = PreparedTextEvent(
        sender="@user:localhost",
        event_id="$followup",
        body="stop if you see this",
        source={"content": {"body": "stop if you see this"}},
        server_timestamp=1000,
    )

    batch = build_coalesced_batch(
        CoalescingKey("!room:localhost", "$thread", "@user:localhost"),
        [
            PendingEvent(
                event=event,
                room=room,
                source_kind="message",
                dispatch_policy_source_kind=ACTIVE_THREAD_FOLLOW_UP_SOURCE_KIND,
            ),
        ],
    )
    handoff = build_dispatch_handoff(batch)
    dispatch_event = _build_batch_dispatch_event(batch)

    assert handoff.ingress.source_kind == "message"
    assert handoff.ingress.dispatch_policy_source_kind == ACTIVE_THREAD_FOLLOW_UP_SOURCE_KIND
    assert isinstance(dispatch_event, PreparedTextEvent)
    assert dispatch_event.source_kind_override is None


def test_single_text_batch_dispatch_event_preserves_bypass_source_kind() -> None:
    """Single text active follow-ups should expose policy without changing source kind."""
    room = _make_room()
    event = _text_event(event_id="$relay", body="@agent relay", server_timestamp=1000)

    batch = build_coalesced_batch(
        CoalescingKey("!room:localhost", None, "@user:localhost"),
        [
            PendingEvent(
                event=event,
                room=room,
                source_kind="message",
                dispatch_policy_source_kind=ACTIVE_THREAD_FOLLOW_UP_SOURCE_KIND,
            ),
        ],
    )
    handoff = build_dispatch_handoff(batch)
    dispatch_event = _build_batch_dispatch_event(batch)

    assert handoff.ingress.source_kind == "message"
    assert handoff.ingress.dispatch_policy_source_kind == ACTIVE_THREAD_FOLLOW_UP_SOURCE_KIND
    assert isinstance(dispatch_event, nio.RoomMessageText)


def test_batch_dispatch_event_preserves_original_sender() -> None:
    """A relay batch must preserve original_sender metadata."""
    room = _make_room()
    relay_event = PreparedTextEvent(
        sender="@bridge:localhost",
        event_id="$relay1",
        body="relayed message",
        source={
            "content": {
                "body": "relayed message",
                ORIGINAL_SENDER_KEY: "@real_user:remote",
            },
        },
        server_timestamp=1000,
    )
    followup = _text_event(
        event_id="$m2",
        body="follow up",
        sender="@bridge:localhost",
        server_timestamp=1001,
    )

    batch = build_coalesced_batch(
        CoalescingKey("!room:localhost", None, "@real_user:remote"),
        [
            PendingEvent(
                event=relay_event,
                room=room,
                source_kind="message",
                trust_internal_payload_metadata=True,
            ),
            PendingEvent(event=followup, room=room, source_kind="message"),
        ],
    )
    dispatch_event = _build_batch_dispatch_event(batch)

    assert isinstance(dispatch_event, PreparedTextEvent)
    content = dispatch_event.source.get("content", {})
    assert content.get(ORIGINAL_SENDER_KEY) == "@real_user:remote"


def test_batch_dispatch_event_preserves_attachment_ids() -> None:
    """Attachment IDs from all events must flow through to the synthetic source."""
    room = _make_room()
    event_with_attachment = PreparedTextEvent(
        sender="@user:localhost",
        event_id="$m1",
        body="see attached",
        source={
            "content": {
                "body": "see attached",
                ATTACHMENT_IDS_KEY: ["att-001"],
            },
        },
        server_timestamp=1000,
    )
    event_with_another = PreparedTextEvent(
        sender="@user:localhost",
        event_id="$m2",
        body="another",
        source={
            "content": {
                "body": "another",
                ATTACHMENT_IDS_KEY: ["att-002"],
            },
        },
        server_timestamp=1001,
    )

    batch = build_coalesced_batch(
        CoalescingKey("!room:localhost", None, "@user:localhost"),
        [
            PendingEvent(
                event=event_with_attachment,
                room=room,
                source_kind="message",
                trust_internal_payload_metadata=True,
            ),
            PendingEvent(
                event=event_with_another,
                room=room,
                source_kind="message",
                trust_internal_payload_metadata=True,
            ),
        ],
    )
    dispatch_event = _build_batch_dispatch_event(batch)

    assert isinstance(dispatch_event, PreparedTextEvent)
    content = dispatch_event.source.get("content", {})
    raw_ids = content.get(ATTACHMENT_IDS_KEY, [])
    assert isinstance(raw_ids, list), "attachment IDs must be a list, not a comma-string"
    assert "att-001" in raw_ids
    assert "att-002" in raw_ids


# ---------------------------------------------------------------------------
# Thread-history guard: command exclusion (BLOCKER 1 from R2 review)
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_newer_command_does_not_suppress_older_message(tmp_path: Path) -> None:
    """A newer !command must not suppress an older legitimate message during backlog replay."""
    bot = _make_bot(tmp_path)
    room = _make_room()
    older_event = PreparedTextEvent(
        sender="@user:localhost",
        event_id="$m1",
        body="What is the project structure?",
        source={"content": {"msgtype": "m.text", "body": "What is the project structure?"}},
        server_timestamp=1000,
    )
    dispatch = _prepared_dispatch(event_id="$m1", body="What is the project structure?")

    # Thread history contains a newer !help command from the same sender
    newer_cmd = ResolvedVisibleMessage(
        sender="@user:localhost",
        body="!help",
        timestamp=2000,
        event_id="$m2",
        content={"body": "!help"},
        thread_id=None,
        latest_event_id="$m2",
    )
    _set_context_histories(dispatch, [newer_cmd])
    send_response = AsyncMock(return_value="$placeholder")
    generate_response = AsyncMock(return_value="$response")
    install_send_response_mock(bot, send_response)
    install_generate_response_mock(bot, generate_response)

    action_mock = AsyncMock(return_value=_respond_dispatch_plan())
    with (
        patch.object(
            bot._turn_controller,
            "_prepare_dispatch",
            new=AsyncMock(return_value=prepared_dispatch_result(dispatch)),
        ),
        patch.object(bot._turn_policy, "plan_turn", new=action_mock),
        patch.object(
            bot._inbound_turn_normalizer,
            "build_dispatch_payload_with_attachments",
            new=AsyncMock(return_value=DispatchPayload(prompt="What is the project structure?")),
        ),
        patch.object(ResponsePayloadPreparer, "_log_dispatch_latency"),
    ):
        await bot._turn_controller._dispatch_text_message(room, older_event, "@user:localhost")

    # The older message must NOT be suppressed — dispatch action should be called
    action_mock.assert_awaited_once()


@pytest.mark.asyncio
async def test_newer_command_with_whitespace_does_not_suppress(tmp_path: Path) -> None:
    """A newer command with leading whitespace must not suppress older messages."""
    bot = _make_bot(tmp_path)
    room = _make_room()
    older_event = PreparedTextEvent(
        sender="@user:localhost",
        event_id="$m1",
        body="hello",
        source={"content": {"msgtype": "m.text", "body": "hello"}},
        server_timestamp=1000,
    )
    dispatch = _prepared_dispatch(event_id="$m1", body="hello")

    newer_cmd = ResolvedVisibleMessage(
        sender="@user:localhost",
        body="  !help  ",
        timestamp=2000,
        event_id="$m2",
        content={"body": "  !help  "},
        thread_id=None,
        latest_event_id="$m2",
    )
    _set_context_histories(dispatch, [newer_cmd])
    send_response = AsyncMock(return_value="$placeholder")
    generate_response = AsyncMock(return_value="$response")
    install_send_response_mock(bot, send_response)
    install_generate_response_mock(bot, generate_response)

    action_mock = AsyncMock(return_value=_respond_dispatch_plan())
    with (
        patch.object(
            bot._turn_controller,
            "_prepare_dispatch",
            new=AsyncMock(return_value=prepared_dispatch_result(dispatch)),
        ),
        patch.object(bot._turn_policy, "plan_turn", new=action_mock),
        patch.object(
            bot._inbound_turn_normalizer,
            "build_dispatch_payload_with_attachments",
            new=AsyncMock(return_value=DispatchPayload(prompt="hello")),
        ),
        patch.object(ResponsePayloadPreparer, "_log_dispatch_latency"),
    ):
        await bot._turn_controller._dispatch_text_message(room, older_event, "@user:localhost")

    action_mock.assert_awaited_once()


# ---------------------------------------------------------------------------
# Thread-history guard: synthetic/automation bypass (BLOCKER 2 from R2 review)
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_scheduled_event_not_suppressed(tmp_path: Path) -> None:
    """Synthetic scheduled events must never be suppressed by the thread-history guard."""
    bot = _make_bot(tmp_path)
    room = _make_room()
    scheduled_event = PreparedTextEvent(
        sender="@mindroom_test_agent:localhost",
        event_id="$s1",
        body="scheduled task output",
        source={
            "content": {"msgtype": "m.text", "body": "scheduled task output", SOURCE_KIND_KEY: "scheduled"},
        },
        server_timestamp=1000,
        source_kind_override="scheduled",
    )
    dispatch = _prepared_dispatch(event_id="$s1", body="scheduled task output", source_kind="scheduled")

    # A newer unresponded message from the same sender exists
    newer_msg = ResolvedVisibleMessage(
        sender="@mindroom_test_agent:localhost",
        body="another scheduled output",
        timestamp=2000,
        event_id="$s2",
        content={"body": "another scheduled output"},
        thread_id=None,
        latest_event_id="$s2",
    )
    _set_context_histories(dispatch, [newer_msg])
    send_response = AsyncMock(return_value="$placeholder")
    generate_response = AsyncMock(return_value="$response")
    install_send_response_mock(bot, send_response)
    install_generate_response_mock(bot, generate_response)

    action_mock = AsyncMock(return_value=_respond_dispatch_plan())
    with (
        patch.object(
            bot._turn_controller,
            "_prepare_dispatch",
            new=AsyncMock(return_value=prepared_dispatch_result(dispatch)),
        ),
        patch.object(bot._turn_policy, "plan_turn", new=action_mock),
        patch.object(
            bot._inbound_turn_normalizer,
            "build_dispatch_payload_with_attachments",
            new=AsyncMock(return_value=DispatchPayload(prompt="scheduled task output")),
        ),
        patch.object(ResponsePayloadPreparer, "_log_dispatch_latency"),
    ):
        await bot._turn_controller._dispatch_text_message(room, scheduled_event, "@mindroom_test_agent:localhost")

    action_mock.assert_awaited_once()


@pytest.mark.asyncio
async def test_hook_event_not_suppressed(tmp_path: Path) -> None:
    """Synthetic hook events must never be suppressed by the thread-history guard."""
    bot = _make_bot(tmp_path)
    room = _make_room()
    hook_event = PreparedTextEvent(
        sender="@mindroom_test_agent:localhost",
        event_id="$h1",
        body="hook result",
        source={"content": {"msgtype": "m.text", "body": "hook result", SOURCE_KIND_KEY: "hook"}},
        server_timestamp=1000,
        source_kind_override="hook",
    )
    dispatch = _prepared_dispatch(event_id="$h1", body="hook result", source_kind="hook")

    newer_msg = ResolvedVisibleMessage(
        sender="@mindroom_test_agent:localhost",
        body="newer message",
        timestamp=2000,
        event_id="$h2",
        content={"body": "newer message"},
        thread_id=None,
        latest_event_id="$h2",
    )
    _set_context_histories(dispatch, [newer_msg])
    send_response = AsyncMock(return_value="$placeholder")
    generate_response = AsyncMock(return_value="$response")
    install_send_response_mock(bot, send_response)
    install_generate_response_mock(bot, generate_response)

    action_mock = AsyncMock(return_value=_respond_dispatch_plan())
    with (
        patch.object(
            bot._turn_controller,
            "_prepare_dispatch",
            new=AsyncMock(return_value=prepared_dispatch_result(dispatch)),
        ),
        patch.object(bot._turn_policy, "plan_turn", new=action_mock),
        patch.object(
            bot._inbound_turn_normalizer,
            "build_dispatch_payload_with_attachments",
            new=AsyncMock(return_value=DispatchPayload(prompt="hook result")),
        ),
        patch.object(ResponsePayloadPreparer, "_log_dispatch_latency"),
    ):
        await bot._turn_controller._dispatch_text_message(room, hook_event, "@mindroom_test_agent:localhost")

    action_mock.assert_awaited_once()


@pytest.mark.asyncio
async def test_multiple_scheduled_fires_not_suppressed(tmp_path: Path) -> None:
    """Two scheduled fires from the same sender must both execute independently."""
    bot = _make_bot(tmp_path)
    room = _make_room()

    first_fire = PreparedTextEvent(
        sender="@mindroom_test_agent:localhost",
        event_id="$s1",
        body="scheduled fire 1",
        source={"content": {"msgtype": "m.text", "body": "scheduled fire 1", SOURCE_KIND_KEY: "scheduled"}},
        server_timestamp=1000,
        source_kind_override="scheduled",
    )
    dispatch = _prepared_dispatch(event_id="$s1", body="scheduled fire 1", source_kind="scheduled")

    # Second scheduled fire is newer and unresponded
    second_fire_msg = ResolvedVisibleMessage(
        sender="@mindroom_test_agent:localhost",
        body="scheduled fire 2",
        timestamp=2000,
        event_id="$s2",
        content={"body": "scheduled fire 2"},
        thread_id=None,
        latest_event_id="$s2",
    )
    _set_context_histories(dispatch, [second_fire_msg])
    send_response = AsyncMock(return_value="$placeholder")
    generate_response = AsyncMock(return_value="$response")
    install_send_response_mock(bot, send_response)
    install_generate_response_mock(bot, generate_response)

    action_mock = AsyncMock(return_value=_respond_dispatch_plan())
    with (
        patch.object(
            bot._turn_controller,
            "_prepare_dispatch",
            new=AsyncMock(return_value=prepared_dispatch_result(dispatch)),
        ),
        patch.object(bot._turn_policy, "plan_turn", new=action_mock),
        patch.object(
            bot._inbound_turn_normalizer,
            "build_dispatch_payload_with_attachments",
            new=AsyncMock(return_value=DispatchPayload(prompt="scheduled fire 1")),
        ),
        patch.object(ResponsePayloadPreparer, "_log_dispatch_latency"),
    ):
        await bot._turn_controller._dispatch_text_message(room, first_fire, "@mindroom_test_agent:localhost")

    # First fire must NOT be suppressed
    action_mock.assert_awaited_once()


# ---------------------------------------------------------------------------
# R3 regression: user-originated synthetics must still be guarded
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_coalesced_user_batch_suppressed_by_thread_guard(tmp_path: Path) -> None:
    """Coalesced user batches (is_synthetic=True, source_kind='user') must be guarded."""
    bot = _make_bot(tmp_path)
    room = _make_room()
    coalesced_event = PreparedTextEvent(
        sender="@user:localhost",
        event_id="$m1",
        body="batched message",
        source={"content": {"msgtype": "m.text", "body": "batched message"}},
        server_timestamp=1000,
        source_kind_override="user",
    )
    dispatch = _prepared_dispatch(event_id="$m1", body="batched message")

    newer_msg = ResolvedVisibleMessage(
        sender="@user:localhost",
        body="newer message",
        timestamp=2000,
        event_id="$m2",
        content={"body": "newer message"},
        thread_id=None,
        latest_event_id="$m2",
    )
    _set_context_histories(dispatch, [newer_msg])

    action_mock = AsyncMock()
    with (
        patch.object(
            bot._turn_controller,
            "_prepare_dispatch",
            new=AsyncMock(return_value=prepared_dispatch_result(dispatch)),
        ),
        patch.object(bot._turn_policy, "plan_turn", new=action_mock),
    ):
        await bot._turn_controller._dispatch_text_message(room, coalesced_event, "@user:localhost")

    # Coalesced user batch MUST be suppressed — not an automation event
    action_mock.assert_not_awaited()
    assert not bot._turn_store.is_handled("$m1")


@pytest.mark.asyncio
async def test_coalesced_media_batch_suppressed_by_replay_snapshot(tmp_path: Path) -> None:
    """Media-backed coalesced user batches must still be suppressed by newer-user replay snapshots."""
    bot = _make_bot(tmp_path)
    room = _make_room()
    image_event = _image_event(event_id="$img1", server_timestamp=1000)
    coalesced_event = PreparedTextEvent(
        sender="@user:localhost",
        event_id="$img1",
        body="[Attached image]",
        source={"content": {"msgtype": "m.text", "body": "[Attached image]"}},
        server_timestamp=1000,
        source_kind_override="image",
    )
    dispatch = _prepared_dispatch(event_id="$img1", body="[Attached image]")

    newer_msg = ResolvedVisibleMessage(
        sender="@user:localhost",
        body="newer message",
        timestamp=2000,
        event_id="$img2",
        content={"body": "newer message"},
        thread_id=None,
        latest_event_id="$img2",
    )
    _set_context_histories(dispatch, [newer_msg])

    action_mock = AsyncMock()
    with (
        patch.object(
            bot._turn_controller,
            "_prepare_dispatch",
            new=AsyncMock(return_value=prepared_dispatch_result(dispatch)),
        ),
        patch.object(bot._turn_policy, "plan_turn", new=action_mock),
    ):
        await bot._turn_controller._dispatch_text_message(
            room,
            coalesced_event,
            "@user:localhost",
            media_events=[image_event],
        )

    # Media-backed coalesced user batch MUST still be suppressed.
    action_mock.assert_not_awaited()
    assert not bot._turn_store.is_handled("$img1")


@pytest.mark.asyncio
async def test_normal_text_command_still_dispatches_as_command(tmp_path: Path) -> None:
    """Non-voice !commands must still take the command execution path."""
    bot = _make_bot(tmp_path, agent_name="router")
    room = _make_room()
    command_event = PreparedTextEvent(
        sender="@user:localhost",
        event_id="$c1",
        body="!schedule tomorrow at 9am turn off the lights",
        source={
            "content": {
                "msgtype": "m.text",
                "body": "!schedule tomorrow at 9am turn off the lights",
            },
        },
        server_timestamp=1000,
    )
    dispatch = _prepared_dispatch(event_id="$c1", body="!schedule tomorrow at 9am turn off the lights")

    handle_cmd_mock = AsyncMock(return_value=True)
    with (
        patch.object(
            bot._turn_controller,
            "_prepare_dispatch",
            new=AsyncMock(return_value=prepared_dispatch_result(dispatch)),
        ),
        patch.object(bot._command_turn_executor, "execute_if_owned", new=handle_cmd_mock),
    ):
        await bot._turn_controller._dispatch_text_message(room, command_event, "@user:localhost")

    handle_cmd_mock.assert_awaited_once()


@pytest.mark.asyncio
async def test_active_voice_follow_up_preserves_voice_command_policy(tmp_path: Path) -> None:
    """Voice active follow-ups should force response policy without becoming commands."""
    bot = _make_bot(tmp_path, agent_name="router")
    room = _make_room()
    voice_command_event = PreparedTextEvent(
        sender="@user:localhost",
        event_id="$voice_command",
        body="!schedule tomorrow at 9am turn off the lights",
        source={
            "content": {
                "msgtype": "m.text",
                "body": "!schedule tomorrow at 9am turn off the lights",
                SOURCE_KIND_KEY: "voice",
            },
        },
        server_timestamp=1000,
        source_kind_override="voice",
    )
    dispatch = _prepared_dispatch(
        event_id="$voice_command",
        body="!schedule tomorrow at 9am turn off the lights",
        thread_id="$thread",
        source_kind="voice",
        dispatch_policy_source_kind=ACTIVE_THREAD_FOLLOW_UP_SOURCE_KIND,
    )

    plan_mock = AsyncMock(return_value=_respond_dispatch_plan())
    execute_command_mock = AsyncMock()
    execute_response_mock = AsyncMock()
    prepare_dispatch_mock = AsyncMock(return_value=prepared_dispatch_result(dispatch))
    with (
        patch.object(
            bot._turn_controller,
            "_prepare_dispatch",
            new=prepare_dispatch_mock,
        ),
        patch.object(bot._turn_policy, "plan_turn", new=plan_mock),
        patch.object(bot._command_turn_executor, "execute_if_owned", new=execute_command_mock),
        patch.object(bot._turn_controller, "_execute_response_action", new=execute_response_mock),
    ):
        await bot._turn_controller._dispatch_text_message(room, voice_command_event, "@user:localhost")

    prepare_dispatch_mock.assert_awaited_once()
    assert prepare_dispatch_mock.await_args.kwargs["use_command_context"] is False
    execute_command_mock.assert_not_awaited()
    plan_mock.assert_awaited_once()
    execute_response_mock.assert_awaited_once()


# ---------------------------------------------------------------------------
# R3 regression: commands must not be suppressed during backlog replay
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_older_command_not_suppressed_during_replay(tmp_path: Path) -> None:
    """An older !help replayed while a newer normal message exists must still dispatch."""
    bot = _make_bot(tmp_path, agent_name="router")
    room = _make_room()
    cmd_event = PreparedTextEvent(
        sender="@user:localhost",
        event_id="$c1",
        body="!help",
        source={"content": {"msgtype": "m.text", "body": "!help"}},
        server_timestamp=1000,
    )
    dispatch = _prepared_dispatch(event_id="$c1", body="!help")

    newer_msg = ResolvedVisibleMessage(
        sender="@user:localhost",
        body="some question",
        timestamp=2000,
        event_id="$c2",
        content={"body": "some question"},
        thread_id=None,
        latest_event_id="$c2",
    )
    _set_context_histories(dispatch, [newer_msg])

    handle_cmd_mock = AsyncMock(return_value=True)
    with (
        patch.object(
            bot._turn_controller,
            "_prepare_dispatch",
            new=AsyncMock(return_value=prepared_dispatch_result(dispatch)),
        ),
        patch.object(bot._command_turn_executor, "execute_if_owned", new=handle_cmd_mock),
    ):
        await bot._turn_controller._dispatch_text_message(room, cmd_event, "@user:localhost")

    # Command must have been dispatched, not suppressed
    handle_cmd_mock.assert_awaited_once()


# ---------------------------------------------------------------------------
# Batch formatted_body merging (BLOCKER 4 from R2 review)
# ---------------------------------------------------------------------------


def _formatted_body_event(
    *,
    event_id: str,
    body: str,
    formatted_body: str,
    mentioned_user_ids: list[str] | None = None,
    sender: str = "@user:localhost",
    server_timestamp: int = 1000,
) -> nio.RoomMessageText:
    """Build a text event with formatted_body (bridge-style pill mentions)."""
    content: dict[str, object] = {
        "msgtype": "m.text",
        "body": body,
        "format": "org.matrix.custom.html",
        "formatted_body": formatted_body,
    }
    if mentioned_user_ids is not None:
        content["m.mentions"] = {"user_ids": mentioned_user_ids}
    return cast(
        "nio.RoomMessageText",
        nio.RoomMessageText.from_dict(
            {
                "event_id": event_id,
                "sender": sender,
                "origin_server_ts": server_timestamp,
                "room_id": "!room:localhost",
                "type": "m.room.message",
                "content": content,
            },
        ),
    )


def test_batch_dispatch_event_preserves_formatted_body_mentions() -> None:
    """Bridge-style pill mentions in formatted_body must survive batch merging."""
    room = _make_room()
    pill_event = _formatted_body_event(
        event_id="$m1",
        body="@agent hello",
        formatted_body='<a href="https://matrix.to/#/@mindroom_test_agent:localhost">agent</a> hello',
        server_timestamp=1000,
    )
    followup = _text_event(event_id="$m2", body="follow up", server_timestamp=1001)

    batch = build_coalesced_batch(
        CoalescingKey("!room:localhost", None, "@user:localhost"),
        [
            PendingEvent(event=pill_event, room=room, source_kind="message"),
            PendingEvent(event=followup, room=room, source_kind="message"),
        ],
    )
    dispatch_event = _build_batch_dispatch_event(batch)

    assert isinstance(dispatch_event, PreparedTextEvent)
    content = dispatch_event.source.get("content", {})
    formatted = content.get("formatted_body", "")
    assert "@mindroom_test_agent:localhost" in formatted
    assert content.get("format") == "org.matrix.custom.html"


def _mentioned_matrix_ids_from_source(source: dict[str, object]) -> list[MatrixID]:
    content = source.get("content", {})
    if not isinstance(content, dict):
        return []
    mentioned_ids: list[MatrixID] = []
    for candidate in (content, content.get("m.new_content")):
        if mentioned_ids or not isinstance(candidate, dict):
            continue
        mentions = candidate.get("m.mentions")
        if not isinstance(mentions, dict):
            continue
        mentioned_ids = [
            MatrixID.parse(user_id) for user_id in mentions.get("user_ids", []) if isinstance(user_id, str)
        ]
    return mentioned_ids


async def _capture_gate_dispatches(
    bot: AgentBot,
    room: nio.MatrixRoom,
    enqueued: Sequence[tuple[nio.Event | PreparedTextEvent, str, str | None, dict[str, object]]],
    *,
    captured_plan_extra_content: list[object] | None = None,
) -> tuple[list[MessageEnvelope], list[list[object]], list[DispatchPayloadWithAttachmentsRequest]]:
    """Dispatch queued events through the real handoff path and capture final envelopes."""
    envelopes: list[MessageEnvelope] = []
    media_batches: list[list[object]] = []
    payload_requests: list[DispatchPayloadWithAttachmentsRequest] = []

    async def record_plan(*args: object, **kwargs: object) -> _DispatchPlan:
        dispatch = cast("PreparedDispatch", args[2])
        envelopes.append(dispatch.envelope)
        media_batches.append(list(cast("list[object] | None", kwargs.get("media_events")) or []))
        if captured_plan_extra_content is not None:
            captured_plan_extra_content.append(kwargs.get("extra_content"))
        return _respond_dispatch_plan()

    async def record_response(*args: object, **_kwargs: object) -> None:
        await prepare_payload_via_seam(bot, args)

    async def record_payload_request(request: DispatchPayloadWithAttachmentsRequest) -> DispatchPayload:
        payload_requests.append(request)
        return DispatchPayload(prompt=request.prompt, attachment_ids=list(request.current_attachment_ids))

    coalescing_thread_id_overrides = {
        event.event_id: metadata["coalescing_thread_id"]
        for event, _source_kind, _dispatch_policy_source_kind, metadata in enqueued
        if "coalescing_thread_id" in metadata
    }
    original_coalescing_thread_id = bot._conversation_resolver.coalescing_thread_id

    async def coalescing_thread_id(room: nio.MatrixRoom, event: nio.Event | PreparedTextEvent) -> str | None:
        if event.event_id in coalescing_thread_id_overrides:
            return cast("str | None", coalescing_thread_id_overrides[event.event_id])
        return await original_coalescing_thread_id(room, event)

    async def extract_dispatch_context(
        _room: nio.MatrixRoom,
        event: nio.Event | PreparedTextEvent,
        **_kwargs: object,
    ) -> object:
        thread_id = EventInfo.from_event(event.source).thread_id
        history: list[ResolvedVisibleMessage] = []
        return dispatch_context_result(
            MessageContext(
                am_i_mentioned=True,
                is_thread=thread_id is not None,
                thread_id=thread_id,
                thread_history=history,
                mentioned_agents=_mentioned_matrix_ids_from_source(event.source),
                has_non_agent_mentions=False,
                replay_guard_history=history,
            ),
        )

    with (
        patch.object(
            bot._conversation_resolver,
            "coalescing_thread_id",
            new=AsyncMock(side_effect=coalescing_thread_id),
        ),
        patch.object(
            bot._conversation_resolver,
            "extract_dispatch_context",
            new=AsyncMock(side_effect=extract_dispatch_context),
        ),
        patch.object(bot._turn_policy, "plan_turn", new=AsyncMock(side_effect=record_plan)),
        patch.object(bot._turn_controller, "_execute_response_action", new=AsyncMock(side_effect=record_response)),
        patch.object(
            bot._inbound_turn_normalizer,
            "build_dispatch_payload_with_attachments",
            new=AsyncMock(side_effect=record_payload_request),
        ),
        patch.object(
            bot._inbound_turn_normalizer,
            "register_batch_media_attachments",
            new=AsyncMock(return_value=_BatchMediaAttachmentResult(attachment_ids=[])),
        ),
    ):
        for event, source_kind, dispatch_policy_source_kind, metadata in enqueued:
            await _enqueue_for_dispatch(
                bot,
                cast("nio.RoomMessageText | PreparedTextEvent", event),
                room,
                source_kind=source_kind,
                dispatch_policy_source_kind=dispatch_policy_source_kind,
                hook_source=cast("str | None", metadata.get("hook_source")),
                message_received_depth=cast("int", metadata.get("message_received_depth", 0)),
                requester_user_id=cast("str", metadata.get("requester_user_id", "@user:localhost")),
                trust_internal_payload_metadata=cast(
                    "bool | None",
                    metadata.get("trust_internal_payload_metadata"),
                ),
            )
        await bot._coalescing_gate.drain_all()

    return envelopes, media_batches, payload_requests


@pytest.mark.asyncio
async def test_gate_final_envelope_preserves_active_voice_source_and_policy(tmp_path: Path) -> None:
    """Active voice follow-ups should keep voice source kind and active policy separate."""
    bot = _make_bot(tmp_path, debounce_ms=0)
    bot.config.agents["test_agent"].thread_mode = "room"
    room = _make_room()
    voice_event = PreparedTextEvent(
        sender="@user:localhost",
        event_id="$voice-active",
        body="!help",
        source={"content": {"msgtype": "m.text", "body": "!help", SOURCE_KIND_KEY: "voice"}},
        server_timestamp=1000,
        source_kind_override="voice",
    )

    envelopes, media_batches, _payload_requests = await _capture_gate_dispatches(
        bot,
        room,
        [
            (
                voice_event,
                "voice",
                ACTIVE_THREAD_FOLLOW_UP_SOURCE_KIND,
                {},
            ),
        ],
    )

    assert [envelope.source_kind for envelope in envelopes] == ["voice"]
    assert envelopes[0].dispatch_policy_source_kind == ACTIVE_THREAD_FOLLOW_UP_SOURCE_KIND
    assert media_batches == [[]]


@pytest.mark.asyncio
async def test_gate_final_envelope_preserves_non_active_voice_command_policy(tmp_path: Path) -> None:
    """Non-active voice transcripts that look like commands should still plan as voice turns."""
    bot = _make_bot(tmp_path, debounce_ms=0)
    bot.config.agents["test_agent"].thread_mode = "room"
    room = _make_room()
    voice_event = PreparedTextEvent(
        sender="@user:localhost",
        event_id="$voice-normal",
        body="!help",
        source={"content": {"msgtype": "m.text", "body": "!help", SOURCE_KIND_KEY: "voice"}},
        server_timestamp=1000,
        source_kind_override="voice",
    )

    envelopes, _media_batches, _payload_requests = await _capture_gate_dispatches(
        bot,
        room,
        [(voice_event, "voice", None, {})],
    )

    assert [envelope.source_kind for envelope in envelopes] == ["voice"]
    assert envelopes[0].dispatch_policy_source_kind is None


@pytest.mark.asyncio
async def test_gate_final_envelope_preserves_active_text_source_and_policy(tmp_path: Path) -> None:
    """Active text follow-ups should stay message source kind with separate active policy."""
    bot = _make_bot(tmp_path, debounce_ms=0)
    room = _make_room()
    event = _text_event(event_id="$text-active", body="I have more context")

    envelopes, _media_batches, _payload_requests = await _capture_gate_dispatches(
        bot,
        room,
        [(event, "message", ACTIVE_THREAD_FOLLOW_UP_SOURCE_KIND, {})],
    )

    assert [envelope.source_kind for envelope in envelopes] == ["message"]
    assert envelopes[0].dispatch_policy_source_kind == ACTIVE_THREAD_FOLLOW_UP_SOURCE_KIND


@pytest.mark.asyncio
async def test_gate_final_envelope_preserves_active_and_normal_media_sources(tmp_path: Path) -> None:
    """Media handoffs should keep modality while active policy stays separate."""
    bot = _make_bot(tmp_path, debounce_ms=0)
    room = _make_room()
    image_event = _image_event(event_id="$image-active", server_timestamp=1000)
    file_event = _file_event(event_id="$file-normal", server_timestamp=1001)

    envelopes, media_batches, _payload_requests = await _capture_gate_dispatches(
        bot,
        room,
        [
            (
                image_event,
                "image",
                ACTIVE_THREAD_FOLLOW_UP_SOURCE_KIND,
                {},
            ),
            (
                file_event,
                "media",
                None,
                {"requester_user_id": "@other:localhost"},
            ),
        ],
    )

    assert [envelope.source_kind for envelope in envelopes] == ["image", "media"]
    assert envelopes[0].dispatch_policy_source_kind == ACTIVE_THREAD_FOLLOW_UP_SOURCE_KIND
    assert envelopes[1].dispatch_policy_source_kind is None
    assert [[event.event_id for event in media_batch] for media_batch in media_batches] == [
        ["$image-active"],
        ["$file-normal"],
    ]


@pytest.mark.asyncio
async def test_gate_final_envelope_preserves_raw_trusted_relay_source_kind(tmp_path: Path) -> None:
    """Raw relay text should stay raw for hydration while handoff source metadata reaches the envelope."""
    bot = _make_bot(tmp_path, debounce_ms=0)
    room = _make_room()
    relay_event = _text_event(
        event_id="$relay",
        body="relayed",
        sender="@mindroom_test_agent:localhost",
        source_kind=None,
        original_sender="@external:example.org",
    )

    envelopes, _media_batches, _payload_requests = await _capture_gate_dispatches(
        bot,
        room,
        [(relay_event, "trusted_internal_relay", None, {"requester_user_id": "@external:example.org"})],
    )

    assert [envelope.source_kind for envelope in envelopes] == ["trusted_internal_relay"]
    assert envelopes[0].requester_id == "@external:example.org"


@pytest.mark.asyncio
async def test_trusted_router_relay_context_uses_handoff_ingress_metadata(tmp_path: Path) -> None:
    """Coalesced relay dispatch should choose router relay context from handoff metadata."""
    bot = _make_bot(tmp_path, debounce_ms=0)
    room = _make_room()
    relay_event = _text_event(
        event_id="$router-relay",
        body="router relay",
        sender="@mindroom_router:localhost",
        original_sender="@external:example.org",
    )
    trusted_context = MessageContext(
        am_i_mentioned=False,
        is_thread=False,
        thread_id=None,
        thread_history=[],
        mentioned_agents=[],
        has_non_agent_mentions=False,
        replay_guard_history=[],
    )

    with (
        patch.object(
            bot._conversation_resolver,
            "extract_trusted_router_relay_context",
            new=AsyncMock(return_value=dispatch_context_result(trusted_context)),
        ) as trusted_context_mock,
        patch.object(
            bot._conversation_resolver,
            "extract_dispatch_context",
            new=AsyncMock(return_value=dispatch_context_result(trusted_context)),
        ) as normal_context_mock,
    ):
        envelopes, _media_batches, _payload_requests = await _capture_gate_dispatches(
            bot,
            room,
            [
                (
                    relay_event,
                    "trusted_internal_relay",
                    None,
                    {"requester_user_id": "@external:example.org"},
                ),
            ],
        )

    trusted_context_mock.assert_awaited_once()
    normal_context_mock.assert_not_awaited()
    assert envelopes[0].source_kind == "trusted_internal_relay"


@pytest.mark.asyncio
async def test_gate_final_envelope_preserves_hook_metadata_with_original_sender(tmp_path: Path) -> None:
    """Hook dispatch should not become a trusted relay just because original sender is present."""
    bot = _make_bot(tmp_path, debounce_ms=0)
    room = _make_room()
    hook_event = _text_event(
        event_id="$hook-dispatch",
        body="@test_agent hook output",
        sender="@mindroom_test_agent:localhost",
        source_kind="hook_dispatch",
        original_sender="@requester:localhost",
    )

    envelopes, _media_batches, _payload_requests = await _capture_gate_dispatches(
        bot,
        room,
        [
            (
                hook_event,
                "hook_dispatch",
                None,
                {
                    "hook_source": "message_received",
                    "message_received_depth": 1,
                    "requester_user_id": "@requester:localhost",
                },
            ),
        ],
    )

    assert [envelope.source_kind for envelope in envelopes] == ["hook_dispatch"]
    assert envelopes[0].hook_source == "message_received"
    assert envelopes[0].message_received_depth == 1


@pytest.mark.asyncio
@pytest.mark.parametrize(
    ("source_kind", "event_id"),
    [
        ("scheduled", "$scheduled"),
        ("hook", "$hook"),
        ("hook_dispatch", "$hook-dispatch"),
        ("trusted_internal_relay", "$relay"),
    ],
)
async def test_automation_and_relay_source_kinds_dispatch_solo_with_human_neighbor(
    tmp_path: Path,
    source_kind: str,
    event_id: str,
) -> None:
    """Automation and relay source kinds should be FIFO barriers with preserved final source kind."""
    bot = _make_bot(tmp_path, debounce_ms=50)
    room = _make_room()
    automated = _text_event(
        event_id=event_id,
        body=f"{source_kind} turn",
        sender="@mindroom_test_agent:localhost",
        source_kind=source_kind,
        original_sender="@requester:localhost",
    )
    human = _text_event(event_id="$human", body="human turn", sender="@requester:localhost", server_timestamp=1001)

    envelopes, _media_batches, _payload_requests = await _capture_gate_dispatches(
        bot,
        room,
        [
            (
                automated,
                source_kind,
                None,
                {
                    "requester_user_id": "@requester:localhost",
                    "hook_source": "message_received" if source_kind.startswith("hook") else None,
                    "message_received_depth": 1 if source_kind.startswith("hook") else 0,
                },
            ),
            (human, "message", None, {"requester_user_id": "@requester:localhost"}),
        ],
    )

    assert [envelope.source_kind for envelope in envelopes] == [source_kind, "message"]
    assert [envelope.source_event_id for envelope in envelopes] == [event_id, "$human"]


@pytest.mark.asyncio
async def test_coalesced_attachment_ids_reach_envelope_and_model_payload(tmp_path: Path) -> None:
    """Coalesced attachment IDs should reach both the final envelope and model payload request."""
    bot = _make_bot(tmp_path, debounce_ms=10)
    room = _make_room()
    first = PreparedTextEvent(
        sender="@user:localhost",
        event_id="$att1",
        body="first attachment",
        source={
            "content": {
                "msgtype": "m.text",
                "body": "first attachment",
                ATTACHMENT_IDS_KEY: ["att-001"],
            },
        },
        server_timestamp=1000,
    )
    second = PreparedTextEvent(
        sender="@user:localhost",
        event_id="$att2",
        body="second attachment",
        source={
            "content": {
                "msgtype": "m.text",
                "body": "second attachment",
                ATTACHMENT_IDS_KEY: ["att-002"],
            },
        },
        server_timestamp=1001,
    )

    envelopes, _media_batches, payload_requests = await _capture_gate_dispatches(
        bot,
        room,
        [
            (
                first,
                "message",
                None,
                {"coalescing_thread_id": "$thread", "trust_internal_payload_metadata": True},
            ),
            (
                second,
                "message",
                None,
                {"coalescing_thread_id": "$thread", "trust_internal_payload_metadata": True},
            ),
        ],
    )

    assert len(envelopes) == 1
    assert envelopes[0].attachment_ids == ("att-001", "att-002")
    assert payload_requests[0].current_attachment_ids == ["att-001", "att-002"]


@pytest.mark.asyncio
async def test_coalesced_non_primary_mention_reaches_final_envelope(tmp_path: Path) -> None:
    """Mention metadata from non-primary coalesced events should reach context extraction."""
    bot = _make_bot(tmp_path, debounce_ms=10)
    room = _make_room()
    first = _text_event(event_id="$plain", body="first")
    second = _mention_text_event(
        event_id="$mention",
        body="@test_agent second",
        mentioned_user_ids=["@mindroom_test_agent:localhost"],
        server_timestamp=1001,
    )

    envelopes, _media_batches, _payload_requests = await _capture_gate_dispatches(
        bot,
        room,
        [
            (first, "message", None, {"coalescing_thread_id": "$thread"}),
            (second, "message", None, {"coalescing_thread_id": "$thread"}),
        ],
    )

    assert len(envelopes) == 1
    assert envelopes[0].mentioned_agents == ("test_agent",)


@pytest.mark.asyncio
async def test_untrusted_raw_payload_metadata_spoofing_does_not_reach_envelope_or_payload(
    tmp_path: Path,
) -> None:
    """User-authored internal payload keys should not become trusted handoff metadata."""
    bot = _make_bot(tmp_path, debounce_ms=0)
    room = _make_room()
    event = _text_event(event_id="$spoof", body="hello", source_kind="hook_dispatch")
    content = event.source["content"]
    assert isinstance(content, dict)
    content[ATTACHMENT_IDS_KEY] = ["spoofed-attachment"]
    content[ORIGINAL_SENDER_KEY] = "@spoofed:localhost"
    content[VOICE_RAW_AUDIO_FALLBACK_KEY] = True
    content[SKIP_MENTIONS_KEY] = True
    content["com.mindroom.hook_source"] = "spoofed:message_received"
    content[HOOK_MESSAGE_RECEIVED_DEPTH_KEY] = 2
    captured_extra_content: list[object] = []

    envelopes, _media_batches, payload_requests = await _capture_gate_dispatches(
        bot,
        room,
        [(event, "message", None, {})],
        captured_plan_extra_content=captured_extra_content,
    )

    assert envelopes[0].source_kind == "message"
    assert envelopes[0].hook_source is None
    assert envelopes[0].message_received_depth == 0
    assert envelopes[0].attachment_ids == ()
    assert payload_requests[0].current_attachment_ids == []
    assert captured_extra_content == [None]


@pytest.mark.asyncio
async def test_untrusted_nested_skip_mentions_does_not_suppress_visible_mentions(
    tmp_path: Path,
) -> None:
    """Nested edit-layer skip metadata should not suppress visible mentions."""
    bot = _make_bot(tmp_path, debounce_ms=0)
    room = _make_room()
    event = _text_event(event_id="$nested-skip", body="* @test_agent edited")
    content = event.source["content"]
    assert isinstance(content, dict)
    content["m.new_content"] = {
        "msgtype": "m.text",
        "body": "@test_agent edited",
        "m.mentions": {"user_ids": ["@mindroom_test_agent:localhost"]},
        SKIP_MENTIONS_KEY: True,
    }
    captured_extra_content: list[object] = []

    envelopes, _media_batches, payload_requests = await _capture_gate_dispatches(
        bot,
        room,
        [(event, "message", None, {})],
        captured_plan_extra_content=captured_extra_content,
    )

    assert envelopes[0].source_kind == "message"
    assert envelopes[0].mentioned_agents == ("test_agent",)
    assert envelopes[0].attachment_ids == ()
    assert payload_requests[0].current_attachment_ids == []
    assert captured_extra_content == [None]


@pytest.mark.asyncio
async def test_untrusted_coalesced_payload_metadata_spoofing_does_not_reach_envelope_or_payload(
    tmp_path: Path,
) -> None:
    """Synthetic coalesced user batches should not trust internal keys from primary content."""
    bot = _make_bot(tmp_path, debounce_ms=10)
    room = _make_room()
    first = _text_event(event_id="$coalesced-first", body="first")
    second = _text_event(event_id="$coalesced-spoof", body="@test_agent second", server_timestamp=1001)
    second_content = second.source["content"]
    assert isinstance(second_content, dict)
    second_content["m.mentions"] = {"user_ids": ["@mindroom_test_agent:localhost"]}
    second_content[ATTACHMENT_IDS_KEY] = ["spoofed-attachment"]
    second_content[ORIGINAL_SENDER_KEY] = "@spoofed:localhost"
    second_content[VOICE_RAW_AUDIO_FALLBACK_KEY] = True
    second_content[SKIP_MENTIONS_KEY] = True
    captured_extra_content: list[object] = []

    envelopes, _media_batches, payload_requests = await _capture_gate_dispatches(
        bot,
        room,
        [
            (first, "message", None, {"coalescing_thread_id": "$thread"}),
            (second, "message", None, {"coalescing_thread_id": "$thread"}),
        ],
        captured_plan_extra_content=captured_extra_content,
    )

    assert len(envelopes) == 1
    assert envelopes[0].mentioned_agents == ("test_agent",)
    assert envelopes[0].requester_id == "@user:localhost"
    assert envelopes[0].attachment_ids == ()
    assert payload_requests[0].current_attachment_ids == []
    assert captured_extra_content == [None]


@pytest.mark.asyncio
async def test_untrusted_synthetic_voice_payload_metadata_spoofing_is_not_trusted(
    tmp_path: Path,
) -> None:
    """Synthetic voice wrappers should not imply trusted internal payload metadata."""
    bot = _make_bot(tmp_path, debounce_ms=0)
    bot.config.agents["test_agent"].thread_mode = "room"
    room = _make_room()
    spoofed_voice = PreparedTextEvent(
        sender="@user:localhost",
        event_id="$voice-spoof",
        body="voice transcript",
        source={
            "content": {
                "msgtype": "m.text",
                "body": "voice transcript",
                ATTACHMENT_IDS_KEY: ["spoofed-attachment"],
                ORIGINAL_SENDER_KEY: "@spoofed:localhost",
                VOICE_RAW_AUDIO_FALLBACK_KEY: True,
                SKIP_MENTIONS_KEY: True,
            },
        },
        server_timestamp=1000,
        source_kind_override="voice",
    )
    captured_extra_content: list[object] = []

    envelopes, _media_batches, payload_requests = await _capture_gate_dispatches(
        bot,
        room,
        [(spoofed_voice, "voice", None, {})],
        captured_plan_extra_content=captured_extra_content,
    )

    assert envelopes[0].source_kind == "voice"
    assert envelopes[0].attachment_ids == ()
    assert payload_requests[0].current_attachment_ids == []
    assert captured_extra_content == [None]


@pytest.mark.asyncio
async def test_trusted_voice_normalized_payload_metadata_reaches_envelope_and_payload(
    tmp_path: Path,
) -> None:
    """Voice normalizer-owned internal metadata should remain trusted when explicitly marked."""
    bot = _make_bot(tmp_path, debounce_ms=0)
    bot.config.agents["test_agent"].thread_mode = "room"
    room = _make_room()
    voice_event = PreparedTextEvent(
        sender="@user:localhost",
        event_id="$voice-trusted",
        body="voice transcript",
        source={
            "content": {
                "msgtype": "m.text",
                "body": "voice transcript",
                ATTACHMENT_IDS_KEY: ["voice-attachment"],
                ORIGINAL_SENDER_KEY: "@user:localhost",
                VOICE_RAW_AUDIO_FALLBACK_KEY: True,
            },
        },
        server_timestamp=1000,
        source_kind_override="voice",
    )
    captured_extra_content: list[object] = []

    envelopes, _media_batches, payload_requests = await _capture_gate_dispatches(
        bot,
        room,
        [(voice_event, "voice", None, {"trust_internal_payload_metadata": True})],
        captured_plan_extra_content=captured_extra_content,
    )

    assert envelopes[0].source_kind == "voice"
    assert envelopes[0].attachment_ids == ("voice-attachment",)
    assert payload_requests[0].current_attachment_ids == ["voice-attachment"]
    assert payload_requests[0].trusted_current_attachment_ids == ["voice-attachment"]
    assert captured_extra_content == [
        {
            ATTACHMENT_IDS_KEY: ["voice-attachment"],
            ORIGINAL_SENDER_KEY: "@user:localhost",
            VOICE_RAW_AUDIO_FALLBACK_KEY: True,
        },
    ]


@pytest.mark.asyncio
async def test_trusted_voice_transcript_metadata_reaches_envelope_and_payload(
    tmp_path: Path,
) -> None:
    """Successful voice transcript metadata should survive handoff as hidden payload guidance."""
    bot = _make_bot(tmp_path, debounce_ms=0)
    bot.config.agents["test_agent"].thread_mode = "room"
    room = _make_room()
    voice_event = PreparedTextEvent(
        sender="@user:localhost",
        event_id="$voice-transcript",
        body="voice transcript",
        source={
            "content": {
                "msgtype": "m.text",
                "body": "voice transcript",
                ATTACHMENT_IDS_KEY: ["voice-attachment"],
                ORIGINAL_SENDER_KEY: "@user:localhost",
                VOICE_TRANSCRIPT_KEY: True,
            },
        },
        server_timestamp=1000,
        source_kind_override="voice",
    )
    captured_extra_content: list[object] = []

    envelopes, _media_batches, payload_requests = await _capture_gate_dispatches(
        bot,
        room,
        [(voice_event, "voice", None, {"trust_internal_payload_metadata": True})],
        captured_plan_extra_content=captured_extra_content,
    )

    assert envelopes[0].source_kind == "voice"
    assert payload_requests[0].current_attachment_ids == ["voice-attachment"]
    assert payload_requests[0].trusted_current_attachment_ids == ["voice-attachment"]
    assert payload_requests[0].voice_transcript is True
    assert captured_extra_content == [
        {
            ATTACHMENT_IDS_KEY: ["voice-attachment"],
            ORIGINAL_SENDER_KEY: "@user:localhost",
            VOICE_TRANSCRIPT_KEY: True,
        },
    ]


@pytest.mark.asyncio
async def test_coalesced_root_voice_attachment_is_trusted_when_later_text_is_primary(tmp_path: Path) -> None:
    """Root voice audio should stay in the current payload when later root text becomes primary."""
    bot = _make_bot(tmp_path, debounce_ms=10)
    room = _make_room()
    voice_event = PreparedTextEvent(
        sender="@user:localhost",
        event_id="$voice-root",
        body="voice transcript",
        source={
            "content": {
                "msgtype": "m.text",
                "body": "voice transcript",
                ATTACHMENT_IDS_KEY: ["voice-attachment"],
                ORIGINAL_SENDER_KEY: "@user:localhost",
                VOICE_RAW_AUDIO_FALLBACK_KEY: True,
            },
        },
        server_timestamp=1000,
        source_kind_override="voice",
    )
    text_event = _text_event(event_id="$typed-root", body="typed follow-up", server_timestamp=1001)

    envelopes, _media_batches, payload_requests = await _capture_gate_dispatches(
        bot,
        room,
        [
            (
                voice_event,
                "voice",
                None,
                {"coalescing_thread_id": None, "trust_internal_payload_metadata": True},
            ),
            (text_event, "message", None, {"coalescing_thread_id": None}),
        ],
    )

    assert len(envelopes) == 1
    assert envelopes[0].attachment_ids == ("voice-attachment",)
    assert payload_requests[0].current_attachment_ids == ["voice-attachment"]
    assert payload_requests[0].trusted_current_attachment_ids == ["voice-attachment"]


@pytest.mark.asyncio
async def test_untrusted_sidecar_payload_metadata_spoofing_does_not_reach_envelope_or_payload(
    tmp_path: Path,
) -> None:
    """User-authored sidecar JSON should hydrate text without trusting internal payload keys."""
    bot = _make_bot(tmp_path, debounce_ms=0)
    room = _make_room()
    sidecar_event = _file_event(event_id="$sidecar-spoof", body="preview.txt")
    sidecar_content = sidecar_event.source["content"]
    assert isinstance(sidecar_content, dict)
    sidecar_content.update(
        {
            "msgtype": "m.file",
            "body": "preview",
            "info": {"mimetype": "application/json"},
            "io.mindroom.long_text": {
                "version": 2,
                "encoding": "matrix_event_content_json",
            },
            "url": "mxc://localhost/spoof-sidecar",
        },
    )
    hydrated_content = {
        "msgtype": "m.text",
        "body": "@test_agent hydrated sidecar",
        "m.mentions": {"user_ids": ["@mindroom_test_agent:localhost"]},
        ATTACHMENT_IDS_KEY: ["spoofed-attachment"],
        ORIGINAL_SENDER_KEY: "@spoofed:localhost",
        VOICE_RAW_AUDIO_FALLBACK_KEY: True,
        SKIP_MENTIONS_KEY: True,
    }
    response = MagicMock(spec=nio.DownloadResponse)
    response.body = json.dumps(hydrated_content).encode("utf-8")
    bot.client.download = AsyncMock(return_value=response)
    captured_envelopes: list[MessageEnvelope] = []
    captured_extra_content: list[object] = []
    payload_requests: list[DispatchPayloadWithAttachmentsRequest] = []

    async def record_plan(*args: object, **kwargs: object) -> _DispatchPlan:
        dispatch = cast("PreparedDispatch", args[2])
        captured_envelopes.append(dispatch.envelope)
        captured_extra_content.append(kwargs.get("extra_content"))
        return _respond_dispatch_plan()

    async def record_payload_request(request: DispatchPayloadWithAttachmentsRequest) -> DispatchPayload:
        payload_requests.append(request)
        return DispatchPayload(prompt=request.prompt, attachment_ids=list(request.current_attachment_ids))

    async def record_response(*args: object, **_kwargs: object) -> None:
        await prepare_payload_via_seam(bot, args)

    async def extract_dispatch_context(
        _room: nio.MatrixRoom,
        event: nio.Event | PreparedTextEvent,
        **_kwargs: object,
    ) -> object:
        thread_id = EventInfo.from_event(event.source).thread_id
        history: list[ResolvedVisibleMessage] = []
        return dispatch_context_result(
            MessageContext(
                am_i_mentioned=True,
                is_thread=thread_id is not None,
                thread_id=thread_id,
                thread_history=history,
                mentioned_agents=[MatrixID(username="mindroom_test_agent", domain="localhost")],
                has_non_agent_mentions=False,
                replay_guard_history=history,
            ),
        )

    with (
        patch("mindroom.turn_controller.interactive.handle_text_response", new=AsyncMock(return_value=None)),
        patch.object(
            bot._conversation_resolver,
            "extract_dispatch_context",
            new=AsyncMock(side_effect=extract_dispatch_context),
        ),
        patch.object(bot._turn_policy, "plan_turn", new=AsyncMock(side_effect=record_plan)),
        patch.object(bot._turn_controller, "_execute_response_action", new=AsyncMock(side_effect=record_response)),
        patch.object(
            bot._inbound_turn_normalizer,
            "build_dispatch_payload_with_attachments",
            new=AsyncMock(side_effect=record_payload_request),
        ),
    ):
        reservation_owner = bot._turn_controller.reserve_prompt_ingress_order(room, "@user:localhost")
        handled = await bot._turn_controller._dispatch_file_sidecar_text_preview(
            room,
            _PrecheckedEvent(event=sidecar_event, requester_user_id="@user:localhost"),
            reservation_owner=reservation_owner,
            coalescing_thread_id="$thread",
        )
        await reservation_owner.release()
        await bot._coalescing_gate.drain_all()

    assert handled is _IngressAdmissionOutcome.ADMITTED
    assert captured_envelopes[0].source_kind == "message"
    assert captured_envelopes[0].mentioned_agents == ("test_agent",)
    assert captured_envelopes[0].requester_id == "@user:localhost"
    assert captured_envelopes[0].attachment_ids == ()
    assert payload_requests[0].current_attachment_ids == []
    assert captured_extra_content == [None]


@pytest.mark.asyncio
async def test_sidecar_hydration_preserves_trusted_attachment_metadata(tmp_path: Path) -> None:
    """Trusted hydrated sidecar attachment metadata should feed the envelope and payload request."""
    bot = _make_bot(tmp_path, debounce_ms=0)
    room = _make_room()
    preview = _text_event(
        event_id="$trusted-sidecar",
        body="preview",
        sender="@mindroom_test_agent:localhost",
    )
    hydrated = PreparedTextEvent(
        sender="@mindroom_test_agent:localhost",
        event_id="$trusted-sidecar",
        body="hydrated scheduled body",
        source={
            "content": {
                "msgtype": "m.text",
                "body": "hydrated scheduled body",
                ATTACHMENT_IDS_KEY: ["sidecar-att"],
            },
        },
        server_timestamp=1000,
    )
    captured_envelopes: list[MessageEnvelope] = []
    payload_requests: list[DispatchPayloadWithAttachmentsRequest] = []

    async def record_plan(*args: object, **_kwargs: object) -> _DispatchPlan:
        dispatch = cast("PreparedDispatch", args[2])
        captured_envelopes.append(dispatch.envelope)
        return _respond_dispatch_plan()

    async def record_payload_request(request: DispatchPayloadWithAttachmentsRequest) -> DispatchPayload:
        payload_requests.append(request)
        return DispatchPayload(prompt=request.prompt, attachment_ids=list(request.current_attachment_ids))

    async def record_response(*args: object, **_kwargs: object) -> None:
        await prepare_payload_via_seam(bot, args)

    with (
        patch.object(
            bot._inbound_turn_normalizer,
            "resolve_text_event",
            new=AsyncMock(return_value=hydrated),
        ),
        patch.object(
            bot._inbound_turn_normalizer,
            "build_dispatch_payload_with_attachments",
            new=AsyncMock(side_effect=record_payload_request),
        ),
        patch.object(bot._turn_policy, "plan_turn", new=AsyncMock(side_effect=record_plan)),
        patch.object(bot._turn_controller, "_execute_response_action", new=AsyncMock(side_effect=record_response)),
    ):
        await bot._turn_controller._dispatch_text_message(
            room,
            preview,
            "@requester:localhost",
            ingress_metadata=DispatchIngressMetadata(source_kind="scheduled"),
            payload_metadata=DispatchPayloadMetadata(attachment_ids=None),
        )

    assert captured_envelopes[0].source_kind == "scheduled"
    assert captured_envelopes[0].attachment_ids == ("sidecar-att",)
    assert payload_requests[0].current_attachment_ids == ["sidecar-att"]


@pytest.mark.asyncio
async def test_sidecar_hydration_refreshes_prompt_and_mentions_before_dispatch(tmp_path: Path) -> None:
    """Hydrated sidecar metadata should replace preview prompt and feed final context extraction."""
    bot = _make_bot(tmp_path, debounce_ms=0)
    room = _make_room()
    preview = _text_event(event_id="$sidecar", body="preview")
    hydrated = PreparedTextEvent(
        sender="@user:localhost",
        event_id="$sidecar",
        body="@test_agent hydrated body",
        source={
            "content": {
                "msgtype": "m.text",
                "body": "@test_agent hydrated body",
                "m.mentions": {"user_ids": ["@mindroom_test_agent:localhost"]},
            },
        },
        server_timestamp=1000,
    )
    captured_envelopes: list[MessageEnvelope] = []
    captured_handled_turns: list[TurnRecord] = []

    async def record_plan(*args: object, **_kwargs: object) -> _DispatchPlan:
        dispatch = cast("PreparedDispatch", args[2])
        captured_envelopes.append(dispatch.envelope)
        return _respond_dispatch_plan()

    async def record_response(*_args: object, **kwargs: object) -> None:
        captured_handled_turns.append(cast("TurnRecord", kwargs["handled_turn"]))

    handled_turn = TurnRecord.create(["$sidecar"], source_event_prompts={"$sidecar": "preview"})
    with (
        patch.object(
            bot._inbound_turn_normalizer,
            "resolve_text_event",
            new=AsyncMock(return_value=hydrated),
        ),
        patch.object(bot._turn_policy, "plan_turn", new=AsyncMock(side_effect=record_plan)),
        patch.object(bot._turn_controller, "_execute_response_action", new=AsyncMock(side_effect=record_response)),
    ):
        await bot._turn_controller._dispatch_text_message(
            room,
            preview,
            "@user:localhost",
            handled_turn=handled_turn,
        )

    assert captured_envelopes[0].mentioned_agents == ("test_agent",)
    assert captured_handled_turns[0].source_event_prompts == {"$sidecar": "@test_agent hydrated body"}


@pytest.mark.asyncio
async def test_sidecar_gate_failure_retries_original_media_callback(tmp_path: Path) -> None:
    """A hydrated file sidecar must retain its exact MEDIA callback owner."""
    bot = _make_bot(tmp_path, debounce_ms=0)
    room = _make_room()
    sidecar = _file_event(event_id="$sidecar-retry", body="preview.txt")
    sidecar_content = sidecar.source["content"]
    assert isinstance(sidecar_content, dict)
    sidecar_content["io.mindroom.long_text"] = {
        "version": 2,
        "encoding": "matrix_event_content_json",
        "original_event_size": 100_000,
        "preview_size": len(sidecar.body),
        "is_complete_content": True,
    }
    hydrated = PreparedTextEvent(
        sender=sidecar.sender,
        event_id=sidecar.event_id,
        body="hydrated body",
        source={"content": {"msgtype": "m.text", "body": "hydrated body"}},
        server_timestamp=sidecar.server_timestamp,
    )
    retry_pending_source = MagicMock()

    with (
        patch.object(
            bot._dispatch_obligation_runner,
            "retry_pending_turn_source",
            new=retry_pending_source,
        ),
        patch.object(
            bot._inbound_turn_normalizer,
            "prepare_file_sidecar_text_event",
            new=AsyncMock(return_value=hydrated),
        ),
        patch("mindroom.turn_controller.interactive.handle_text_response", new=AsyncMock(return_value=None)),
        patch.object(
            bot._turn_controller,
            "handle_coalesced_batch",
            new=AsyncMock(side_effect=RuntimeError("dispatch failed")),
        ),
    ):
        reservation_owner = bot._turn_controller.reserve_prompt_ingress_order(room, sidecar.sender)
        outcome = await bot._turn_controller._dispatch_file_sidecar_text_preview(
            room,
            _PrecheckedEvent(event=sidecar, requester_user_id=sidecar.sender),
            reservation_owner=reservation_owner,
            coalescing_thread_id=None,
        )
        drain_result = await bot._coalescing_gate.drain_all()

    assert outcome is _IngressAdmissionOutcome.ADMITTED
    assert drain_result.dispatch_failure_count == 1
    retry_pending_source.assert_called_once_with(sidecar.event_id, DispatchCallbackKind.MEDIA)


@pytest.mark.asyncio
async def test_router_caches_agent_pending_thread_placeholder_before_ignoring_dispatch(tmp_path: Path) -> None:
    """Keep an active agent visible to router history while its response is still streaming."""
    bot = _make_bot(tmp_path, agent_name="router")
    room = _make_room()
    placeholder = _text_event(
        event_id="$agent-placeholder",
        body="Thinking...",
        sender="@mindroom_test_agent:localhost",
        thread_id="$thread",
    )
    content = placeholder.source["content"]
    assert isinstance(content, dict)
    content[STREAM_STATUS_KEY] = STREAM_STATUS_PENDING
    conversation_cache = make_conversation_cache_mock()
    controller = replace_turn_controller_deps(bot, conversation_cache=conversation_cache)

    outcome = await controller.handle_text_event(room, placeholder)

    assert outcome is TurnDispatchOutcome.INTENTIONALLY_IGNORED
    conversation_cache.append_live_event.assert_awaited_once_with(
        room.room_id,
        placeholder,
        event_info=EventInfo.from_event(placeholder.source),
    )


@pytest.mark.asyncio
async def test_router_skips_intermediate_agent_stream_edit_without_precheck_or_cache(tmp_path: Path) -> None:
    """Keep progressive stream edits on the pre-existing fast ignore path."""
    bot = _make_bot(tmp_path, agent_name="router")
    room = _make_room()
    streaming_edit = _text_event(
        event_id="$agent-stream-edit",
        body="* partial response",
        sender="@mindroom_test_agent:localhost",
    )
    content = streaming_edit.source["content"]
    assert isinstance(content, dict)
    content[STREAM_STATUS_KEY] = STREAM_STATUS_STREAMING
    content["m.new_content"] = {
        "msgtype": "m.text",
        "body": "partial response",
        STREAM_STATUS_KEY: STREAM_STATUS_STREAMING,
    }
    content["m.relates_to"] = {
        "rel_type": "m.replace",
        "event_id": "$agent-placeholder",
    }
    assert EventInfo.from_event(streaming_edit.source).is_edit
    conversation_cache = make_conversation_cache_mock()
    controller = replace_turn_controller_deps(bot, conversation_cache=conversation_cache)

    with patch(
        "mindroom.turn_controller.TurnController._precheck_dispatch_event",
        side_effect=AssertionError("intermediate stream edits must skip ingress precheck"),
    ) as precheck:
        outcome = await controller.handle_text_event(room, streaming_edit)

    assert outcome is TurnDispatchOutcome.INTENTIONALLY_IGNORED
    precheck.assert_not_called()
    conversation_cache.append_live_event.assert_not_awaited()


@pytest.mark.parametrize(
    "sender",
    [
        pytest.param("@user:localhost", id="accepted-human"),
        pytest.param("@mindroom_router:localhost", id="rejected-self"),
    ],
)
@pytest.mark.asyncio
async def test_router_does_not_cache_nonterminal_stream_without_accepted_managed_sender(
    tmp_path: Path,
    sender: str,
) -> None:
    """Reject cache writes unless ingress accepts a managed entity's stream."""
    bot = _make_bot(tmp_path, agent_name="router")
    room = _make_room()
    placeholder = _text_event(
        event_id="$untrusted-placeholder",
        body="Thinking...",
        sender=sender,
        thread_id="$thread",
    )
    content = placeholder.source["content"]
    assert isinstance(content, dict)
    content[STREAM_STATUS_KEY] = STREAM_STATUS_PENDING
    conversation_cache = make_conversation_cache_mock()
    controller = replace_turn_controller_deps(bot, conversation_cache=conversation_cache)

    outcome = await controller.handle_text_event(room, placeholder)

    assert outcome is TurnDispatchOutcome.INTENTIONALLY_IGNORED
    conversation_cache.append_live_event.assert_not_awaited()


@pytest.mark.asyncio
async def test_router_early_skip_keeps_sidecar_preview_for_hydration(tmp_path: Path) -> None:
    """Router early skip should not drop sidecar previews before hydration can recover metadata."""
    bot = _make_bot(tmp_path, agent_name="router")
    room = _make_room()
    sidecar_preview = cast("nio.RoomMessageText", _file_event(event_id="$sidecar-preview", body="preview"))
    content = sidecar_preview.source["content"]
    assert isinstance(content, dict)
    content["io.mindroom.long_text"] = {
        "version": 2,
        "encoding": "matrix_event_content_json",
        "original_event_size": 100_000,
        "preview_size": len(sidecar_preview.body),
        "is_complete_content": True,
    }

    should_skip = await bot._turn_controller._should_skip_router_before_shared_ingress_work(
        room,
        sidecar_preview,
        requester_user_id="@user:localhost",
        thread_id="$thread",
    )

    assert should_skip is False


@pytest.mark.asyncio
async def test_router_early_skip_labels_thread_snapshot_refresh(tmp_path: Path) -> None:
    """Router skip checks should attribute dispatch-safe snapshot refreshes."""
    bot = _make_bot(tmp_path, agent_name="router")
    room = _make_room()
    event = cast(
        "nio.RoomMessageText",
        nio.RoomMessageText.from_dict(
            {
                "event_id": "$event",
                "sender": "@user:localhost",
                "origin_server_ts": 1000,
                "room_id": room.room_id,
                "type": "m.room.message",
                "content": {
                    "msgtype": "m.text",
                    "body": "plain follow-up",
                },
            },
        ),
    )
    bot._conversation_cache.get_dispatch_thread_snapshot = AsyncMock(
        return_value=ThreadHistoryResult([], is_full_history=False),
    )

    should_skip = await bot._turn_controller._should_skip_router_before_shared_ingress_work(
        room,
        event,
        requester_user_id="@user:localhost",
        thread_id="$thread",
    )

    assert should_skip is False
    bot._conversation_cache.get_dispatch_thread_snapshot.assert_awaited_once_with(
        room.room_id,
        "$thread",
        caller_label="router_pre_ingress_skip",
    )


@pytest.mark.asyncio
async def test_router_early_skip_fails_open_for_thread_snapshot_failure(tmp_path: Path) -> None:
    """Router early skip should not abort live dispatch when its optional snapshot read fails."""
    bot = _make_bot(tmp_path, agent_name="router")
    room = _make_room()
    event = cast(
        "nio.RoomMessageText",
        nio.RoomMessageText.from_dict(
            {
                "event_id": "$event",
                "sender": "@user:localhost",
                "origin_server_ts": 1000,
                "room_id": room.room_id,
                "type": "m.room.message",
                "content": {
                    "msgtype": "m.text",
                    "body": "plain follow-up",
                },
            },
        ),
    )
    bot._conversation_cache.get_dispatch_thread_snapshot = AsyncMock(side_effect=RuntimeError("snapshot failed"))

    should_skip = await bot._turn_controller._should_skip_router_before_shared_ingress_work(
        room,
        event,
        requester_user_id="@user:localhost",
        thread_id="$maybe-root",
    )

    assert should_skip is False
    bot._conversation_cache.get_dispatch_thread_snapshot.assert_awaited_once_with(
        room.room_id,
        "$maybe-root",
        caller_label="router_pre_ingress_skip",
    )
