"""Event cache write coordination: outbound update queueing, streaming-edit coalescing, timing logs, and same-room serialization."""

from __future__ import annotations

import asyncio
from typing import TYPE_CHECKING
from unittest.mock import AsyncMock, MagicMock, Mock, patch

import nio
import pytest

import mindroom.timing as timing_module
from mindroom.background_tasks import wait_for_background_tasks
from mindroom.constants import (
    STREAM_STATUS_CANCELLED,
    STREAM_STATUS_COMPLETED,
    STREAM_STATUS_ERROR,
    STREAM_STATUS_KEY,
    STREAM_STATUS_PENDING,
    STREAM_STATUS_STREAMING,
)
from mindroom.matrix.cache import ThreadHistoryResult, thread_writes
from mindroom.matrix.cache.outbound_thread_reservations import OutboundThreadReservations
from mindroom.matrix.cache.thread_cache_state import ThreadAppendOutcome
from mindroom.matrix.cache.write_coordinator import EventCacheWriteCoordinator
from mindroom.matrix.conversation_cache import MatrixConversationCache
from mindroom.matrix.event_info import EventInfo
from mindroom.matrix.thread_bookkeeping import MutationThreadImpact
from tests.threading_helpers import (
    ThreadingBehaviorTestBase,
    _conversation_runtime,
    _make_client_mock,
    _message,
    _outbound_plain_edit_content,
    _outbound_streaming_edit_content,
    _runtime_event_cache,
    _runtime_write_coordinator,
    _thread_mutation_cache_ops,
    _wait_for_room_cache_idle,
    thread_history_result,
)

if TYPE_CHECKING:
    from collections.abc import Callable, Coroutine
    from typing import Any

    from mindroom.bot import AgentBot


class TestThreadingBehavior(ThreadingBehaviorTestBase):
    """Threading behavior tests moved verbatim from tests/test_threading_error.py."""

    @pytest.mark.asyncio
    async def test_get_event_queues_persistent_cache_fill_through_room_write_barrier(self) -> None:
        """Point-event cache fills should use the same room-ordered coordinator as other durable writes."""
        event_cache = _runtime_event_cache()
        event_cache.get_event = AsyncMock(return_value=None)
        event_cache.store_event = AsyncMock()
        coordinator = _runtime_write_coordinator()
        client = _make_client_mock()
        client.room_get_event = AsyncMock(
            return_value=nio.RoomGetEventResponse.from_dict(
                {
                    "content": {"body": "hello", "msgtype": "m.text"},
                    "event_id": "$event:localhost",
                    "sender": "@user:localhost",
                    "origin_server_ts": 1234567890,
                    "room_id": "!test:localhost",
                    "type": "m.room.message",
                },
            ),
        )
        access = MatrixConversationCache(
            logger=MagicMock(),
            runtime=_conversation_runtime(
                client=client,
                event_cache=event_cache,
                coordinator=coordinator,
            ),
        )

        with patch.object(coordinator, "queue_room_update", wraps=coordinator.queue_room_update) as mock_queue:
            await access.get_event("!test:localhost", "  $event:localhost  ")

        event_cache.store_event.assert_awaited_once()
        stored_event_id, stored_room_id, stored_event_source = event_cache.store_event.await_args.args
        assert stored_event_id == "$event:localhost"
        assert stored_room_id == "!test:localhost"
        assert stored_event_source["event_id"] == "$event:localhost"
        mock_queue.assert_called_once()

    @pytest.mark.asyncio
    async def test_same_turn_point_cache_hit_skips_room_write_barrier(self) -> None:
        """A memoized point read should not wait behind unrelated room writes."""
        event_cache = _runtime_event_cache()
        coordinator = _runtime_write_coordinator()
        response = nio.RoomGetEventResponse.from_dict(
            {
                "content": {"body": "hello", "msgtype": "m.text"},
                "event_id": "$event:localhost",
                "sender": "@user:localhost",
                "origin_server_ts": 1234567890,
                "room_id": "!test:localhost",
                "type": "m.room.message",
            },
        )
        access = MatrixConversationCache(
            logger=MagicMock(),
            runtime=_conversation_runtime(
                client=_make_client_mock(),
                event_cache=event_cache,
                coordinator=coordinator,
            ),
        )

        with (
            patch(
                "mindroom.matrix.conversation_cache._cached_room_get_event",
                new=AsyncMock(return_value=(response, None)),
            ) as get_event,
            patch.object(
                coordinator,
                "wait_for_prior_room_updates",
                new=AsyncMock(),
            ) as wait_for_prior_room_updates,
        ):
            async with access.turn_scope():
                first = await access.get_event("!test:localhost", "$event:localhost")
                second = await access.get_event("!test:localhost", "$event:localhost")

        assert first is response
        assert second is response
        get_event.assert_awaited_once()
        wait_for_prior_room_updates.assert_awaited_once_with(
            "!test:localhost",
            coordination_scope=event_cache.principal_id,
        )

    @pytest.mark.asyncio
    async def test_local_bot_redaction_ignores_cache_failure_after_successful_redact(self, bot: AgentBot) -> None:
        """A successful local redact should delegate advisory bookkeeping through the cache facade."""
        bot.client = AsyncMock(spec=nio.AsyncClient)
        bot.client.room_redact = AsyncMock(
            return_value=nio.RoomRedactResponse(
                event_id="$redaction:localhost",
                room_id="!test:localhost",
            ),
        )
        bot._conversation_cache.notify_outbound_redaction = Mock()

        result = await bot._redact_message_event(
            room_id="!test:localhost",
            event_id="$target:localhost",
            reason="cleanup",
        )

        assert result is True
        bot.client.room_redact.assert_awaited_once_with(
            "!test:localhost",
            "$target:localhost",
            reason="cleanup",
        )
        bot._conversation_cache.notify_outbound_redaction.assert_called_once_with(
            "!test:localhost",
            "$target:localhost",
        )

    @pytest.mark.asyncio
    async def test_queue_room_cache_update_forwards_false_emit_timing(self) -> None:
        """Room cache facade must not fall through to the coordinator timing default."""
        cache_ops, _logger, _event_cache = _thread_mutation_cache_ops()
        observed_emit_timing: list[bool] = []

        class _RecordingCoordinator:
            def queue_room_update(
                self,
                room_id: str,
                update_coro_factory: Callable[[], Coroutine[Any, Any, object]],
                *,
                name: str,
                emit_timing: bool = True,
                coalesce_key: tuple[str, str] | None = None,
                coalesce_log_context: dict[str, object] | None = None,
                coordination_scope: str | None = None,
            ) -> asyncio.Task[object]:
                del room_id, name, coalesce_key, coalesce_log_context, coordination_scope
                observed_emit_timing.append(emit_timing)
                return asyncio.create_task(update_coro_factory())

        async def update() -> None:
            return None

        cache_ops.runtime.event_cache_write_coordinator = _RecordingCoordinator()
        task = cache_ops.queue_room_cache_update("!room:localhost", update, name="matrix_cache_test_update")
        await task

        assert observed_emit_timing == [False]

    @pytest.mark.asyncio
    async def test_queue_thread_cache_update_forwards_default_coordinator_options(self) -> None:
        """Thread cache facade should always forward the expanded coordinator options."""
        cache_ops, _logger, event_cache = _thread_mutation_cache_ops()
        observed_options: list[tuple[object, object, object, object]] = []

        class _RecordingCoordinator:
            def queue_thread_update(
                self,
                room_id: str,
                thread_id: str,
                update_coro_factory: Callable[[], Coroutine[Any, Any, object]],
                *,
                name: str,
                emit_timing: object = "missing",
                coalesce_key: object = "missing",
                coalesce_log_context: object = "missing",
                coordination_scope: object = "missing",
            ) -> asyncio.Task[object]:
                del room_id, thread_id, name
                observed_options.append((emit_timing, coalesce_key, coalesce_log_context, coordination_scope))
                return asyncio.create_task(update_coro_factory())

        async def update() -> None:
            return None

        cache_ops.runtime.event_cache_write_coordinator = _RecordingCoordinator()
        task = cache_ops.queue_thread_cache_update(
            "!room:localhost",
            "$thread:localhost",
            update,
            name="matrix_cache_test_update",
        )
        await task

        assert observed_options == [(False, None, None, event_cache.principal_id)]

    @pytest.mark.asyncio
    async def test_ten_reserved_thread_streams_overlap_and_preserve_each_thread_order(self) -> None:
        """Reserved edits should overlap across threads while retaining exact per-thread order."""
        room_id = "!test:localhost"
        thread_count = 10
        event_cache = _runtime_event_cache()
        coordinator = EventCacheWriteCoordinator(
            logger=MagicMock(),
            background_task_owner=object(),
        )
        access = MatrixConversationCache(
            logger=MagicMock(),
            runtime=_conversation_runtime(
                client=_make_client_mock(),
                event_cache=event_cache,
                coordinator=coordinator,
            ),
        )
        thread_ids = [f"$thread-{index}:localhost" for index in range(thread_count)]
        original_event_ids = [f"$stream-{index}:localhost" for index in range(thread_count)]

        for thread_id, original_event_id in zip(thread_ids, original_event_ids, strict=True):
            access.notify_outbound_message(
                room_id,
                original_event_id,
                {
                    "body": "Working",
                    "msgtype": "m.notice",
                    STREAM_STATUS_KEY: STREAM_STATUS_PENDING,
                    "m.relates_to": {
                        "rel_type": "m.thread",
                        "event_id": thread_id,
                        "is_falling_back": True,
                        "m.in_reply_to": {"event_id": thread_id},
                    },
                },
            )
        await asyncio.gather(
            *(
                coordinator.wait_for_thread_idle(
                    room_id,
                    thread_id,
                    coordination_scope=event_cache.principal_id,
                )
                for thread_id in thread_ids
            ),
        )
        assert access._outbound._reservations.active_count == thread_count

        event_cache.apply_thread_mutation_append.reset_mock()
        entered_by_thread = {thread_id: asyncio.Event() for thread_id in thread_ids}
        release_first_edits = asyncio.Event()
        active_threads: set[str] = set()
        max_active_thread_count = 0
        appended_event_ids_by_thread: dict[str, list[str]] = {thread_id: [] for thread_id in thread_ids}

        async def blocking_record_append(
            _room_id: str,
            thread_id: str,
            event_source: dict[str, object],
            *,
            append_failed_reason: str,
        ) -> ThreadAppendOutcome:
            nonlocal max_active_thread_count
            assert append_failed_reason == "outbound_append_failed"
            active_threads.add(thread_id)
            max_active_thread_count = max(max_active_thread_count, len(active_threads))
            entered_by_thread[thread_id].set()
            await release_first_edits.wait()
            active_threads.remove(thread_id)
            appended_event_ids_by_thread[thread_id].append(str(event_source["event_id"]))
            return ThreadAppendOutcome.APPENDED

        event_cache.apply_thread_mutation_append = AsyncMock(side_effect=blocking_record_append)

        with (
            patch.object(coordinator, "queue_room_update", wraps=coordinator.queue_room_update) as room_updates,
            patch.object(coordinator, "queue_thread_update", wraps=coordinator.queue_thread_update) as thread_updates,
        ):
            try:
                for stream_index, (_thread_id, original_event_id) in enumerate(
                    zip(thread_ids, original_event_ids, strict=True),
                ):
                    for edit_index in range(3):
                        body = f"stream {stream_index} update {edit_index}"
                        access.notify_outbound_message(
                            room_id,
                            f"$edit-{stream_index}-{edit_index}:localhost",
                            {
                                "body": f"* {body}",
                                "msgtype": "m.notice",
                                "m.new_content": {
                                    "body": body,
                                    "msgtype": "m.notice",
                                    STREAM_STATUS_KEY: STREAM_STATUS_STREAMING,
                                },
                                "m.relates_to": {
                                    "rel_type": "m.replace",
                                    "event_id": original_event_id,
                                },
                            },
                        )
                    final_body = f"stream {stream_index} final"
                    access.notify_outbound_message(
                        room_id,
                        f"$edit-{stream_index}-final:localhost",
                        {
                            "body": f"* {final_body}",
                            "msgtype": "m.text",
                            "m.new_content": {
                                "body": final_body,
                                "msgtype": "m.text",
                                STREAM_STATUS_KEY: STREAM_STATUS_COMPLETED,
                            },
                            "m.relates_to": {
                                "rel_type": "m.replace",
                                "event_id": original_event_id,
                            },
                        },
                    )

                await asyncio.wait_for(
                    asyncio.gather(*(entered.wait() for entered in entered_by_thread.values())),
                    timeout=1.0,
                )
                assert max_active_thread_count == thread_count
                assert room_updates.call_count == 0
                assert thread_updates.call_count == thread_count * 4
            finally:
                release_first_edits.set()
                await coordinator.close()

        for stream_index, thread_id in enumerate(thread_ids):
            assert appended_event_ids_by_thread[thread_id] == [
                f"$edit-{stream_index}-0:localhost",
                f"$edit-{stream_index}-2:localhost",
                f"$edit-{stream_index}-final:localhost",
            ]
        assert access._outbound._reservations.active_count == 0

    @pytest.mark.asyncio
    @pytest.mark.parametrize("terminal_status", [STREAM_STATUS_CANCELLED, STREAM_STATUS_ERROR])
    async def test_terminal_failure_releases_reservation_after_queued_edits(
        self,
        terminal_status: str,
    ) -> None:
        """Failure and cancellation should release claims after capturing terminal thread scope."""
        room_id = "!test:localhost"
        thread_id = "$thread:localhost"
        original_event_id = "$stream:localhost"
        event_cache = _runtime_event_cache()
        coordinator = EventCacheWriteCoordinator(
            logger=MagicMock(),
            background_task_owner=object(),
        )
        access = MatrixConversationCache(
            logger=MagicMock(),
            runtime=_conversation_runtime(
                client=_make_client_mock(),
                event_cache=event_cache,
                coordinator=coordinator,
            ),
        )
        access.notify_outbound_message(
            room_id,
            original_event_id,
            {
                "body": "Working",
                "msgtype": "m.notice",
                STREAM_STATUS_KEY: STREAM_STATUS_PENDING,
                "m.relates_to": {"rel_type": "m.thread", "event_id": thread_id},
            },
        )
        await coordinator.wait_for_thread_idle(
            room_id,
            thread_id,
            coordination_scope=event_cache.principal_id,
        )
        assert access._outbound._reservations.active_count == 1
        event_cache.apply_thread_mutation_append.reset_mock()

        for event_id, body, stream_status in (
            ("$edit-progress:localhost", "partial", STREAM_STATUS_STREAMING),
            ("$edit-terminal:localhost", "terminal", terminal_status),
        ):
            access.notify_outbound_message(
                room_id,
                event_id,
                {
                    "body": f"* {body}",
                    "msgtype": "m.text",
                    "m.new_content": {
                        "body": body,
                        "msgtype": "m.text",
                        STREAM_STATUS_KEY: stream_status,
                    },
                    "m.relates_to": {
                        "rel_type": "m.replace",
                        "event_id": original_event_id,
                    },
                },
            )

        assert access._outbound._reservations.active_count == 0
        await coordinator.wait_for_thread_idle(
            room_id,
            thread_id,
            coordination_scope=event_cache.principal_id,
        )
        appended_event_ids = [
            call.args[2]["event_id"] for call in event_cache.apply_thread_mutation_append.await_args_list
        ]
        assert appended_event_ids == ["$edit-progress:localhost", "$edit-terminal:localhost"]
        assert not coordinator._room_states

    @pytest.mark.asyncio
    async def test_edit_event_id_transition_retains_thread_without_duplicate_terminal_write(self) -> None:
        """A retained edit alias should route a transitioned terminal target to the same thread."""
        room_id = "!test:localhost"
        thread_id = "$thread:localhost"
        original_event_id = "$stream:localhost"
        first_edit_event_id = "$edit-first:localhost"
        event_cache = _runtime_event_cache()
        coordinator = EventCacheWriteCoordinator(
            logger=MagicMock(),
            background_task_owner=object(),
        )
        access = MatrixConversationCache(
            logger=MagicMock(),
            runtime=_conversation_runtime(
                client=_make_client_mock(),
                event_cache=event_cache,
                coordinator=coordinator,
            ),
        )
        access.notify_outbound_message(
            room_id,
            original_event_id,
            {
                "body": "Working",
                "msgtype": "m.notice",
                STREAM_STATUS_KEY: STREAM_STATUS_PENDING,
                "m.relates_to": {"rel_type": "m.thread", "event_id": thread_id},
            },
        )
        await coordinator.wait_for_thread_idle(
            room_id,
            thread_id,
            coordination_scope=event_cache.principal_id,
        )
        event_cache.apply_thread_mutation_append.reset_mock()
        access.notify_outbound_message(
            room_id,
            first_edit_event_id,
            {
                "body": "* partial",
                "msgtype": "m.notice",
                "m.new_content": {
                    "body": "partial",
                    "msgtype": "m.notice",
                    STREAM_STATUS_KEY: STREAM_STATUS_STREAMING,
                },
                "m.relates_to": {
                    "rel_type": "m.replace",
                    "event_id": original_event_id,
                },
            },
        )
        access.notify_outbound_message(
            room_id,
            "$edit-terminal:localhost",
            {
                "body": "* final",
                "msgtype": "m.text",
                "m.new_content": {
                    "body": "final",
                    "msgtype": "m.text",
                    STREAM_STATUS_KEY: STREAM_STATUS_COMPLETED,
                },
                "m.relates_to": {
                    "rel_type": "m.replace",
                    "event_id": first_edit_event_id,
                },
            },
        )

        await coordinator.wait_for_thread_idle(
            room_id,
            thread_id,
            coordination_scope=event_cache.principal_id,
        )
        assert [call.args[2]["event_id"] for call in event_cache.apply_thread_mutation_append.await_args_list] == [
            first_edit_event_id,
            "$edit-terminal:localhost",
        ]
        assert all(call.args[1] == thread_id for call in event_cache.apply_thread_mutation_append.await_args_list)
        assert access._outbound._reservations.active_count == 0

    def test_thread_reservations_isolate_principals_rooms_and_expire(self) -> None:
        """Reservation keys should not collide and lazy TTL cleanup should bound stale claims."""
        now = 10.0
        reservations = OutboundThreadReservations(
            ttl_seconds=5.0,
            max_reservations=3,
            clock=lambda: now,
        )
        reservations.reserve("principal-a", "!room-a", "$event", "$thread-a")
        reservations.reserve("principal-b", "!room-a", "$event", "$thread-b")
        reservations.reserve("principal-a", "!room-b", "$event", "$thread-c")

        assert reservations.resolve("principal-a", "!room-a", "$event") == "$thread-a"
        assert reservations.resolve("principal-b", "!room-a", "$event") == "$thread-b"
        assert reservations.resolve("principal-a", "!room-b", "$event") == "$thread-c"

        now = 16.0
        assert reservations.active_count == 0
        assert reservations.resolve("principal-a", "!room-a", "$event") is None

    def test_thread_reservations_evict_soonest_expiring_claim_at_capacity(self) -> None:
        """Capacity cleanup should evict the claim nearest expiration."""
        now = 10.0
        reservations = OutboundThreadReservations(
            ttl_seconds=5.0,
            max_reservations=2,
            clock=lambda: now,
        )
        reservations.reserve("principal-a", "!room", "$oldest", "$thread-oldest")
        now = 11.0
        reservations.reserve("principal-a", "!room", "$newer", "$thread-newer")
        reservations.reserve("principal-a", "!room", "$newest", "$thread-newest")

        assert reservations.active_count == 2
        assert reservations.resolve("principal-a", "!room", "$oldest") is None
        assert reservations.resolve("principal-a", "!room", "$newer") == "$thread-newer"
        assert reservations.resolve("principal-a", "!room", "$newest") == "$thread-newest"

    @pytest.mark.asyncio
    async def test_outbound_nonterminal_streaming_edits_coalesce_pending_cache_updates(
        self,
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        """Pending outbound stream edits for the same event should collapse to the latest edit."""
        monkeypatch.setenv("MINDROOM_TIMING", "1")
        timing_logger = MagicMock()
        monkeypatch.setattr(timing_module, "logger", timing_logger)
        event_cache = _runtime_event_cache()
        coordinator_logger = MagicMock()
        coordinator = EventCacheWriteCoordinator(
            logger=coordinator_logger,
            background_task_owner=object(),
        )
        access = MatrixConversationCache(
            logger=MagicMock(),
            runtime=_conversation_runtime(
                client=_make_client_mock(),
                event_cache=event_cache,
                coordinator=coordinator,
            ),
        )
        blocker_started = asyncio.Event()
        release_blocker = asyncio.Event()

        async def blocker() -> None:
            blocker_started.set()
            await release_blocker.wait()

        try:
            blocker_task = coordinator.queue_thread_update(
                "!test:localhost",
                "$thread:localhost",
                blocker,
                name="matrix_cache_blocker",
                coordination_scope=event_cache.principal_id,
            )
            await asyncio.wait_for(blocker_started.wait(), timeout=1.0)

            for index in range(3):
                access.notify_outbound_message(
                    "!test:localhost",
                    f"$edit-{index}:localhost",
                    _outbound_streaming_edit_content(body=f"stream update {index}"),
                )

            release_blocker.set()
            await blocker_task
            await asyncio.wait_for(
                coordinator.wait_for_thread_idle(
                    "!test:localhost",
                    "$thread:localhost",
                    coordination_scope=event_cache.principal_id,
                ),
                timeout=1.0,
            )
        finally:
            release_blocker.set()
            await coordinator.close()

        event_cache.apply_thread_mutation_append.assert_awaited_once()
        _room_id, _thread_id, appended_event = event_cache.apply_thread_mutation_append.await_args.args
        assert appended_event["event_id"] == "$edit-2:localhost"
        assert appended_event["content"]["m.new_content"]["body"] == "stream update 2"
        assert any(
            call.kwargs.get("coalesced_update_count") == 2
            and call.kwargs.get("stream_status") == STREAM_STATUS_STREAMING
            and "original_event_id" not in call.kwargs
            and "latest_event_id" not in call.kwargs
            for call in coordinator_logger.info.call_args_list
        )
        assert any(
            call.args == ("Event cache update timing",)
            and call.kwargs["barrier_kind"] == "thread"
            and call.kwargs["operation"] == "matrix_cache_notify_outbound_event"
            and call.kwargs["coalesced_update_count"] == 2
            and call.kwargs["predecessor_wait_ms"] >= 0.0
            and call.kwargs["update_run_ms"] >= 0.0
            and "room_id" not in call.kwargs
            and "thread_id" not in call.kwargs
            for call in timing_logger.debug.call_args_list
        )

    @pytest.mark.asyncio
    async def test_outbound_final_streaming_edit_is_preserved_after_coalesced_intermediates(self) -> None:
        """A final outbound stream edit should stay queued behind the latest intermediate edit."""
        event_cache = _runtime_event_cache()
        coordinator = EventCacheWriteCoordinator(
            logger=MagicMock(),
            background_task_owner=object(),
        )
        access = MatrixConversationCache(
            logger=MagicMock(),
            runtime=_conversation_runtime(
                client=_make_client_mock(),
                event_cache=event_cache,
                coordinator=coordinator,
            ),
        )
        blocker_started = asyncio.Event()
        release_blocker = asyncio.Event()

        async def blocker() -> None:
            blocker_started.set()
            await release_blocker.wait()

        try:
            blocker_task = coordinator.queue_thread_update(
                "!test:localhost",
                "$thread:localhost",
                blocker,
                name="matrix_cache_blocker",
                coordination_scope=event_cache.principal_id,
            )
            await asyncio.wait_for(blocker_started.wait(), timeout=1.0)

            access.notify_outbound_message(
                "!test:localhost",
                "$edit-1:localhost",
                _outbound_streaming_edit_content(body="older intermediate"),
            )
            access.notify_outbound_message(
                "!test:localhost",
                "$edit-2:localhost",
                _outbound_streaming_edit_content(body="latest intermediate"),
            )
            access.notify_outbound_message(
                "!test:localhost",
                "$edit-final:localhost",
                _outbound_streaming_edit_content(
                    body="final answer",
                    stream_status=STREAM_STATUS_COMPLETED,
                ),
            )

            release_blocker.set()
            await blocker_task
            await asyncio.wait_for(
                coordinator.wait_for_thread_idle(
                    "!test:localhost",
                    "$thread:localhost",
                    coordination_scope=event_cache.principal_id,
                ),
                timeout=1.0,
            )
        finally:
            release_blocker.set()
            await coordinator.close()

        appended_event_ids = [
            call.args[2]["event_id"] for call in event_cache.apply_thread_mutation_append.await_args_list
        ]
        assert appended_event_ids == ["$edit-2:localhost", "$edit-final:localhost"]
        final_content = event_cache.apply_thread_mutation_append.await_args_list[-1].args[2]["content"]
        assert final_content["m.new_content"][STREAM_STATUS_KEY] == STREAM_STATUS_COMPLETED

    @pytest.mark.asyncio
    async def test_outbound_plain_edits_are_not_coalesced(self) -> None:
        """Normal outbound edits should retain the existing one-cache-update-per-edit behavior."""
        event_cache = _runtime_event_cache()
        coordinator = EventCacheWriteCoordinator(
            logger=MagicMock(),
            background_task_owner=object(),
        )
        access = MatrixConversationCache(
            logger=MagicMock(),
            runtime=_conversation_runtime(
                client=_make_client_mock(),
                event_cache=event_cache,
                coordinator=coordinator,
            ),
        )
        blocker_started = asyncio.Event()
        release_blocker = asyncio.Event()

        async def blocker() -> None:
            blocker_started.set()
            await release_blocker.wait()

        try:
            blocker_task = coordinator.queue_thread_update(
                "!test:localhost",
                "$thread:localhost",
                blocker,
                name="matrix_cache_blocker",
                coordination_scope="test-principal",
            )
            await asyncio.wait_for(blocker_started.wait(), timeout=1.0)

            access.notify_outbound_message(
                "!test:localhost",
                "$plain-edit-1:localhost",
                _outbound_plain_edit_content(body="plain edit 1"),
            )
            access.notify_outbound_message(
                "!test:localhost",
                "$plain-edit-2:localhost",
                _outbound_plain_edit_content(body="plain edit 2"),
            )

            release_blocker.set()
            await blocker_task
            await asyncio.wait_for(
                coordinator.wait_for_thread_idle(
                    "!test:localhost",
                    "$thread:localhost",
                    coordination_scope="test-principal",
                ),
                timeout=1.0,
            )
        finally:
            release_blocker.set()
            await coordinator.close()

        appended_event_ids = [
            call.args[2]["event_id"] for call in event_cache.apply_thread_mutation_append.await_args_list
        ]
        assert appended_event_ids == ["$plain-edit-1:localhost", "$plain-edit-2:localhost"]

    @pytest.mark.asyncio
    async def test_queue_room_update_logs_timing_breakdown_when_enabled(
        self,
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        """Room-scoped cache updates should log predecessor wait versus update time."""
        monkeypatch.setenv("MINDROOM_TIMING", "1")
        timing_logger = MagicMock()
        monkeypatch.setattr(timing_module, "logger", timing_logger)
        coordinator = EventCacheWriteCoordinator(
            logger=MagicMock(),
            background_task_owner=object(),
        )
        first_started = asyncio.Event()
        allow_first_finish = asyncio.Event()

        async def first_update() -> str:
            first_started.set()
            await allow_first_finish.wait()
            return "first"

        async def second_update() -> str:
            return "second"

        try:
            first_task = coordinator.queue_room_update(
                "!room:localhost",
                first_update,
                name="matrix_cache_first_update",
                coordination_scope="test-principal",
            )
            await asyncio.wait_for(first_started.wait(), timeout=1.0)
            second_task = coordinator.queue_room_update(
                "!room:localhost",
                second_update,
                name="matrix_cache_second_update",
                coordination_scope="test-principal",
            )
            await asyncio.sleep(0)
            allow_first_finish.set()
            assert await first_task == "first"
            assert await second_task == "second"
        finally:
            await coordinator.close()

        timing_calls = [
            call for call in timing_logger.debug.call_args_list if call.args == ("Event cache update timing",)
        ]
        assert any(
            call.kwargs["barrier_kind"] == "room"
            and call.kwargs["operation"] == "matrix_cache_second_update"
            and call.kwargs["queued_behind_predecessor"] is True
            and call.kwargs["predecessor_count"] >= 1
            and call.kwargs["predecessor_wait_ms"] >= 0.0
            and call.kwargs["update_run_ms"] >= 0.0
            and call.kwargs["total_ms"] >= call.kwargs["update_run_ms"]
            for call in timing_calls
        )

    @pytest.mark.asyncio
    async def test_queue_room_update_logs_wait_from_raw_interval_when_enabled(
        self,
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        """Predecessor wait should come from the raw pre-update interval, not rounded subtraction."""
        monkeypatch.setenv("MINDROOM_TIMING", "1")
        timing_logger = MagicMock()
        monkeypatch.setattr(timing_module, "logger", timing_logger)
        perf_counter_values = iter([0.0, 0.00002, 0.00016, 0.00016])
        monkeypatch.setattr(
            "mindroom.matrix.cache.write_coordinator.time.perf_counter",
            lambda: next(perf_counter_values),
        )
        coordinator = EventCacheWriteCoordinator(
            logger=MagicMock(),
            background_task_owner=object(),
        )

        async def update() -> str:
            return "ok"

        try:
            task = coordinator.queue_room_update(
                "!room:localhost",
                update,
                name="matrix_cache_single_update",
                coordination_scope="test-principal",
            )
            assert await task == "ok"
        finally:
            await coordinator.close()

        timing_call = next(
            call
            for call in timing_logger.debug.call_args_list
            if call.args == ("Event cache update timing",) and call.kwargs["operation"] == "matrix_cache_single_update"
        )
        assert timing_call.kwargs["predecessor_wait_ms"] == 0.0
        assert timing_call.kwargs["update_run_ms"] == 0.1
        assert timing_call.kwargs["total_ms"] == 0.2

    @pytest.mark.asyncio
    async def test_queue_room_update_logs_full_predecessor_chain_when_enabled(
        self,
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        """Room-scoped cache updates should report the full queued predecessor chain length."""
        monkeypatch.setenv("MINDROOM_TIMING", "1")
        timing_logger = MagicMock()
        monkeypatch.setattr(timing_module, "logger", timing_logger)
        coordinator = EventCacheWriteCoordinator(
            logger=MagicMock(),
            background_task_owner=object(),
        )
        first_started = asyncio.Event()
        allow_first_finish = asyncio.Event()
        second_started = asyncio.Event()
        allow_second_finish = asyncio.Event()

        async def first_update() -> str:
            first_started.set()
            await allow_first_finish.wait()
            return "first"

        async def second_update() -> str:
            second_started.set()
            await allow_second_finish.wait()
            return "second"

        async def third_update() -> str:
            return "third"

        try:
            first_task = coordinator.queue_room_update(
                "!room:localhost",
                first_update,
                name="matrix_cache_first_update",
                coordination_scope="test-principal",
            )
            await asyncio.wait_for(first_started.wait(), timeout=1.0)
            second_task = coordinator.queue_room_update(
                "!room:localhost",
                second_update,
                name="matrix_cache_second_update",
                coordination_scope="test-principal",
            )
            third_task = coordinator.queue_room_update(
                "!room:localhost",
                third_update,
                name="matrix_cache_third_update",
                coordination_scope="test-principal",
            )
            await asyncio.sleep(0)
            allow_first_finish.set()
            await asyncio.wait_for(second_started.wait(), timeout=1.0)
            allow_second_finish.set()
            assert await first_task == "first"
            assert await second_task == "second"
            assert await third_task == "third"
        finally:
            await coordinator.close()

        timing_call = next(
            call
            for call in timing_logger.debug.call_args_list
            if call.args == ("Event cache update timing",) and call.kwargs["operation"] == "matrix_cache_third_update"
        )
        assert timing_call.kwargs["predecessor_count"] == 2
        assert timing_call.kwargs["queued_behind_predecessor"] is True

    @pytest.mark.asyncio
    async def test_queue_room_update_skips_timing_overhead_when_disabled(
        self,
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        """Disabled timing should not touch the perf_counter instrumentation path."""
        monkeypatch.delenv("MINDROOM_TIMING", raising=False)
        monkeypatch.setattr(
            "mindroom.matrix.cache.write_coordinator.time.perf_counter",
            Mock(side_effect=AssertionError("perf_counter should stay unused when timing is disabled")),
        )
        coordinator = EventCacheWriteCoordinator(
            logger=MagicMock(),
            background_task_owner=object(),
        )

        async def update() -> str:
            return "ok"

        try:
            task = coordinator.queue_room_update(
                "!room:localhost",
                update,
                name="matrix_cache_single_update",
                coordination_scope="test-principal",
            )
            assert await task == "ok"
        finally:
            await coordinator.close()

    @pytest.mark.asyncio
    async def test_append_live_event_logs_phase_breakdown_when_enabled(
        self,
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        """Live ingress appends should expose resolver, queue, and cache-write timings."""
        monkeypatch.setenv("MINDROOM_TIMING", "1")
        timing_logger = MagicMock()
        monkeypatch.setattr(timing_module, "logger", timing_logger)
        event_cache = _runtime_event_cache()
        event_cache.apply_thread_mutation_append = AsyncMock(return_value=ThreadAppendOutcome.APPENDED)
        access = MatrixConversationCache(
            logger=MagicMock(),
            runtime=_conversation_runtime(event_cache=event_cache),
        )
        access._live._resolver.resolve_thread_impact_for_mutation = AsyncMock(
            return_value=MutationThreadImpact.threaded("$thread:localhost"),
        )
        event = nio.RoomMessageText.from_dict(
            {
                "type": "m.room.message",
                "room_id": "!room:localhost",
                "event_id": "$reply:localhost",
                "sender": "@user:localhost",
                "origin_server_ts": 1234,
                "content": {
                    "body": "hello",
                    "msgtype": "m.text",
                    "m.relates_to": {"rel_type": "m.thread", "event_id": "$thread:localhost"},
                },
            },
        )

        await access.append_live_event(
            "!room:localhost",
            event,
            event_info=EventInfo.from_event(event.source),
        )

        append_calls = [
            call for call in timing_logger.debug.call_args_list if call.args == ("Live event cache append timing",)
        ]
        assert any(
            call.kwargs["thread_id"] == "$thread:localhost"
            and call.kwargs["event_id"] == "$reply:localhost"
            and call.kwargs["impact_state"] == "threaded"
            and call.kwargs["impact_resolution_ms"] >= 0.0
            and call.kwargs["queue_and_update_ms"] >= 0.0
            and call.kwargs["append_ms"] >= 0.0
            and call.kwargs["outcome"] == "ok"
            for call in append_calls
        )

    @pytest.mark.asyncio
    async def test_append_live_event_logs_append_failure_outcome_when_enabled(
        self,
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        """Live ingress timing should classify append misses as append failures."""
        monkeypatch.setenv("MINDROOM_TIMING", "1")
        timing_logger = MagicMock()
        monkeypatch.setattr(timing_module, "logger", timing_logger)
        event_cache = _runtime_event_cache()
        event_cache.apply_thread_mutation_append = AsyncMock(return_value=ThreadAppendOutcome.SNAPSHOT_MISSING)
        access = MatrixConversationCache(
            logger=MagicMock(),
            runtime=_conversation_runtime(event_cache=event_cache),
        )
        access._live._resolver.resolve_thread_impact_for_mutation = AsyncMock(
            return_value=MutationThreadImpact.threaded("$thread:localhost"),
        )
        event = nio.RoomMessageText.from_dict(
            {
                "type": "m.room.message",
                "room_id": "!room:localhost",
                "event_id": "$reply:localhost",
                "sender": "@user:localhost",
                "origin_server_ts": 1234,
                "content": {
                    "body": "hello",
                    "msgtype": "m.text",
                    "m.relates_to": {"rel_type": "m.thread", "event_id": "$thread:localhost"},
                },
            },
        )

        await access.append_live_event(
            "!room:localhost",
            event,
            event_info=EventInfo.from_event(event.source),
        )

        timing_call = next(
            call for call in timing_logger.debug.call_args_list if call.args == ("Live event cache append timing",)
        )
        assert timing_call.kwargs["thread_id"] == "$thread:localhost"
        assert timing_call.kwargs["appended"] is False
        assert timing_call.kwargs["outcome"] == "append_failed"

    @pytest.mark.asyncio
    async def test_append_live_event_skips_timing_overhead_when_disabled(
        self,
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        """Disabled timing should not touch the live-append perf_counter instrumentation path."""
        monkeypatch.delenv("MINDROOM_TIMING", raising=False)
        monkeypatch.setattr(
            "mindroom.matrix.cache.thread_writes.time.perf_counter",
            Mock(side_effect=AssertionError("perf_counter should stay unused when timing is disabled")),
        )
        cache_ops, _logger, event_cache = _thread_mutation_cache_ops()

        class _InlineCoordinator:
            def queue_room_update(
                self,
                room_id: str,
                update_coro_factory: Callable[[], Coroutine[Any, Any, object]],
                *,
                name: str,
                emit_timing: bool = False,
                coalesce_key: tuple[str, str] | None = None,
                coalesce_log_context: dict[str, object] | None = None,
                coordination_scope: str | None = None,
            ) -> asyncio.Task[object]:
                del room_id, name, emit_timing, coalesce_key, coalesce_log_context, coordination_scope
                return asyncio.create_task(update_coro_factory())

            def queue_thread_update(
                self,
                room_id: str,
                thread_id: str,
                update_coro_factory: Callable[[], Coroutine[Any, Any, object]],
                *,
                name: str,
                emit_timing: bool = False,
                coalesce_key: tuple[str, str] | None = None,
                coalesce_log_context: dict[str, object] | None = None,
                coordination_scope: str | None = None,
            ) -> asyncio.Task[object]:
                del thread_id
                return self.queue_room_update(
                    room_id,
                    update_coro_factory,
                    name=name,
                    emit_timing=emit_timing,
                    coalesce_key=coalesce_key,
                    coalesce_log_context=coalesce_log_context,
                    coordination_scope=coordination_scope,
                )

        cache_ops.runtime.event_cache_write_coordinator = _InlineCoordinator()
        resolver = MagicMock()
        resolver.resolve_thread_impact_for_mutation = AsyncMock(
            return_value=MutationThreadImpact.threaded("$thread:localhost"),
        )
        policy = thread_writes.ThreadLiveWritePolicy(
            resolver=resolver,
            cache_ops=cache_ops,
        )
        event = nio.RoomMessageText.from_dict(
            {
                "type": "m.room.message",
                "room_id": "!room:localhost",
                "event_id": "$reply:localhost",
                "sender": "@user:localhost",
                "origin_server_ts": 1234,
                "content": {
                    "body": "hello",
                    "msgtype": "m.text",
                    "m.relates_to": {"rel_type": "m.thread", "event_id": "$thread:localhost"},
                },
            },
        )

        await policy.append_live_event(
            "!room:localhost",
            event,
            event_info=EventInfo.from_event(event.source),
        )

        event_cache.apply_thread_mutation_append.assert_awaited_once()
        assert event_cache.apply_thread_mutation_append.await_args.kwargs["append_failed_reason"] == (
            "live_append_failed"
        )
        event_cache.mark_thread_gap.assert_not_awaited()

    @pytest.mark.asyncio
    async def test_live_event_cache_update_recovers_after_same_room_failure(self) -> None:
        """A failed same-room cache update should not block the next queued write."""
        first_update_started = asyncio.Event()
        allow_first_failure = asyncio.Event()
        second_update_finished = asyncio.Event()
        owner = object()
        coordinator = EventCacheWriteCoordinator(
            logger=MagicMock(),
            background_task_owner=owner,
        )
        access = MatrixConversationCache(
            logger=MagicMock(),
            runtime=_conversation_runtime(coordinator=coordinator),
        )

        async def failing_update() -> None:
            first_update_started.set()
            await allow_first_failure.wait()
            msg = "update failed"
            raise RuntimeError(msg)

        async def second_update() -> None:
            second_update_finished.set()

        access.runtime.event_cache_write_coordinator.queue_room_update(
            "!test:localhost",
            lambda: failing_update(),
            name="matrix_cache_first_update",
            coordination_scope="test-principal",
        )
        await asyncio.wait_for(first_update_started.wait(), timeout=1.0)

        access.runtime.event_cache_write_coordinator.queue_room_update(
            "!test:localhost",
            lambda: second_update(),
            name="matrix_cache_second_update",
            coordination_scope="test-principal",
        )
        await asyncio.sleep(0)
        assert second_update_finished.is_set() is False

        allow_first_failure.set()
        await wait_for_background_tasks(timeout=1.0, owner=owner)

        assert second_update_finished.is_set()

    @pytest.mark.asyncio
    async def test_shared_event_cache_write_coordinator_serializes_same_room_updates_across_accesses(self) -> None:
        """Same-room cache writes should serialize even when different bots enqueue them."""
        first_update_started = asyncio.Event()
        release_first_update = asyncio.Event()
        second_update_started = asyncio.Event()
        owner = object()
        coordinator = EventCacheWriteCoordinator(
            logger=MagicMock(),
            background_task_owner=owner,
        )
        first_access = MatrixConversationCache(
            logger=MagicMock(),
            runtime=_conversation_runtime(coordinator=coordinator),
        )
        second_access = MatrixConversationCache(
            logger=MagicMock(),
            runtime=_conversation_runtime(coordinator=coordinator),
        )

        async def first_update() -> None:
            first_update_started.set()
            await release_first_update.wait()

        async def second_update() -> None:
            second_update_started.set()

        first_access.runtime.event_cache_write_coordinator.queue_room_update(
            "!test:localhost",
            lambda: first_update(),
            name="matrix_cache_first_update",
            coordination_scope="test-principal",
        )
        await asyncio.wait_for(first_update_started.wait(), timeout=1.0)

        second_access.runtime.event_cache_write_coordinator.queue_room_update(
            "!test:localhost",
            lambda: second_update(),
            name="matrix_cache_second_update",
            coordination_scope="test-principal",
        )
        await asyncio.sleep(0)
        assert second_update_started.is_set() is False

        release_first_update.set()
        await wait_for_background_tasks(timeout=1.0, owner=owner)

        assert second_update_started.is_set()

    @pytest.mark.parametrize("principal_count", [10, 100])
    @pytest.mark.asyncio
    async def test_shared_event_cache_write_coordinator_parallelizes_distinct_principals(
        self,
        principal_count: int,
    ) -> None:
        """Principal-isolated caches should not share a same-room serialization lane."""
        started = [asyncio.Event() for _ in range(principal_count)]
        release_updates = asyncio.Event()
        coordinator = EventCacheWriteCoordinator(
            logger=MagicMock(),
            background_task_owner=object(),
        )

        async def update(index: int) -> None:
            started[index].set()
            await release_updates.wait()

        tasks = [
            coordinator.queue_room_update(
                "!test:localhost",
                lambda index=index: update(index),
                name=f"matrix_cache_principal_update_{index}",
                coordination_scope=f"@agent-{index}:localhost",
            )
            for index in range(principal_count)
        ]
        try:
            await asyncio.wait_for(
                asyncio.gather(*(event.wait() for event in started)),
                timeout=1.0,
            )
        finally:
            release_updates.set()
            await asyncio.gather(*tasks)

    @pytest.mark.asyncio
    async def test_coordination_keys_do_not_alias_on_component_delimiters(self) -> None:
        """Structured scope and room keys should remain distinct for arbitrary strings."""
        started = [asyncio.Event(), asyncio.Event()]
        release_updates = asyncio.Event()
        coordinator = EventCacheWriteCoordinator(
            logger=MagicMock(),
            background_task_owner=object(),
        )

        async def update(index: int) -> None:
            started[index].set()
            await release_updates.wait()

        tasks = [
            coordinator.queue_room_update(
                "room",
                lambda: update(0),
                name="matrix_cache_delimited_scope",
                coordination_scope="scope\x1fother",
            ),
            coordinator.queue_room_update(
                "other\x1froom",
                lambda: update(1),
                name="matrix_cache_delimited_room",
                coordination_scope="scope",
            ),
        ]
        try:
            await asyncio.wait_for(
                asyncio.gather(*(event.wait() for event in started)),
                timeout=1.0,
            )
        finally:
            release_updates.set()
            await asyncio.gather(*tasks)

    @pytest.mark.asyncio
    async def test_shared_event_cache_write_coordinator_allows_other_thread_updates_while_one_thread_runs(
        self,
    ) -> None:
        """Same-room thread updates should not serialize across unrelated threads."""
        first_update_started = asyncio.Event()
        release_first_update = asyncio.Event()
        second_update_started = asyncio.Event()
        owner = object()
        coordinator = EventCacheWriteCoordinator(
            logger=MagicMock(),
            background_task_owner=owner,
        )

        async def first_update() -> None:
            first_update_started.set()
            await release_first_update.wait()

        async def second_update() -> None:
            second_update_started.set()

        coordinator.queue_thread_update(
            "!test:localhost",
            "$thread-a:localhost",
            lambda: first_update(),
            name="matrix_cache_first_thread_update",
            coordination_scope="test-principal",
        )
        await asyncio.wait_for(first_update_started.wait(), timeout=1.0)

        coordinator.queue_thread_update(
            "!test:localhost",
            "$thread-b:localhost",
            lambda: second_update(),
            name="matrix_cache_second_thread_update",
            coordination_scope="test-principal",
        )
        await asyncio.sleep(0)
        assert second_update_started.is_set()

        release_first_update.set()
        await wait_for_background_tasks(timeout=1.0, owner=owner)

    @pytest.mark.asyncio
    async def test_shared_event_cache_write_coordinator_keeps_pending_room_barrier_across_blocked_threads(  # noqa: PLR0915
        self,
    ) -> None:
        """A queued room update should keep later unrelated threads blocked until the room segment clears."""
        first_thread_started = asyncio.Event()
        release_first_thread = asyncio.Event()
        room_update_started = asyncio.Event()
        release_room_update = asyncio.Event()
        second_thread_started = asyncio.Event()
        release_second_thread = asyncio.Event()
        sibling_thread_started = asyncio.Event()
        release_sibling_thread = asyncio.Event()
        owner = object()
        coordinator = EventCacheWriteCoordinator(
            logger=MagicMock(),
            background_task_owner=owner,
        )

        async def first_thread_update() -> None:
            first_thread_started.set()
            await release_first_thread.wait()

        async def room_update() -> None:
            room_update_started.set()
            await release_room_update.wait()

        async def second_thread_update() -> None:
            second_thread_started.set()
            await release_second_thread.wait()

        async def sibling_thread_update() -> None:
            sibling_thread_started.set()
            await release_sibling_thread.wait()

        first_thread_task = coordinator.queue_thread_update(
            "!test:localhost",
            "$thread-a:localhost",
            first_thread_update,
            name="matrix_cache_first_thread_update",
            coordination_scope="test-principal",
        )
        await asyncio.wait_for(first_thread_started.wait(), timeout=1.0)

        room_task = coordinator.queue_room_update(
            "!test:localhost",
            room_update,
            name="matrix_cache_room_update",
            coordination_scope="test-principal",
        )
        second_thread_task = coordinator.queue_thread_update(
            "!test:localhost",
            "$thread-a:localhost",
            second_thread_update,
            name="matrix_cache_second_thread_update",
            coordination_scope="test-principal",
        )
        sibling_thread_task = coordinator.queue_thread_update(
            "!test:localhost",
            "$thread-b:localhost",
            sibling_thread_update,
            name="matrix_cache_sibling_thread_update",
            coordination_scope="test-principal",
        )
        try:
            await asyncio.sleep(0.05)
            assert room_update_started.is_set() is False
            assert second_thread_started.is_set() is False
            assert sibling_thread_started.is_set() is False

            release_first_thread.set()
            await asyncio.wait_for(first_thread_task, timeout=1.0)
            await asyncio.wait_for(room_update_started.wait(), timeout=1.0)

            await asyncio.sleep(0.05)
            assert second_thread_started.is_set() is False
            assert sibling_thread_started.is_set() is False

            release_room_update.set()
            await asyncio.wait_for(room_task, timeout=1.0)
            await asyncio.wait_for(second_thread_started.wait(), timeout=1.0)
            await asyncio.wait_for(sibling_thread_started.wait(), timeout=1.0)

            release_second_thread.set()
            release_sibling_thread.set()
            await asyncio.wait_for(
                asyncio.gather(
                    second_thread_task,
                    sibling_thread_task,
                ),
                timeout=1.0,
            )
        finally:
            release_first_thread.set()
            release_room_update.set()
            release_second_thread.set()
            release_sibling_thread.set()
            await asyncio.wait_for(
                asyncio.gather(
                    first_thread_task,
                    room_task,
                    second_thread_task,
                    sibling_thread_task,
                    return_exceptions=True,
                ),
                timeout=1.0,
            )

    @pytest.mark.asyncio
    async def test_get_thread_history_does_not_wait_for_other_thread_update(self) -> None:
        """Thread reads should not stall behind unrelated thread updates in the same room."""
        access = MatrixConversationCache(
            logger=MagicMock(),
            runtime=_conversation_runtime(),
        )
        other_thread_update_started = asyncio.Event()
        release_other_thread_update = asyncio.Event()
        fetch_started = asyncio.Event()

        async def blocking_other_thread_update() -> None:
            other_thread_update_started.set()
            await release_other_thread_update.wait()

        async def fetch_history(
            _room_id: str,
            _thread_id: str,
            *,
            caller_label: str,
            coordinator_queue_wait_ms: float,
        ) -> ThreadHistoryResult:
            assert caller_label == "unknown"
            assert coordinator_queue_wait_ms >= 0.0
            fetch_started.set()
            return thread_history_result(
                [_message(event_id="$thread-a:localhost", body="Root")],
                is_full_history=True,
            )

        access._reads.fetch_thread_history_from_client = AsyncMock(side_effect=fetch_history)
        access.runtime.event_cache_write_coordinator.queue_thread_update(
            "!test:localhost",
            "$thread-b:localhost",
            lambda: blocking_other_thread_update(),
            name="matrix_cache_blocking_other_thread_update",
            coordination_scope="test-principal",
        )
        await asyncio.wait_for(other_thread_update_started.wait(), timeout=1.0)

        history = await asyncio.wait_for(
            access.get_thread_history("!test:localhost", "$thread-a:localhost"),
            timeout=1.0,
        )

        assert fetch_started.is_set()
        assert [message.body for message in history] == ["Root"]
        release_other_thread_update.set()
        await _wait_for_room_cache_idle(access.runtime.event_cache_write_coordinator)

    @pytest.mark.asyncio
    async def test_point_read_barrier_does_not_wait_for_later_room_update(self) -> None:
        """A point read must wait for predecessors without being extended by later writes."""
        first_started = asyncio.Event()
        release_first = asyncio.Event()
        second_started = asyncio.Event()
        release_second = asyncio.Event()
        coordinator = _runtime_write_coordinator()

        async def first_update() -> None:
            first_started.set()
            await release_first.wait()

        async def second_update() -> None:
            second_started.set()
            await release_second.wait()

        first_task = coordinator.queue_room_update(
            "!test:localhost",
            first_update,
            name="first_room_update",
            coordination_scope="test-principal",
        )
        await first_started.wait()
        read_barrier = asyncio.create_task(
            coordinator.wait_for_prior_room_updates(
                "!test:localhost",
                coordination_scope="test-principal",
            ),
        )
        await asyncio.sleep(0)
        second_task = coordinator.queue_room_update(
            "!test:localhost",
            second_update,
            name="later_room_update",
            coordination_scope="test-principal",
        )

        try:
            release_first.set()
            await second_started.wait()
            await asyncio.wait_for(read_barrier, timeout=1.0)
            assert second_task.done() is False
        finally:
            release_first.set()
            release_second.set()
            await asyncio.gather(first_task, second_task, return_exceptions=True)
            await coordinator.close()

    @pytest.mark.asyncio
    async def test_point_read_barrier_waits_for_queued_predecessor(self) -> None:
        """The point-read fence includes prior thread writes but excludes later room writes."""
        first_started = asyncio.Event()
        release_first = asyncio.Event()
        queued_predecessor_started = asyncio.Event()
        release_queued_predecessor = asyncio.Event()
        later_started = asyncio.Event()
        release_later = asyncio.Event()
        coordinator = _runtime_write_coordinator()

        async def first_update() -> None:
            first_started.set()
            await release_first.wait()

        async def queued_predecessor() -> None:
            queued_predecessor_started.set()
            await release_queued_predecessor.wait()

        async def later_update() -> None:
            later_started.set()
            await release_later.wait()

        first_task = coordinator.queue_room_update(
            "!test:localhost",
            first_update,
            name="first_room_update",
            coordination_scope="test-principal",
        )
        await first_started.wait()
        queued_predecessor_task = coordinator.queue_thread_update(
            "!test:localhost",
            "$thread:localhost",
            queued_predecessor,
            name="queued_predecessor",
            coordination_scope="test-principal",
        )
        read_barrier = asyncio.create_task(
            coordinator.wait_for_prior_room_updates(
                "!test:localhost",
                coordination_scope="test-principal",
            ),
        )
        await asyncio.sleep(0)
        later_task = coordinator.queue_room_update(
            "!test:localhost",
            later_update,
            name="later_room_update",
            coordination_scope="test-principal",
        )

        try:
            release_first.set()
            await queued_predecessor_started.wait()
            await asyncio.sleep(0)
            assert read_barrier.done() is False

            release_queued_predecessor.set()
            await later_started.wait()
            await asyncio.wait_for(read_barrier, timeout=1.0)
            assert later_task.done() is False
        finally:
            release_first.set()
            release_queued_predecessor.set()
            release_later.set()
            await asyncio.gather(
                first_task,
                queued_predecessor_task,
                later_task,
                return_exceptions=True,
            )
            await coordinator.close()

    @pytest.mark.asyncio
    async def test_wait_for_thread_idle_ignores_cancelled_room_fence_for_unrelated_thread(self) -> None:
        """Thread reads should ignore cancelled room fences that only preserve write ordering."""
        first_thread_started = asyncio.Event()
        release_first_thread = asyncio.Event()
        second_thread_started = asyncio.Event()
        release_second_thread = asyncio.Event()
        coordinator = _runtime_write_coordinator()

        async def first_thread_update() -> None:
            first_thread_started.set()
            await release_first_thread.wait()

        async def second_thread_update() -> None:
            second_thread_started.set()
            await release_second_thread.wait()

        async def cancelled_room_update() -> None:
            msg = "Cancelled room cache update should not start"
            raise AssertionError(msg)

        first_thread_task = coordinator.queue_thread_update(
            "!test:localhost",
            "$thread-a:localhost",
            first_thread_update,
            name="matrix_cache_first_thread_update",
            coordination_scope="test-principal",
        )
        second_thread_task = coordinator.queue_thread_update(
            "!test:localhost",
            "$thread-b:localhost",
            second_thread_update,
            name="matrix_cache_second_thread_update",
            coordination_scope="test-principal",
        )
        await asyncio.wait_for(first_thread_started.wait(), timeout=1.0)
        await asyncio.wait_for(second_thread_started.wait(), timeout=1.0)

        cancelled_room_task = coordinator.queue_room_update(
            "!test:localhost",
            cancelled_room_update,
            name="matrix_cache_cancelled_room_update",
            coordination_scope="test-principal",
        )
        try:
            cancelled_room_task.cancel()
            await asyncio.gather(cancelled_room_task, return_exceptions=True)

            await asyncio.wait_for(
                coordinator.wait_for_thread_idle(
                    "!test:localhost",
                    "$thread-c:localhost",
                    coordination_scope="test-principal",
                    ignore_cancelled_room_fences=True,
                ),
                timeout=0.1,
            )
            assert first_thread_task.done() is False
            assert second_thread_task.done() is False
        finally:
            release_first_thread.set()
            release_second_thread.set()
            if not cancelled_room_task.done():
                cancelled_room_task.cancel()
            await asyncio.wait_for(
                asyncio.gather(
                    first_thread_task,
                    second_thread_task,
                    cancelled_room_task,
                    return_exceptions=True,
                ),
                timeout=1.0,
            )

    @pytest.mark.asyncio
    async def test_get_thread_history_ignores_cancelled_room_fence_for_unrelated_thread(self) -> None:
        """Public thread-history reads should bypass cancelled room fences without waiting for other threads."""
        first_thread_started = asyncio.Event()
        release_first_thread = asyncio.Event()
        second_thread_started = asyncio.Event()
        release_second_thread = asyncio.Event()
        coordinator = _runtime_write_coordinator()
        access = MatrixConversationCache(
            logger=MagicMock(),
            runtime=_conversation_runtime(coordinator=coordinator),
        )

        async def first_thread_update() -> None:
            first_thread_started.set()
            await release_first_thread.wait()

        async def second_thread_update() -> None:
            second_thread_started.set()
            await release_second_thread.wait()

        async def cancelled_room_update() -> None:
            msg = "Cancelled room cache update should not start"
            raise AssertionError(msg)

        access._reads.fetch_thread_history_from_client = AsyncMock(
            return_value=thread_history_result(
                [_message(event_id="$thread-c:localhost", body="Root")],
                is_full_history=True,
            ),
        )
        first_thread_task = coordinator.queue_thread_update(
            "!test:localhost",
            "$thread-a:localhost",
            first_thread_update,
            name="matrix_cache_first_thread_update",
            coordination_scope="test-principal",
        )
        second_thread_task = coordinator.queue_thread_update(
            "!test:localhost",
            "$thread-b:localhost",
            second_thread_update,
            name="matrix_cache_second_thread_update",
            coordination_scope="test-principal",
        )
        await asyncio.wait_for(first_thread_started.wait(), timeout=1.0)
        await asyncio.wait_for(second_thread_started.wait(), timeout=1.0)

        cancelled_room_task = coordinator.queue_room_update(
            "!test:localhost",
            cancelled_room_update,
            name="matrix_cache_cancelled_room_update",
            coordination_scope="test-principal",
        )
        try:
            cancelled_room_task.cancel()
            await asyncio.gather(cancelled_room_task, return_exceptions=True)

            history = await asyncio.wait_for(
                access.get_thread_history("!test:localhost", "$thread-c:localhost"),
                timeout=0.1,
            )

            assert [message.body for message in history] == ["Root"]
            access._reads.fetch_thread_history_from_client.assert_awaited_once()
            fetch_args = access._reads.fetch_thread_history_from_client.await_args
            assert fetch_args is not None
            assert fetch_args.args == ("!test:localhost", "$thread-c:localhost")
            assert fetch_args.kwargs["caller_label"] == "unknown"
            assert fetch_args.kwargs["coordinator_queue_wait_ms"] >= 0.0
            assert first_thread_task.done() is False
            assert second_thread_task.done() is False
        finally:
            release_first_thread.set()
            release_second_thread.set()
            if not cancelled_room_task.done():
                cancelled_room_task.cancel()
            await asyncio.wait_for(
                asyncio.gather(
                    first_thread_task,
                    second_thread_task,
                    cancelled_room_task,
                    return_exceptions=True,
                ),
                timeout=1.0,
            )

    @pytest.mark.asyncio
    async def test_cancelled_room_cache_update_does_not_start_queued_coro(self) -> None:
        """Cancelling a queued room update before it runs should not invoke its coroutine factory."""
        blocker_started = asyncio.Event()
        release_blocker = asyncio.Event()
        queued_update_started = asyncio.Event()
        owner = object()
        coordinator = EventCacheWriteCoordinator(
            logger=MagicMock(),
            background_task_owner=owner,
        )
        access = MatrixConversationCache(
            logger=MagicMock(),
            runtime=_conversation_runtime(coordinator=coordinator),
        )

        async def blocking_update() -> None:
            blocker_started.set()
            await release_blocker.wait()

        async def queued_update() -> None:
            queued_update_started.set()

        access.runtime.event_cache_write_coordinator.queue_room_update(
            "!test:localhost",
            lambda: blocking_update(),
            name="matrix_cache_blocking_update",
            coordination_scope="test-principal",
        )
        await asyncio.wait_for(blocker_started.wait(), timeout=1.0)

        queued_task = access.runtime.event_cache_write_coordinator.queue_room_update(
            "!test:localhost",
            lambda: queued_update(),
            name="matrix_cache_queued_update",
            coordination_scope="test-principal",
        )
        queued_task.cancel()
        await asyncio.gather(queued_task, return_exceptions=True)

        release_blocker.set()
        await wait_for_background_tasks(timeout=1.0, owner=owner)

        assert queued_update_started.is_set() is False

    @pytest.mark.asyncio
    async def test_cancelled_room_cache_update_keeps_follow_up_update_behind_running_predecessor(self) -> None:
        """Cancelling a queued room update must not break the same-room serialization chain."""
        first_update_started = asyncio.Event()
        release_first_update = asyncio.Event()
        third_update_started = asyncio.Event()
        owner = object()
        coordinator = EventCacheWriteCoordinator(
            logger=MagicMock(),
            background_task_owner=owner,
        )
        access = MatrixConversationCache(
            logger=MagicMock(),
            runtime=_conversation_runtime(coordinator=coordinator),
        )

        async def first_update() -> None:
            first_update_started.set()
            await release_first_update.wait()

        async def cancelled_update() -> None:
            msg = "Cancelled room cache update should not start"
            raise AssertionError(msg)

        async def third_update() -> None:
            third_update_started.set()

        access.runtime.event_cache_write_coordinator.queue_room_update(
            "!test:localhost",
            lambda: first_update(),
            name="matrix_cache_first_update",
            coordination_scope="test-principal",
        )
        await asyncio.wait_for(first_update_started.wait(), timeout=1.0)

        cancelled_task = access.runtime.event_cache_write_coordinator.queue_room_update(
            "!test:localhost",
            lambda: cancelled_update(),
            name="matrix_cache_cancelled_update",
            coordination_scope="test-principal",
        )
        cancelled_task.cancel()
        await asyncio.gather(cancelled_task, return_exceptions=True)

        access.runtime.event_cache_write_coordinator.queue_room_update(
            "!test:localhost",
            lambda: third_update(),
            name="matrix_cache_third_update",
            coordination_scope="test-principal",
        )
        await asyncio.sleep(0)
        assert third_update_started.is_set() is False

        release_first_update.set()
        await wait_for_background_tasks(timeout=1.0, owner=owner)

        assert third_update_started.is_set()

    @pytest.mark.asyncio
    async def test_cancelled_room_cache_update_keeps_follow_up_thread_update_behind_all_predecessors(  # noqa: PLR0915
        self,
    ) -> None:
        """Cancelling a room update must not let a later thread update skip unfinished room predecessors."""
        first_thread_started = asyncio.Event()
        release_first_thread = asyncio.Event()
        second_thread_started = asyncio.Event()
        release_second_thread = asyncio.Event()
        follow_up_thread_started = asyncio.Event()
        release_follow_up_thread = asyncio.Event()
        owner = object()
        coordinator = EventCacheWriteCoordinator(
            logger=MagicMock(),
            background_task_owner=owner,
        )

        async def first_thread_update() -> None:
            first_thread_started.set()
            await release_first_thread.wait()

        async def second_thread_update() -> None:
            second_thread_started.set()
            await release_second_thread.wait()

        async def cancelled_room_update() -> None:
            msg = "Cancelled room cache update should not start"
            raise AssertionError(msg)

        async def follow_up_thread_update() -> None:
            follow_up_thread_started.set()
            await release_follow_up_thread.wait()

        first_thread_task = coordinator.queue_thread_update(
            "!test:localhost",
            "$thread-a:localhost",
            first_thread_update,
            name="matrix_cache_first_thread_update",
            coordination_scope="test-principal",
        )
        second_thread_task = coordinator.queue_thread_update(
            "!test:localhost",
            "$thread-b:localhost",
            second_thread_update,
            name="matrix_cache_second_thread_update",
            coordination_scope="test-principal",
        )
        await asyncio.wait_for(first_thread_started.wait(), timeout=1.0)
        await asyncio.wait_for(second_thread_started.wait(), timeout=1.0)

        cancelled_room_task = coordinator.queue_room_update(
            "!test:localhost",
            cancelled_room_update,
            name="matrix_cache_cancelled_room_update",
            coordination_scope="test-principal",
        )
        follow_up_thread_task = coordinator.queue_thread_update(
            "!test:localhost",
            "$thread-c:localhost",
            follow_up_thread_update,
            name="matrix_cache_follow_up_thread_update",
            coordination_scope="test-principal",
        )
        try:
            cancelled_room_task.cancel()
            await asyncio.gather(cancelled_room_task, return_exceptions=True)

            await asyncio.sleep(0.05)
            assert follow_up_thread_started.is_set() is False
            assert follow_up_thread_task.done() is False

            release_first_thread.set()
            await asyncio.wait_for(first_thread_task, timeout=1.0)

            await asyncio.sleep(0.05)
            assert follow_up_thread_started.is_set() is False
            assert follow_up_thread_task.done() is False

            release_second_thread.set()
            await asyncio.wait_for(second_thread_task, timeout=1.0)
            await asyncio.wait_for(follow_up_thread_started.wait(), timeout=1.0)
            assert follow_up_thread_task.done() is False

            release_follow_up_thread.set()
            await asyncio.wait_for(follow_up_thread_task, timeout=1.0)
        finally:
            release_first_thread.set()
            release_second_thread.set()
            release_follow_up_thread.set()
            if not cancelled_room_task.done():
                cancelled_room_task.cancel()
            pending_tasks = [first_thread_task, second_thread_task, cancelled_room_task]
            if follow_up_thread_task is not None:
                pending_tasks.append(follow_up_thread_task)
            await asyncio.wait_for(
                asyncio.gather(
                    *pending_tasks,
                    return_exceptions=True,
                ),
                timeout=1.0,
            )

    @pytest.mark.asyncio
    async def test_cancelled_room_cache_update_still_blocks_later_thread_updates_queued_after_cancel(  # noqa: PLR0915
        self,
    ) -> None:
        """Cancelling a queued room update must not let later thread work overtake the earlier room segment."""
        first_thread_started = asyncio.Event()
        release_first_thread = asyncio.Event()
        second_thread_started = asyncio.Event()
        release_second_thread = asyncio.Event()
        follow_up_thread_started = asyncio.Event()
        release_follow_up_thread = asyncio.Event()
        coordinator = _runtime_write_coordinator()

        async def first_thread_update() -> None:
            first_thread_started.set()
            await release_first_thread.wait()

        async def second_thread_update() -> None:
            second_thread_started.set()
            await release_second_thread.wait()

        async def cancelled_room_update() -> None:
            msg = "Cancelled room cache update should not start"
            raise AssertionError(msg)

        async def follow_up_thread_update() -> None:
            follow_up_thread_started.set()
            await release_follow_up_thread.wait()

        first_thread_task = coordinator.queue_thread_update(
            "!test:localhost",
            "$thread-a:localhost",
            first_thread_update,
            name="matrix_cache_first_thread_update",
            coordination_scope="test-principal",
        )
        second_thread_task = coordinator.queue_thread_update(
            "!test:localhost",
            "$thread-b:localhost",
            second_thread_update,
            name="matrix_cache_second_thread_update",
            coordination_scope="test-principal",
        )
        await asyncio.wait_for(first_thread_started.wait(), timeout=1.0)
        await asyncio.wait_for(second_thread_started.wait(), timeout=1.0)

        cancelled_room_task = coordinator.queue_room_update(
            "!test:localhost",
            cancelled_room_update,
            name="matrix_cache_cancelled_room_update",
            coordination_scope="test-principal",
        )
        follow_up_thread_task: asyncio.Task[object] | None = None
        try:
            cancelled_room_task.cancel()
            await asyncio.gather(cancelled_room_task, return_exceptions=True)

            follow_up_thread_task = coordinator.queue_thread_update(
                "!test:localhost",
                "$thread-c:localhost",
                follow_up_thread_update,
                name="matrix_cache_follow_up_thread_update",
                coordination_scope="test-principal",
            )

            await asyncio.sleep(0.05)
            assert follow_up_thread_started.is_set() is False
            assert follow_up_thread_task.done() is False

            release_first_thread.set()
            await asyncio.wait_for(first_thread_task, timeout=1.0)

            await asyncio.sleep(0.05)
            assert follow_up_thread_started.is_set() is False
            assert follow_up_thread_task.done() is False

            release_second_thread.set()
            await asyncio.wait_for(second_thread_task, timeout=1.0)
            await asyncio.wait_for(follow_up_thread_started.wait(), timeout=1.0)

            release_follow_up_thread.set()
            await asyncio.wait_for(follow_up_thread_task, timeout=1.0)
        finally:
            release_first_thread.set()
            release_second_thread.set()
            release_follow_up_thread.set()
            if not cancelled_room_task.done():
                cancelled_room_task.cancel()
            await asyncio.wait_for(
                asyncio.gather(
                    first_thread_task,
                    second_thread_task,
                    cancelled_room_task,
                    follow_up_thread_task,
                    return_exceptions=True,
                ),
                timeout=1.0,
            )
