"""Tests for the SQLite-backed Matrix thread event cache."""

from __future__ import annotations

import asyncio
import json
import sqlite3
from contextlib import closing
from dataclasses import replace
from typing import TYPE_CHECKING
from unittest.mock import ANY, AsyncMock, MagicMock, patch

import nio
import pytest
from nio.api import RelationshipType

import mindroom.matrix.cache.sqlite_event_cache as event_cache_module
from mindroom.bot_runtime_view import BotRuntimeState
from mindroom.config.agent import AgentConfig
from mindroom.config.main import Config
from mindroom.config.models import ModelConfig
from mindroom.conversation_resolver import ConversationResolver, ConversationResolverDeps, _ThreadIdLookup
from mindroom.matrix.cache import (
    ConversationEventCache,
    ThreadAppendOutcome,
    ThreadCacheGap,
    event_normalization,
    sqlite_event_cache_events,
    sqlite_event_cache_threads,
    thread_cache_rejection_reason,
)
from mindroom.matrix.cache.event_batching import group_lookup_events_by_room
from mindroom.matrix.cache.sqlite_event_cache import SqliteEventCache
from mindroom.matrix.cache.thread_history_result import thread_history_result
from mindroom.matrix.cache.thread_reads import ThreadReadMode
from mindroom.matrix.cache.write_coordinator import EventCacheWriteCoordinator
from mindroom.matrix.client_thread_history import (
    BulkThreadRefreshStats,
    fetch_thread_history,
)
from mindroom.matrix.client_visible_messages import ResolvedVisibleMessage
from mindroom.matrix.conversation_cache import MatrixConversationCache, _cached_room_get_event
from mindroom.matrix.event_info import EventInfo
from mindroom.matrix.thread_diagnostics import (
    THREAD_HISTORY_DEGRADED_DIAGNOSTIC,
)
from mindroom.timing import DispatchPipelineTiming
from tests.conftest import (
    agent_response_should_respond,
    bind_runtime_paths,
    create_mock_room,
    runtime_paths_for,
    test_runtime_paths,
)
from tests.event_cache_test_support import replace_thread_unconditionally as _replace_thread
from tests.identity_helpers import entity_ids

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

    from mindroom.matrix.cache import ThreadHistoryResult


def _conversation_cache_for_thread_reads(
    tmp_path: Path,
    event_cache: ConversationEventCache,
    *,
    client: object,
) -> MatrixConversationCache:
    runtime_paths = test_runtime_paths(tmp_path)
    config = bind_runtime_paths(
        Config(
            agents={"code": AgentConfig(display_name="Code", rooms=["!room:localhost"])},
            models={"default": ModelConfig(provider="test", id="test-model")},
        ),
        runtime_paths,
    )
    runtime = BotRuntimeState(
        client=client,
        config=config,
        runtime_paths=runtime_paths,
        enable_streaming=False,
        orchestrator=None,
        event_cache=event_cache,
        event_cache_write_coordinator=None,
    )
    return MatrixConversationCache(logger=MagicMock(), runtime=runtime)


def _set_dispatch_thread_read_timeout(conversation_cache: MatrixConversationCache, seconds: float) -> None:
    runtime_paths = conversation_cache.runtime.runtime_paths
    conversation_cache.runtime.runtime_paths = replace(
        runtime_paths,
        process_env={
            **runtime_paths.process_env,
            "MINDROOM_DISPATCH_THREAD_READ_TIMEOUT_SECONDS": str(seconds),
        },
    )


def _pending_thread_cache_update_wait_tasks() -> set[asyncio.Task[object]]:
    return {
        task
        for task in asyncio.all_tasks()
        if not task.done()
        and task.get_coro().__qualname__.endswith("ThreadReadPolicy._wait_for_pending_thread_cache_updates")
    }


def test_sqlite_event_cache_is_explicit_concrete_cache(tmp_path: Path) -> None:
    """The SQLite cache implementation should be named at the boundary."""
    cache = SqliteEventCache(tmp_path / "event_cache.db")

    assert cache.db_path == tmp_path / "event_cache.db"


def _make_text_event(
    *,
    event_id: str,
    sender: str,
    body: str,
    server_timestamp: int,
    source_content: dict[str, object],
) -> MagicMock:
    event = MagicMock(spec=nio.RoomMessageText)
    event.event_id = event_id
    event.sender = sender
    event.body = body
    event.server_timestamp = server_timestamp
    normalized_content = dict(source_content)
    normalized_content.setdefault("msgtype", "m.text")
    event.source = {
        "type": "m.room.message",
        "content": normalized_content,
    }
    return event


def _cache_source(event: nio.Event) -> dict[str, object]:
    source = dict(event.source)
    content = dict(source.get("content", {}))
    content.setdefault("msgtype", "m.text")
    source["content"] = content
    source.setdefault("event_id", event.event_id)
    source.setdefault("sender", event.sender)
    source.setdefault("origin_server_ts", event.server_timestamp)
    return source


def _make_room_get_event_response(event: nio.Event) -> MagicMock:
    response = MagicMock(spec=nio.RoomGetEventResponse)
    response.event = event
    return response


def _relation_key(
    event_id: str,
    rel_type: RelationshipType,
    *,
    event_type: str = "m.room.message",
    direction: nio.MessageDirection = nio.MessageDirection.back,
    limit: int | None = None,
) -> tuple[str, RelationshipType, str, nio.MessageDirection, int | None]:
    return (event_id, rel_type, event_type, direction, limit)


def _make_relations_client(
    *,
    root_event: nio.Event,
    relations: dict[
        tuple[str, RelationshipType, str, nio.MessageDirection, int | None],
        Iterable[nio.Event] | Exception,
    ],
) -> MagicMock:
    client = MagicMock()
    client.room_get_event = AsyncMock(return_value=_make_room_get_event_response(root_event))

    def room_get_event_relations(
        _room_id: str,
        event_id: str,
        *,
        rel_type: RelationshipType | None = None,
        event_type: str | None = None,
        direction: nio.MessageDirection = nio.MessageDirection.back,
        limit: int | None = None,
    ) -> object:
        assert rel_type is not None
        assert event_type is not None
        value = relations.get((event_id, rel_type, event_type, direction, limit), [])

        async def iterator() -> object:
            if isinstance(value, Exception):
                raise value
            for event in value:
                yield event

        return iterator()

    client.room_get_event_relations = MagicMock(side_effect=room_get_event_relations)
    room_scan_chunk: list[nio.Event] = [root_event]
    seen_event_ids = {getattr(root_event, "event_id", None)}
    for value in relations.values():
        if isinstance(value, Exception):
            continue
        for event in value:
            event_id = getattr(event, "event_id", None)
            if event_id in seen_event_ids:
                continue
            seen_event_ids.add(event_id)
            room_scan_chunk.insert(-1, event)
    client.room_messages = AsyncMock(
        return_value=nio.RoomMessagesResponse(
            room_id="!room:localhost",
            chunk=room_scan_chunk,
            start="",
            end=None,
        ),
    )
    return client


async def _seed_thread_cache(
    cache: ConversationEventCache,
    *,
    room_id: str,
    thread_id: str,
    events: list[dict[str, object]],
) -> None:
    """Seed one authoritative cached thread snapshot for tests."""
    await _replace_thread(cache, room_id, thread_id, events)


def test_event_cache_normalization_is_backend_neutral() -> None:
    """Cache payload normalization should stay backend-neutral."""
    normalized_event = event_normalization.normalize_event_source_for_cache(
        {
            "type": "m.room.message",
            "content": {"body": "hello"},
            "com.mindroom.dispatch_pipeline_timing": {"resolution_ms": 12},
        },
        event_id="$event",
        sender="@user:localhost",
        origin_server_ts=1234,
    )

    assert normalized_event == {
        "type": "m.room.message",
        "content": {"body": "hello"},
        "event_id": "$event",
        "sender": "@user:localhost",
        "origin_server_ts": 1234,
    }


def test_group_lookup_events_by_room_normalizes_and_preserves_order() -> None:
    """Lookup event batch grouping should be shared by durable cache backends."""
    grouped_events = group_lookup_events_by_room(
        [
            (
                "$a",
                "!alpha:localhost",
                {
                    "type": "m.room.message",
                    "content": {"body": "alpha first"},
                    "com.mindroom.dispatch_pipeline_timing": {"resolution_ms": 12},
                },
            ),
            (
                "$b",
                "!beta:localhost",
                {
                    "type": "m.room.message",
                    "event_id": "$already-present",
                    "content": {"body": "beta first"},
                },
            ),
            (
                "$c",
                "!alpha:localhost",
                {
                    "type": "m.room.message",
                    "content": {"body": "alpha second"},
                },
            ),
        ],
    )

    assert list(grouped_events) == ["!alpha:localhost", "!beta:localhost"]
    assert grouped_events == {
        "!alpha:localhost": [
            (
                "$a",
                {
                    "type": "m.room.message",
                    "content": {"body": "alpha first"},
                    "event_id": "$a",
                },
            ),
            (
                "$c",
                {
                    "type": "m.room.message",
                    "content": {"body": "alpha second"},
                    "event_id": "$c",
                },
            ),
        ],
        "!beta:localhost": [
            (
                "$b",
                {
                    "type": "m.room.message",
                    "event_id": "$already-present",
                    "content": {"body": "beta first"},
                },
            ),
        ],
    }


@pytest.mark.asyncio
async def test_conversation_cache_thread_reads_forward_client_fetch_metadata(
    tmp_path: Path,
) -> None:
    """Thread read modes should preserve the facade metadata passed to client fetchers."""
    event_cache = SqliteEventCache(tmp_path / "event_cache.db")
    await event_cache.initialize()
    client = MagicMock()
    conversation_cache = _conversation_cache_for_thread_reads(tmp_path, event_cache, client=client)
    read_modes = [
        ("get_thread_history", "fetch_thread_history", True, 50.0),
        ("get_dispatch_thread_snapshot", "fetch_dispatch_thread_snapshot", False, 75.0),
        ("get_dispatch_thread_history", "fetch_dispatch_thread_history", True, 100.0),
    ]
    post_coordinator_read_starts = {
        "get_thread_history": 1.06,
        "get_dispatch_thread_snapshot": 2.08,
        "get_dispatch_thread_history": 3.11,
    }
    fetchers = {
        name: AsyncMock(return_value=thread_history_result([], is_full_history=is_full_history))
        for _method_name, name, is_full_history, _queue_wait_ms in read_modes
    }

    try:
        with (
            patch("mindroom.matrix.conversation_cache.fetch_thread_history", fetchers["fetch_thread_history"]),
            patch(
                "mindroom.matrix.conversation_cache.fetch_dispatch_thread_snapshot",
                fetchers["fetch_dispatch_thread_snapshot"],
            ),
            patch(
                "mindroom.matrix.conversation_cache.fetch_dispatch_thread_history",
                fetchers["fetch_dispatch_thread_history"],
            ),
            patch(
                "mindroom.matrix.cache.thread_reads.time.perf_counter",
                side_effect=[
                    1.0,
                    1.05,
                    1.06,
                    2.0,
                    2.01,
                    2.075,
                    2.075,
                    2.075,
                    2.08,
                    3.0,
                    3.01,
                    3.1,
                    3.1,
                    3.1,
                    3.11,
                ],
            ),
        ):
            read_methods = {
                "get_thread_history": conversation_cache.get_thread_history,
                "get_dispatch_thread_snapshot": conversation_cache.get_dispatch_thread_snapshot,
                "get_dispatch_thread_history": conversation_cache.get_dispatch_thread_history,
            }
            for method_name, _name, is_full_history, _queue_wait_ms in read_modes:
                result = await read_methods[method_name](
                    "!room:localhost",
                    "$thread:localhost",
                    caller_label=f"caller-{method_name}",
                )
                assert result.is_full_history is is_full_history

        for method_name, name, _is_full_history, queue_wait_ms in read_modes:
            fetchers[name].assert_awaited_once_with(
                client,
                "!room:localhost",
                "$thread:localhost",
                event_cache=event_cache,
                trusted_sender_ids=conversation_cache._trusted_sender_ids(),
                caller_label=f"caller-{method_name}",
                coordinator_queue_wait_ms=queue_wait_ms,
                post_coordinator_read_started=post_coordinator_read_starts[method_name],
                # Always supplied now: the refill no longer depends on a write coordinator.
                refill=ANY,
            )
            assert callable(fetchers[name].await_args.kwargs["refill"])
    finally:
        await event_cache.close()


@pytest.mark.asyncio
async def test_dispatch_thread_read_degrades_when_cache_coordinator_never_drains(
    tmp_path: Path,
) -> None:
    """Dispatch-safe reads should not wait unbounded for advisory cache coordination."""
    event_cache = SqliteEventCache(tmp_path / "event_cache.db")
    await event_cache.initialize()
    client = MagicMock()
    conversation_cache = _conversation_cache_for_thread_reads(tmp_path, event_cache, client=client)

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

    coordinator = MagicMock()
    coordinator.wait_for_thread_idle = AsyncMock(side_effect=never_idle)
    conversation_cache.runtime.event_cache_write_coordinator = coordinator
    _set_dispatch_thread_read_timeout(conversation_cache, 0.01)

    try:
        result = await asyncio.wait_for(
            conversation_cache.get_dispatch_thread_snapshot(
                "!room:localhost",
                "$thread:localhost",
                caller_label="dispatch_context",
            ),
            timeout=0.2,
        )
    finally:
        await event_cache.close()

    assert result == []
    assert result.is_full_history is False
    assert result.diagnostics["thread_read_degraded"] is True
    assert result.diagnostics["thread_read_error"] == "cache_coordinator_timeout"
    assert result.diagnostics["thread_read_source"] == "degraded"
    assert result.diagnostics["caller_label"] == "dispatch_context"
    coordinator.wait_for_thread_idle.assert_awaited_once()


@pytest.mark.asyncio
async def test_dispatch_thread_read_timeout_does_not_cancel_pending_cache_write(
    tmp_path: Path,
) -> None:
    """Timeouts around dispatch-safe coordinator waits must not cancel cache mutation tasks."""
    event_cache = SqliteEventCache(tmp_path / "event_cache.db")
    await event_cache.initialize()
    client = MagicMock()
    conversation_cache = _conversation_cache_for_thread_reads(tmp_path, event_cache, client=client)
    coordinator = EventCacheWriteCoordinator(logger=MagicMock())
    release_write = asyncio.Event()
    write_started = asyncio.Event()

    async def pending_cache_write() -> None:
        write_started.set()
        await release_write.wait()

    pending_write_task = coordinator.queue_thread_update(
        "!room:localhost",
        "$thread:localhost",
        pending_cache_write,
        name="matrix_cache_pending_test_write",
        coordination_scope=event_cache.principal_id,
    )
    conversation_cache.runtime.event_cache_write_coordinator = coordinator
    _set_dispatch_thread_read_timeout(conversation_cache, 0.01)
    baseline_wait_tasks = _pending_thread_cache_update_wait_tasks()

    try:
        await asyncio.wait_for(write_started.wait(), timeout=0.2)
        with patch(
            "mindroom.matrix.conversation_cache.fetch_dispatch_thread_snapshot",
            AsyncMock(side_effect=AssertionError("coordinator timeout should not fetch")),
        ):
            result = await asyncio.wait_for(
                conversation_cache.get_dispatch_thread_snapshot(
                    "!room:localhost",
                    "$thread:localhost",
                    caller_label="dispatch_context",
                ),
                timeout=0.2,
            )

        assert result.diagnostics["thread_read_error"] == "cache_coordinator_timeout"
        assert pending_write_task.cancelled() is False
        assert pending_write_task.done() is False
        await asyncio.sleep(0)
        assert _pending_thread_cache_update_wait_tasks() == baseline_wait_tasks
    finally:
        release_write.set()
        await pending_write_task
        await event_cache.close()


@pytest.mark.asyncio
async def test_dispatch_thread_read_degrades_when_fetcher_stalls(
    tmp_path: Path,
) -> None:
    """Dispatch-safe reads should not wait indefinitely on a direct Matrix read-through."""
    event_cache = SqliteEventCache(tmp_path / "event_cache.db")
    await event_cache.initialize()
    client = MagicMock()
    conversation_cache = _conversation_cache_for_thread_reads(tmp_path, event_cache, client=client)

    release_fetch = asyncio.Event()

    async def never_returns(*_args: object, **_kwargs: object) -> ThreadHistoryResult:
        await release_fetch.wait()
        return thread_history_result([], is_full_history=False)

    coordinator = EventCacheWriteCoordinator(logger=MagicMock())
    conversation_cache.runtime.event_cache_write_coordinator = coordinator
    _set_dispatch_thread_read_timeout(conversation_cache, 0.01)

    try:
        with patch(
            "mindroom.matrix.conversation_cache.fetch_dispatch_thread_snapshot",
            AsyncMock(side_effect=never_returns),
        ):
            result = await asyncio.wait_for(
                conversation_cache.get_dispatch_thread_snapshot(
                    "!room:localhost",
                    "$thread:localhost",
                    caller_label="dispatch_context",
                ),
                timeout=0.2,
            )
    finally:
        release_fetch.set()
        await coordinator.close()
        await event_cache.close()

    assert result == []
    assert result.is_full_history is False
    assert result.diagnostics["thread_read_degraded"] is True
    assert result.diagnostics["thread_read_error"] == "dispatch_read_timeout"
    assert result.diagnostics["thread_read_source"] == "degraded"
    assert result.diagnostics["caller_label"] == "dispatch_context"
    assert "dispatch_fetch_wait_ms" in result.diagnostics


@pytest.mark.asyncio
async def test_dispatch_context_waits_for_strict_thread_history_after_degraded_snapshot(
    tmp_path: Path,
) -> None:
    """A proven thread must fall back to strict history before dispatch planning."""
    runtime_paths = test_runtime_paths(tmp_path)
    config = bind_runtime_paths(
        Config(
            agents={
                "primary": AgentConfig(display_name="Primary", rooms=["!room:localhost"]),
                "secondary": AgentConfig(display_name="Secondary", rooms=["!room:localhost"]),
            },
            models={"default": ModelConfig(provider="test", id="test-model")},
        ),
        runtime_paths,
    )
    route_paths = runtime_paths_for(config)
    route_ids = entity_ids(config, route_paths)
    runtime = BotRuntimeState(
        client=MagicMock(),
        config=config,
        runtime_paths=runtime_paths,
        enable_streaming=False,
        orchestrator=None,
        event_cache=MagicMock(),
        event_cache_write_coordinator=None,
    )
    resolver = ConversationResolver(
        ConversationResolverDeps(
            runtime=runtime,
            logger=MagicMock(),
            runtime_paths=runtime_paths,
            agent_name="primary",
            matrix_id=route_ids["primary"],
            conversation_cache=MagicMock(),
        ),
    )
    degraded_history = thread_history_result(
        [],
        is_full_history=False,
        diagnostics={THREAD_HISTORY_DEGRADED_DIAGNOSTIC: True},
    )
    strict_history = thread_history_result(
        [
            ResolvedVisibleMessage.synthetic(
                sender=route_ids["primary"].full_id,
                body="I can handle this.",
                event_id="$agent-reply",
                thread_id="$thread:localhost",
            ),
            ResolvedVisibleMessage.synthetic(
                sender="@requester:localhost",
                body="Please continue.",
                event_id="$user-follow-up",
                thread_id="$thread:localhost",
            ),
        ],
        is_full_history=True,
    )
    event_info = MagicMock(spec=EventInfo)

    with (
        patch.object(
            resolver,
            "_explicit_thread_id_for_event",
            AsyncMock(return_value=_ThreadIdLookup(thread_id="$thread:localhost", thread_history=degraded_history)),
        ),
        patch.object(
            resolver,
            "_read_thread_messages",
            AsyncMock(return_value=strict_history),
        ) as read_thread_messages,
    ):
        result = await resolver._resolve_thread_context(
            "!room:localhost",
            "$incoming:localhost",
            event_info,
            mode=ThreadReadMode.DISPATCH_SNAPSHOT,
            caller_label="dispatch_context",
        )

    assert result.is_thread is True
    assert result.thread_id == "$thread:localhost"
    assert result.thread_history == strict_history
    assert result.requires_model_history_refresh is False
    assert result.replay_guard_degraded is False
    read_thread_messages.assert_awaited_once_with(
        "!room:localhost",
        "$thread:localhost",
        mode=ThreadReadMode.STRICT_FULL,
        caller_label="dispatch_context_strict_thread_fallback",
    )
    assert agent_response_should_respond(
        agent_name="primary",
        am_i_mentioned=False,
        is_thread=True,
        room=create_mock_room("!room:localhost", ["primary", "secondary"], config),
        thread_history=result.thread_history,
        config=config,
        runtime_paths=route_paths,
        sender_id="@requester:localhost",
        available_responders_in_room=[route_ids["primary"], route_ids["secondary"]],
    )


@pytest.mark.asyncio
async def test_dispatch_thread_read_uses_single_deadline_after_coordinator_wait(
    tmp_path: Path,
) -> None:
    """Dispatch fetches should not receive a fresh timeout after the coordinator wait spends the budget."""
    event_cache = SqliteEventCache(tmp_path / "event_cache.db")
    await event_cache.initialize()
    client = MagicMock()
    conversation_cache = _conversation_cache_for_thread_reads(tmp_path, event_cache, client=client)

    coordinator = MagicMock()
    coordinator.wait_for_thread_idle = AsyncMock(return_value=None)
    conversation_cache.runtime.event_cache_write_coordinator = coordinator
    _set_dispatch_thread_read_timeout(conversation_cache, 1.0)

    clock_values = iter([100.0, 100.0, 101.25, 101.25, 101.25, 101.25])

    def perf_counter() -> float:
        return next(clock_values, 101.25)

    try:
        with (
            patch("mindroom.matrix.cache.thread_reads.time.perf_counter", side_effect=perf_counter),
            patch(
                "mindroom.matrix.conversation_cache.fetch_dispatch_thread_snapshot",
                AsyncMock(side_effect=AssertionError("spent dispatch deadline must not start fetch")),
            ) as fetch_dispatch_thread_snapshot,
        ):
            result = await conversation_cache.get_dispatch_thread_snapshot(
                "!room:localhost",
                "$thread:localhost",
                caller_label="dispatch_context",
            )
    finally:
        await event_cache.close()

    assert result == []
    assert result.is_full_history is False
    assert result.diagnostics["thread_read_degraded"] is True
    assert result.diagnostics["thread_read_error"] == "dispatch_read_timeout"
    assert result.diagnostics["thread_read_source"] == "degraded"
    assert "dispatch_fetch_wait_ms" in result.diagnostics
    coordinator.wait_for_thread_idle.assert_awaited_once()
    fetch_dispatch_thread_snapshot.assert_not_awaited()


@pytest.mark.asyncio
async def test_strict_thread_history_uses_no_stale_fetch_without_dispatch_timeout(
    tmp_path: Path,
) -> None:
    """Post-lock strict reads should wait normally but still reject stale fallback."""
    event_cache = SqliteEventCache(tmp_path / "event_cache.db")
    await event_cache.initialize()
    client = MagicMock()
    conversation_cache = _conversation_cache_for_thread_reads(tmp_path, event_cache, client=client)

    coordinator = EventCacheWriteCoordinator(logger=conversation_cache.logger)
    conversation_cache.runtime.event_cache_write_coordinator = coordinator
    fetched_history = thread_history_result([], is_full_history=True)

    try:
        with (
            patch.object(
                coordinator,
                "wait_for_thread_idle",
                wraps=coordinator.wait_for_thread_idle,
            ) as wait_for_thread_idle,
            patch(
                "mindroom.matrix.conversation_cache.refresh_thread_history_from_source",
                AsyncMock(return_value=fetched_history),
            ) as refresh_thread_history,
            patch(
                "mindroom.matrix.conversation_cache.fetch_thread_history",
                AsyncMock(side_effect=AssertionError("strict reads must not allow stale fallback")),
            ),
        ):
            result = await conversation_cache.get_strict_thread_history(
                "!room:localhost",
                "$thread:localhost",
                caller_label="dispatch_post_lock_refresh",
            )
    finally:
        await coordinator.close()
        await event_cache.close()

    assert result.is_full_history is True
    wait_for_thread_idle.assert_awaited_once()
    refresh_thread_history.assert_awaited_once()
    assert refresh_thread_history.await_args.kwargs["allow_stale_fallback"] is False


@pytest.mark.asyncio
async def test_strict_source_refresh_bypasses_usable_cache(
    tmp_path: Path,
) -> None:
    """Explicit source refresh should serialize one Matrix fetch without accepting a cache hit."""
    event_cache = SqliteEventCache(tmp_path / "event_cache.db")
    await event_cache.initialize()
    client = object()
    conversation_cache = _conversation_cache_for_thread_reads(tmp_path, event_cache, client=client)
    coordinator = EventCacheWriteCoordinator(logger=conversation_cache.logger)
    conversation_cache.runtime.event_cache_write_coordinator = coordinator
    barrier_entered = asyncio.Event()
    release_write = asyncio.Event()
    write_started = asyncio.Event()
    wait_for_thread_idle = coordinator.wait_for_thread_idle

    async def pending_cache_write() -> None:
        write_started.set()
        await release_write.wait()

    async def observed_wait_for_thread_idle(
        room_id: str,
        thread_id: str,
        *,
        ignore_cancelled_room_fences: bool = False,
        coordination_scope: str,
    ) -> None:
        barrier_entered.set()
        await wait_for_thread_idle(
            room_id,
            thread_id,
            ignore_cancelled_room_fences=ignore_cancelled_room_fences,
            coordination_scope=coordination_scope,
        )

    pending_write_task = coordinator.queue_thread_update(
        "!room:localhost",
        "$thread:localhost",
        pending_cache_write,
        name="matrix_cache_pending_source_refresh_test_write",
        coordination_scope=event_cache.principal_id,
    )
    fetched_history = thread_history_result(
        [ResolvedVisibleMessage.synthetic(sender="@bot:localhost", body="Target", event_id="$target")],
        is_full_history=True,
    )

    try:
        await asyncio.wait_for(write_started.wait(), timeout=5.0)
        with (
            patch.object(
                coordinator,
                "wait_for_thread_idle",
                side_effect=observed_wait_for_thread_idle,
            ),
            patch(
                "mindroom.matrix.conversation_cache.fetch_dispatch_thread_history",
                AsyncMock(side_effect=AssertionError("source refresh must bypass cache selection")),
            ) as cache_thread_history,
            patch(
                "mindroom.matrix.conversation_cache.refresh_thread_history_from_source",
                AsyncMock(return_value=fetched_history),
            ) as refresh_thread_history,
        ):
            read_task = asyncio.create_task(
                conversation_cache.refresh_strict_thread_history_from_source(
                    "!room:localhost",
                    "$thread:localhost",
                    caller_label="startup_auto_resume_freshness",
                ),
            )
            await asyncio.wait_for(barrier_entered.wait(), timeout=5.0)
            refresh_thread_history.assert_not_awaited()
            release_write.set()
            result = await asyncio.wait_for(read_task, timeout=5.0)
    finally:
        release_write.set()
        await pending_write_task
        await coordinator.close()
        await event_cache.close()

    assert [message.event_id for message in result] == ["$target"]
    assert result.is_full_history is True
    refresh_thread_history.assert_awaited_once()
    assert refresh_thread_history.await_args.args[:4] == (
        client,
        "!room:localhost",
        "$thread:localhost",
        event_cache,
    )
    assert refresh_thread_history.await_args.kwargs["allow_stale_fallback"] is False
    cache_thread_history.assert_not_awaited()


@pytest.mark.asyncio
async def test_startup_source_refresh_does_not_join_foreground_refill(
    tmp_path: Path,
) -> None:
    """Startup refresh should finish without joining a blocked foreground refill."""
    event_cache = SqliteEventCache(tmp_path / "event_cache.db")
    await event_cache.initialize()
    client = object()
    conversation_cache = _conversation_cache_for_thread_reads(tmp_path, event_cache, client=client)
    coordinator = EventCacheWriteCoordinator(logger=conversation_cache.logger)
    conversation_cache.runtime.event_cache_write_coordinator = coordinator
    first_source_call_started = asyncio.Event()
    release_first_source_call = asyncio.Event()
    second_source_call_started = asyncio.Event()
    source_call_count = 0
    foreground_refill: asyncio.Task[ThreadHistoryResult] | None = None
    startup_refresh: asyncio.Task[ThreadHistoryResult] | None = None
    foreground_history = thread_history_result(
        [ResolvedVisibleMessage.synthetic(sender="@bot:localhost", body="Foreground", event_id="$foreground")],
        is_full_history=True,
    )
    startup_history = thread_history_result(
        [ResolvedVisibleMessage.synthetic(sender="@bot:localhost", body="Startup", event_id="$startup")],
        is_full_history=True,
        diagnostics={"thread_read_source": "homeserver"},
    )

    async def source_refresh(*_args: object, **kwargs: object) -> ThreadHistoryResult:
        nonlocal source_call_count
        source_call_count += 1
        if source_call_count == 1:
            first_source_call_started.set()
            await release_first_source_call.wait()
            return foreground_history
        assert kwargs["hydrate_sidecars"] is True
        assert kwargs["allow_stale_fallback"] is False
        assert kwargs["caller_label"] == "startup_auto_resume_freshness"
        second_source_call_started.set()
        return startup_history

    try:
        with patch(
            "mindroom.matrix.conversation_cache.refresh_thread_history_from_source",
            AsyncMock(side_effect=source_refresh),
        ):
            foreground_refill = asyncio.create_task(
                conversation_cache._refill_thread_from_client(
                    "!room:localhost",
                    "$thread:localhost",
                    cache_reject_diagnostics=None,
                    wants_full_history=True,
                    allows_stale_fallback=False,
                ),
            )
            await asyncio.wait_for(first_source_call_started.wait(), timeout=1.0)
            foreground_single_flight = dict(conversation_cache._refill_single_flight._in_flight)
            assert len(foreground_single_flight) == 1

            startup_refresh = asyncio.create_task(
                conversation_cache.refresh_startup_thread_history_from_source(
                    "!room:localhost",
                    "$thread:localhost",
                    caller_label="startup_auto_resume_freshness",
                ),
            )
            await asyncio.wait_for(second_source_call_started.wait(), timeout=1.0)
            startup_result = await asyncio.wait_for(startup_refresh, timeout=1.0)

            assert [message.event_id for message in startup_result] == ["$startup"]
            assert startup_result.is_full_history is True
            assert foreground_refill.done() is False
            assert conversation_cache._refill_single_flight._in_flight == foreground_single_flight

            release_first_source_call.set()
            foreground_result = await asyncio.wait_for(foreground_refill, timeout=1.0)
            assert [message.event_id for message in foreground_result] == ["$foreground"]
    finally:
        release_first_source_call.set()
        await asyncio.gather(
            *(task for task in (foreground_refill, startup_refresh) if task is not None),
            return_exceptions=True,
        )
        await coordinator.close()
        await event_cache.close()


@pytest.mark.asyncio
async def test_live_read_does_not_wait_for_running_startup_source_refresh(
    tmp_path: Path,
) -> None:
    """A running startup scan must not occupy the live same-thread read lane."""
    room_id = "!room:localhost"
    thread_id = "$thread:localhost"
    event_cache = SqliteEventCache(tmp_path / "event_cache.db")
    await event_cache.initialize()
    await _replace_thread(event_cache, room_id, thread_id, [_clear_payload(thread_id, body="Cached root")])
    conversation_cache = _conversation_cache_for_thread_reads(tmp_path, event_cache, client=MagicMock())
    coordinator = EventCacheWriteCoordinator(logger=conversation_cache.logger)
    conversation_cache.runtime.event_cache_write_coordinator = coordinator
    source_call_started = asyncio.Event()
    release_source_call = asyncio.Event()
    startup_refresh: asyncio.Task[ThreadHistoryResult] | None = None
    live_read: asyncio.Task[ThreadHistoryResult] | None = None

    async def blocking_source_refresh(*_args: object, **_kwargs: object) -> ThreadHistoryResult:
        source_call_started.set()
        await release_source_call.wait()
        return thread_history_result([], is_full_history=True)

    try:
        with patch(
            "mindroom.matrix.conversation_cache.refresh_thread_history_from_source",
            AsyncMock(side_effect=blocking_source_refresh),
        ):
            startup_refresh = asyncio.create_task(
                conversation_cache.refresh_startup_thread_history_from_source(
                    room_id,
                    thread_id,
                    caller_label="startup_auto_resume_freshness",
                ),
            )
            await asyncio.wait_for(source_call_started.wait(), timeout=1.0)

            live_read = asyncio.create_task(
                conversation_cache.get_dispatch_thread_history(
                    room_id,
                    thread_id,
                    caller_label="live_dispatch",
                ),
            )
            live_result = await asyncio.wait_for(live_read, timeout=1.0)

            assert [message.event_id for message in live_result] == [thread_id]
            assert startup_refresh.done() is False

            release_source_call.set()
            startup_result = await asyncio.wait_for(startup_refresh, timeout=1.0)
            assert startup_result == []
            assert startup_result.is_full_history is True
    finally:
        release_source_call.set()
        await asyncio.gather(
            *(task for task in (startup_refresh, live_read) if task is not None),
            return_exceptions=True,
        )
        await coordinator.close()
        await event_cache.close()


@pytest.mark.asyncio
async def test_cancelled_startup_source_refresh_leaves_no_shared_refill_state(
    tmp_path: Path,
) -> None:
    """Cancelling startup refresh should not retain coordinator or singleflight state."""
    event_cache = SqliteEventCache(tmp_path / "event_cache.db")
    await event_cache.initialize()
    conversation_cache = _conversation_cache_for_thread_reads(tmp_path, event_cache, client=object())
    coordinator = EventCacheWriteCoordinator(logger=conversation_cache.logger)
    conversation_cache.runtime.event_cache_write_coordinator = coordinator
    source_call_started = asyncio.Event()
    release_source_call = asyncio.Event()
    startup_refresh: asyncio.Task[ThreadHistoryResult] | None = None

    async def blocking_source_refresh(*_args: object, **_kwargs: object) -> ThreadHistoryResult:
        source_call_started.set()
        await release_source_call.wait()
        return thread_history_result([], is_full_history=True)

    try:
        with patch(
            "mindroom.matrix.conversation_cache.refresh_thread_history_from_source",
            AsyncMock(side_effect=blocking_source_refresh),
        ):
            startup_refresh = asyncio.create_task(
                conversation_cache.refresh_startup_thread_history_from_source(
                    "!room:localhost",
                    "$thread:localhost",
                    caller_label="startup_auto_resume_freshness",
                ),
            )
            await asyncio.wait_for(source_call_started.wait(), timeout=1.0)
            startup_refresh.cancel()
            with pytest.raises(asyncio.CancelledError):
                await asyncio.wait_for(startup_refresh, timeout=1.0)

            assert conversation_cache._refill_single_flight._in_flight == {}
            assert coordinator._room_states == {}
    finally:
        release_source_call.set()
        await asyncio.gather(
            *(task for task in (startup_refresh,) if task is not None),
            return_exceptions=True,
        )
        await coordinator.close()
        await event_cache.close()


@pytest.mark.asyncio
async def test_strict_thread_history_propagates_cache_coordinator_timeout(
    tmp_path: Path,
) -> None:
    """Post-lock strict reads must not be converted into degraded dispatch results."""
    event_cache = SqliteEventCache(tmp_path / "event_cache.db")
    await event_cache.initialize()
    client = MagicMock()
    conversation_cache = _conversation_cache_for_thread_reads(tmp_path, event_cache, client=client)

    coordinator = MagicMock()
    coordinator.wait_for_thread_idle = AsyncMock(side_effect=TimeoutError("strict wait timed out"))
    conversation_cache.runtime.event_cache_write_coordinator = coordinator

    try:
        with (
            patch(
                "mindroom.matrix.conversation_cache.refresh_thread_history_from_source",
                AsyncMock(side_effect=AssertionError("strict read should not fetch after timeout")),
            ) as refresh_thread_history,
            pytest.raises(TimeoutError, match="strict wait timed out"),
        ):
            await conversation_cache.get_strict_thread_history(
                "!room:localhost",
                "$thread:localhost",
                caller_label="dispatch_post_lock_refresh",
            )
    finally:
        await event_cache.close()

    coordinator.wait_for_thread_idle.assert_awaited_once()
    refresh_thread_history.assert_not_awaited()


@pytest.mark.asyncio
async def test_conversation_cache_startup_prewarm_bulk_refresh_preserves_metadata(
    tmp_path: Path,
) -> None:
    """Startup prewarm should call the bulk room refresher with fixed metadata."""
    event_cache = SqliteEventCache(tmp_path / "event_cache.db")
    await event_cache.initialize()
    client = MagicMock()
    conversation_cache = _conversation_cache_for_thread_reads(tmp_path, event_cache, client=client)
    stats = BulkThreadRefreshStats(
        requested_threads=1,
        usable_threads=1,
        missing_root_ids=frozenset(),
        room_scan_pages=1,
        scanned_event_count=2,
    )
    bulk_refresh_room_thread_histories = AsyncMock(return_value=stats)

    try:
        with patch(
            "mindroom.matrix.conversation_cache.bulk_refresh_room_thread_histories",
            bulk_refresh_room_thread_histories,
        ):
            result = await conversation_cache._bulk_refresh_startup_threads(
                "!room:localhost",
                ["$thread:localhost"],
            )

        assert result == stats
        bulk_refresh_room_thread_histories.assert_awaited_once_with(
            client,
            "!room:localhost",
            event_cache,
            thread_root_ids=["$thread:localhost"],
            caller_label="startup_thread_prewarm",
            max_scan_pages=20,
        )
    finally:
        await event_cache.close()


@pytest.mark.asyncio
async def test_thread_snapshot_storage_exposes_direct_gap_reads(tmp_path: Path) -> None:
    """A stored snapshot should expose the newest gap marker recorded against its thread."""
    db, _maintenance_report, _generation = await event_cache_module._initialize_event_cache_db(
        tmp_path / "event_cache.db",
    )

    try:
        await sqlite_event_cache_threads.replace_thread_locked(
            db,
            principal_id="__mindroom_default_principal__",
            room_id="!room:localhost",
            thread_id="$thread_root",
            events=[
                {
                    "event_id": "$thread_root",
                    "sender": "@user:localhost",
                    "origin_server_ts": 1000,
                    "type": "m.room.message",
                    "content": {"body": "Root message", "msgtype": "m.text"},
                },
            ],
            stored_at=100.0,
            fetch_started_at=100.0,
        )
        with patch("mindroom.matrix.cache.sqlite_event_cache_threads.time.time", return_value=200.0):
            await sqlite_event_cache_threads.mark_thread_gap_locked(
                db,
                principal_id="__mindroom_default_principal__",
                room_id="!room:localhost",
                thread_id="$thread_root",
                reason="thread_stale",
            )
            await sqlite_event_cache_threads.mark_room_gap_locked(
                db,
                principal_id="__mindroom_default_principal__",
                room_id="!room:localhost",
                reason="room_stale",
            )
        await db.commit()

        gap = await sqlite_event_cache_threads.load_thread_cache_gap(
            db,
            principal_id="__mindroom_default_principal__",
            room_id="!room:localhost",
            thread_id="$thread_root",
        )
    finally:
        await db.close()

    # One marker per thread, not a thread column joined against a room column: the room-scoped
    # marker fanned out onto this thread's row and, arriving no earlier, owns the reason.
    assert gap is not None
    assert gap.gap_marked_at == 200.0
    assert gap.gap_reason == "room_stale"
    assert thread_cache_rejection_reason(gap) == "room_stale"


@pytest.mark.asyncio
async def test_sqlite_gap_markers_are_monotonic(tmp_path: Path) -> None:
    """An older gap marker must not downgrade a newer one, at either scope."""
    db, _maintenance_report, _generation = await event_cache_module._initialize_event_cache_db(
        tmp_path / "event_cache.db",
    )

    try:
        with patch("mindroom.matrix.cache.sqlite_event_cache_threads.time.time", return_value=200.0):
            await sqlite_event_cache_threads.mark_thread_gap_locked(
                db,
                principal_id="__mindroom_default_principal__",
                room_id="!room:localhost",
                thread_id="$thread_root",
                reason="newer_thread_marker",
            )
            await sqlite_event_cache_threads.mark_room_gap_locked(
                db,
                principal_id="__mindroom_default_principal__",
                room_id="!room:localhost",
                reason="newer_room_marker",
            )
        with patch("mindroom.matrix.cache.sqlite_event_cache_threads.time.time", return_value=100.0):
            await sqlite_event_cache_threads.mark_thread_gap_locked(
                db,
                principal_id="__mindroom_default_principal__",
                room_id="!room:localhost",
                thread_id="$thread_root",
                reason="older_thread_marker",
            )
            await sqlite_event_cache_threads.mark_room_gap_locked(
                db,
                principal_id="__mindroom_default_principal__",
                room_id="!room:localhost",
                reason="older_room_marker",
            )
        await db.commit()

        gap = await sqlite_event_cache_threads.load_thread_cache_gap(
            db,
            principal_id="__mindroom_default_principal__",
            room_id="!room:localhost",
            thread_id="$thread_root",
        )
    finally:
        await db.close()

    assert gap is not None
    assert gap.gap_marked_at == 200.0
    assert gap.gap_reason == "newer_room_marker"


@pytest.mark.parametrize(
    ("gap", "expected_reason"),
    [
        pytest.param(None, None, id="no_marker_is_usable"),
        pytest.param(
            ThreadCacheGap(gap_marked_at=100.0, gap_reason="limited_sync_timeline"),
            "limited_sync_timeline",
            id="marker_reports_its_reason",
        ),
        pytest.param(
            ThreadCacheGap(gap_marked_at=100.0, gap_reason=None),
            "thread_gap_marked",
            id="reasonless_marker_still_rejects",
        ),
    ],
)
def test_thread_cache_rejection_reason_rule_table(
    gap: ThreadCacheGap | None,
    expected_reason: str | None,
) -> None:
    """The snapshot gate asks exactly one question: is a gap recorded against this thread."""
    assert thread_cache_rejection_reason(gap) == expected_reason


@pytest.mark.asyncio
async def test_thread_gap_marked_midflight_survives_the_replacement(tmp_path: Path) -> None:
    """A gap marked after a fetch began is not covered by that fetch, so it outlives it."""
    cache = SqliteEventCache(tmp_path / "event_cache.db")
    await cache.initialize()
    root_source = {
        "event_id": "$thread_root",
        "sender": "@user:localhost",
        "origin_server_ts": 1000,
        "type": "m.room.message",
        "content": {"body": "Root message", "msgtype": "m.text"},
    }

    try:
        await _replace_thread(cache, "!room:localhost", "$thread_root", [root_source], fetch_started_at=100.0)
        with patch("mindroom.matrix.cache.sqlite_event_cache_threads.time.time", return_value=200.0):
            await cache.mark_thread_gap("!room:localhost", "$thread_root", reason="live_thread_mutation")

        # This fetch started before the marker, so it cannot have seen what the marker describes.
        stored_behind_marker = await cache.replace_thread(
            "!room:localhost",
            "$thread_root",
            [root_source],
            expected_membership_epoch=await cache.room_membership_epoch("!room:localhost"),
            fetch_started_at=150.0,
        )
        gap_after_uncovered_fetch = await cache.get_thread_cache_gap("!room:localhost", "$thread_root")

        # This one started after it, so it covers the marker and clears it.
        stored_after_marker = await cache.replace_thread(
            "!room:localhost",
            "$thread_root",
            [root_source],
            expected_membership_epoch=await cache.room_membership_epoch("!room:localhost"),
            fetch_started_at=250.0,
        )
        gap_after_covering_fetch = await cache.get_thread_cache_gap("!room:localhost", "$thread_root")
    finally:
        await cache.close()

    # The snapshot installs either way -- refusing it would strand the thread -- but the marker is
    # what decides whether the next read may use it.
    assert stored_behind_marker
    assert gap_after_uncovered_fetch is not None
    assert gap_after_uncovered_fetch.gap_marked_at == 200.0
    assert thread_cache_rejection_reason(gap_after_uncovered_fetch) == "live_thread_mutation"

    assert stored_after_marker
    assert gap_after_covering_fetch is None


@pytest.mark.asyncio
async def test_room_gap_marked_midflight_survives_the_replacement(tmp_path: Path) -> None:
    """The room-scoped marker follows the same covering rule once it has fanned out."""
    cache = SqliteEventCache(tmp_path / "event_cache.db")
    await cache.initialize()
    root_source = {
        "event_id": "$thread_root",
        "sender": "@user:localhost",
        "origin_server_ts": 1000,
        "type": "m.room.message",
        "content": {"body": "Root message", "msgtype": "m.text"},
    }

    try:
        await _replace_thread(cache, "!room:localhost", "$thread_root", [root_source], fetch_started_at=100.0)
        with patch("mindroom.matrix.cache.sqlite_event_cache_threads.time.time", return_value=200.0):
            await cache.mark_room_threads_gap("!room:localhost", reason="sync_thread_lookup_unavailable")

        stored = await cache.replace_thread(
            "!room:localhost",
            "$thread_root",
            [root_source],
            expected_membership_epoch=await cache.room_membership_epoch("!room:localhost"),
            fetch_started_at=150.0,
        )
        gap = await cache.get_thread_cache_gap("!room:localhost", "$thread_root")
    finally:
        await cache.close()

    assert stored
    assert gap is not None
    assert gap.gap_marked_at == 200.0
    assert thread_cache_rejection_reason(gap) == "sync_thread_lookup_unavailable"


@pytest.mark.asyncio
async def test_event_cache_store_and_retrieve(event_cache: ConversationEventCache) -> None:
    """Stored events should round-trip in timestamp order."""
    cache = event_cache

    try:
        await _seed_thread_cache(
            cache,
            room_id="!room:localhost",
            thread_id="$thread_root",
            events=[
                {
                    "event_id": "$reply",
                    "sender": "@agent:localhost",
                    "origin_server_ts": 2000,
                    "type": "m.room.message",
                    "content": {
                        "body": "Reply in thread",
                        "msgtype": "m.text",
                        "m.relates_to": {"rel_type": "m.thread", "event_id": "$thread_root"},
                    },
                },
                {
                    "event_id": "$thread_root",
                    "sender": "@user:localhost",
                    "origin_server_ts": 1000,
                    "type": "m.room.message",
                    "content": {"body": "Root message", "msgtype": "m.text"},
                },
            ],
        )

        cached_events = await cache.get_thread_events("!room:localhost", "$thread_root")
    finally:
        await cache.close()

    assert cached_events is not None
    assert [event["event_id"] for event in cached_events] == ["$thread_root", "$reply"]


@pytest.mark.asyncio
async def test_get_recent_room_thread_ids_orders_by_latest_event_in_each_thread(
    event_cache: ConversationEventCache,
) -> None:
    """Recent thread IDs should be ordered by the freshest cached event per thread, not by root timestamp."""
    cache = event_cache

    try:
        await _seed_thread_cache(
            cache,
            room_id="!room:localhost",
            thread_id="$thread_old_root_recent_reply",
            events=[
                {
                    "event_id": "$thread_old_root_recent_reply",
                    "sender": "@user:localhost",
                    "origin_server_ts": 1000,
                    "type": "m.room.message",
                    "content": {"body": "Old root", "msgtype": "m.text"},
                },
                {
                    "event_id": "$recent_reply",
                    "sender": "@agent:localhost",
                    "origin_server_ts": 9000,
                    "type": "m.room.message",
                    "content": {
                        "body": "Recent reply",
                        "msgtype": "m.text",
                        "m.relates_to": {
                            "rel_type": "m.thread",
                            "event_id": "$thread_old_root_recent_reply",
                        },
                    },
                },
            ],
        )
        await _seed_thread_cache(
            cache,
            room_id="!room:localhost",
            thread_id="$thread_recent_root_no_replies",
            events=[
                {
                    "event_id": "$thread_recent_root_no_replies",
                    "sender": "@user:localhost",
                    "origin_server_ts": 5000,
                    "type": "m.room.message",
                    "content": {"body": "Recent root", "msgtype": "m.text"},
                },
            ],
        )
        await _seed_thread_cache(
            cache,
            room_id="!other_room:localhost",
            thread_id="$thread_other_room",
            events=[
                {
                    "event_id": "$thread_other_room",
                    "sender": "@user:localhost",
                    "origin_server_ts": 99999,
                    "type": "m.room.message",
                    "content": {"body": "Other room root", "msgtype": "m.text"},
                },
            ],
        )

        all_recent = await cache.get_recent_room_thread_ids("!room:localhost", limit=10)
        first_only = await cache.get_recent_room_thread_ids("!room:localhost", limit=1)
    finally:
        await cache.close()

    assert all_recent == [
        "$thread_old_root_recent_reply",
        "$thread_recent_root_no_replies",
    ]
    assert first_only == ["$thread_old_root_recent_reply"]


@pytest.mark.asyncio
async def test_get_recent_room_events_warm_path(
    event_cache: ConversationEventCache,
) -> None:
    """Recent room event lookups should filter by room, type, timestamp, and limit."""
    cache = event_cache

    events = [
        (
            "$approval_old",
            "!room:localhost",
            {
                "event_id": "$approval_old",
                "sender": "@bot:localhost",
                "origin_server_ts": 1000,
                "type": "io.mindroom.tool_approval",
                "content": {"approval_id": "old"},
            },
        ),
        (
            "$approval_recent_1",
            "!room:localhost",
            {
                "event_id": "$approval_recent_1",
                "sender": "@bot:localhost",
                "origin_server_ts": 3000,
                "type": "io.mindroom.tool_approval",
                "content": {"approval_id": "recent-1"},
            },
        ),
        (
            "$message_newer",
            "!room:localhost",
            {
                "event_id": "$message_newer",
                "sender": "@user:localhost",
                "origin_server_ts": 5000,
                "type": "m.room.message",
                "content": {"body": "ignore", "msgtype": "m.text"},
            },
        ),
        (
            "$approval_other_room",
            "!other-room:localhost",
            {
                "event_id": "$approval_other_room",
                "sender": "@bot:localhost",
                "origin_server_ts": 6000,
                "type": "io.mindroom.tool_approval",
                "content": {"approval_id": "other-room"},
            },
        ),
        (
            "$approval_recent_2",
            "!room:localhost",
            {
                "event_id": "$approval_recent_2",
                "sender": "@bot:localhost",
                "origin_server_ts": 7000,
                "type": "io.mindroom.tool_approval",
                "content": {"approval_id": "recent-2"},
            },
        ),
    ]

    try:
        await cache.store_events_batch(events)
        all_recent = await cache.get_recent_room_events(
            "!room:localhost",
            event_type="io.mindroom.tool_approval",
            since_ts_ms=2000,
        )
        first_only = await cache.get_recent_room_events(
            "!room:localhost",
            event_type="io.mindroom.tool_approval",
            since_ts_ms=2000,
            limit=1,
        )
    finally:
        await cache.close()

    assert [event["event_id"] for event in all_recent] == ["$approval_recent_2", "$approval_recent_1"]
    assert [event["event_id"] for event in first_only] == ["$approval_recent_2"]


@pytest.mark.asyncio
async def test_event_cache_preserves_insertion_order_for_same_timestamp_events(
    event_cache: ConversationEventCache,
) -> None:
    """Cached reads should preserve the stored order when timestamps tie."""
    cache = event_cache

    try:
        await _seed_thread_cache(
            cache,
            room_id="!room:localhost",
            thread_id="$thread_root",
            events=[
                {
                    "event_id": "$thread_root",
                    "sender": "@user:localhost",
                    "origin_server_ts": 1000,
                    "type": "m.room.message",
                    "content": {"body": "Root message", "msgtype": "m.text"},
                },
                {
                    "event_id": "$zzz_parent",
                    "sender": "@user:localhost",
                    "origin_server_ts": 2000,
                    "type": "m.room.message",
                    "content": {
                        "body": "Parent",
                        "msgtype": "m.text",
                        "m.relates_to": {"m.in_reply_to": {"event_id": "$thread_root"}},
                    },
                },
                {
                    "event_id": "$aaa_child",
                    "sender": "@user:localhost",
                    "origin_server_ts": 2000,
                    "type": "m.room.message",
                    "content": {
                        "body": "Child",
                        "msgtype": "m.text",
                        "m.relates_to": {"m.in_reply_to": {"event_id": "$zzz_parent"}},
                    },
                },
            ],
        )

        cached_events = await cache.get_thread_events("!room:localhost", "$thread_root")
    finally:
        await cache.close()

    assert cached_events is not None
    assert [event["event_id"] for event in cached_events] == [
        "$thread_root",
        "$zzz_parent",
        "$aaa_child",
    ]


@pytest.mark.asyncio
async def test_individual_event_cache_store_and_retrieve(event_cache: ConversationEventCache) -> None:
    """Individually cached events should round-trip by event ID."""
    cache = event_cache

    try:
        await cache.store_events_batch(
            [
                (
                    "$reply",
                    "!room:localhost",
                    {
                        "event_id": "$reply",
                        "sender": "@agent:localhost",
                        "origin_server_ts": 2000,
                        "type": "m.room.message",
                        "content": {"body": "Reply in thread", "msgtype": "m.text"},
                    },
                ),
            ],
        )

        cached_event = await cache.get_event("!room:localhost", "$reply")
        missing_event = await cache.get_event("!room:localhost", "$missing")
    finally:
        await cache.close()

    assert cached_event is not None
    assert cached_event["event_id"] == "$reply"
    assert cached_event["content"]["body"] == "Reply in thread"
    assert missing_event is None


def _clear_payload(
    event_id: str,
    *,
    body: str = "clear",
    thread_root_id: str | None = None,
    edit_of: str | None = None,
    origin_server_ts: int = 1000,
) -> dict[str, object]:
    content: dict[str, object] = {"body": body, "msgtype": "m.text"}
    if thread_root_id is not None:
        content["m.relates_to"] = {"rel_type": "m.thread", "event_id": thread_root_id}
    if edit_of is not None:
        content["body"] = f"* {body}"
        content["m.new_content"] = {"body": body, "msgtype": "m.text"}
        content["m.relates_to"] = {"rel_type": "m.replace", "event_id": edit_of}
    return {
        "event_id": event_id,
        "sender": "@user:localhost",
        "origin_server_ts": origin_server_ts,
        "type": "m.room.message",
        "content": content,
    }


def _opaque_payload(
    event_id: str,
    *,
    thread_root_id: str | None = None,
    origin_server_ts: int = 1000,
) -> dict[str, object]:
    content: dict[str, object] = {
        "algorithm": "m.megolm.v1.aes-sha2",
        "ciphertext": "opaque ciphertext",
        "device_id": "DEVICE",
        "sender_key": "sender-key",
        "session_id": "session",
    }
    if thread_root_id is not None:
        content["m.relates_to"] = {"rel_type": "m.thread", "event_id": thread_root_id}
    return {
        "event_id": event_id,
        "sender": "@user:localhost",
        "origin_server_ts": origin_server_ts,
        "type": "m.room.encrypted",
        "content": content,
    }


@pytest.mark.asyncio
@pytest.mark.parametrize("arrival_order", [("clear", "opaque"), ("opaque", "clear")])
async def test_point_payload_upgrade_is_monotonic_across_arrival_orders(
    event_cache: ConversationEventCache,
    arrival_order: tuple[str, str],
) -> None:
    """Opaque ciphertext must never replace a decrypted point payload in either arrival order.

    The divergent thread roots make index derivation observable: a refused payload must
    contribute no thread index rows, so the index always describes the accepted payload.
    """
    room_id = "!room:localhost"
    event_id = "$mixed:localhost"
    payloads = {
        "clear": _clear_payload(event_id, body="decrypted", thread_root_id="$clear-root:localhost"),
        "opaque": _opaque_payload(event_id, thread_root_id="$opaque-root:localhost"),
    }

    for payload_kind in arrival_order:
        await event_cache.store_event(event_id, room_id, payloads[payload_kind])

    cached_event = await event_cache.get_event(room_id, event_id)
    assert cached_event is not None
    assert cached_event["type"] == "m.room.message"
    assert cached_event["content"]["body"] == "decrypted"
    assert await event_cache.get_thread_id_for_event(room_id, event_id) == "$clear-root:localhost"
    assert await event_cache.get_thread_id_for_event(room_id, "$clear-root:localhost") == "$clear-root:localhost"
    if arrival_order == ("clear", "opaque"):
        assert await event_cache.get_thread_id_for_event(room_id, "$opaque-root:localhost") is None


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "batch_order",
    [
        ("clear", "opaque"),
        ("opaque", "clear"),
        ("opaque", "clear", "opaque"),
    ],
)
async def test_duplicate_ids_in_one_batch_converge_on_clear_payload(
    event_cache: ConversationEventCache,
    batch_order: tuple[str, ...],
) -> None:
    """Duplicate event IDs inside one batch must converge on the decrypted payload."""
    room_id = "!room:localhost"
    event_id = "$duplicated:localhost"
    thread_root_id = "$root:localhost"
    payloads = {
        "clear": _clear_payload(event_id, body="decrypted", thread_root_id=thread_root_id),
        "opaque": _opaque_payload(event_id, thread_root_id=thread_root_id),
    }

    await event_cache.store_events_batch(
        [(event_id, room_id, payloads[payload_kind]) for payload_kind in batch_order],
    )

    cached_event = await event_cache.get_event(room_id, event_id)
    assert cached_event is not None
    assert cached_event["type"] == "m.room.message"
    assert cached_event["content"]["body"] == "decrypted"
    assert await event_cache.get_thread_id_for_event(room_id, event_id) == thread_root_id
    assert await event_cache.get_thread_id_for_event(room_id, thread_root_id) == thread_root_id


@pytest.mark.asyncio
async def test_chained_thread_relations_keep_the_middle_event_as_its_own_root(
    event_cache: ConversationEventCache,
) -> None:
    """An event that is both a thread child and another event's root maps to itself.

    The derived index rows repeat that middle event ID under two different thread IDs: once as a
    child of the outer root, and once as the self row every learned root gets. A batched upsert has
    to collapse the repeat to the row the sequential write left behind, because
    ``ON CONFLICT DO UPDATE`` cannot touch the same row twice in one statement.
    """
    room_id = "!room:localhost"
    child_id = "$child:localhost"
    middle_id = "$middle:localhost"
    outer_id = "$outer:localhost"

    await event_cache.store_events_batch(
        [
            (child_id, room_id, _clear_payload(child_id, body="child", thread_root_id=middle_id)),
            (middle_id, room_id, _clear_payload(middle_id, body="middle", thread_root_id=outer_id)),
        ],
    )

    assert await event_cache.get_thread_id_for_event(room_id, child_id) == middle_id
    assert await event_cache.get_thread_id_for_event(room_id, middle_id) == middle_id
    assert await event_cache.get_thread_id_for_event(room_id, outer_id) == outer_id


@pytest.mark.asyncio
async def test_repeated_edit_ids_in_one_batch_keep_the_last_edit_index_row(
    event_cache: ConversationEventCache,
) -> None:
    """A batch naming one edit event twice indexes that edit once, keeping the last payload.

    Both occurrences are accepted -- clear content never loses to clear content -- so the derived
    edit-index rows repeat the same ``edit_event_id``, which a batched upsert has to collapse.
    """
    room_id = "!room:localhost"
    original_id = "$original:localhost"
    edit_id = "$edit:localhost"

    await event_cache.store_events_batch(
        [(original_id, room_id, _clear_payload(original_id, body="original"))],
    )
    await event_cache.store_events_batch(
        [
            (edit_id, room_id, _clear_payload(edit_id, body="first edit", edit_of=original_id)),
            (edit_id, room_id, _clear_payload(edit_id, body="second edit", edit_of=original_id)),
        ],
    )

    latest_edit = await event_cache.get_latest_edit(room_id, original_id)
    assert latest_edit is not None
    assert latest_edit["event_id"] == edit_id
    assert latest_edit["content"]["m.new_content"]["body"] == "second edit"


@pytest.mark.asyncio
async def test_one_write_settles_proven_and_unproven_thread_roots_together(
    event_cache: ConversationEventCache,
) -> None:
    """Re-parenting the only child proves the new root and unproves the old one in one write."""
    room_id = "!room:localhost"
    child_id = "$child:localhost"
    old_root_id = "$old-root:localhost"
    new_root_id = "$new-root:localhost"

    await event_cache.store_events_batch(
        [(child_id, room_id, _clear_payload(child_id, body="first", thread_root_id=old_root_id))],
    )
    assert await event_cache.get_thread_id_for_event(room_id, old_root_id) == old_root_id

    await event_cache.store_events_batch(
        [(child_id, room_id, _clear_payload(child_id, body="reparented", thread_root_id=new_root_id))],
    )

    assert await event_cache.get_thread_id_for_event(room_id, child_id) == new_root_id
    assert await event_cache.get_thread_id_for_event(room_id, new_root_id) == new_root_id
    assert await event_cache.get_thread_id_for_event(room_id, old_root_id) is None


@pytest.mark.asyncio
async def test_repeated_event_in_a_snapshot_keeps_its_last_position_on_every_backend(
    event_cache: ConversationEventCache,
) -> None:
    """A snapshot of ``A, B, A-last`` reads back as ``B, A`` on both backends.

    Membership rows are ordered by ``origin_server_ts`` and then by the sequence value each write
    draws, so when the timestamps tie the write order decides the read order. The sequential loop
    rewrote ``A`` after ``B``, leaving ``A`` newer. A batched upsert that collapsed the repeat to
    ``A``'s *first* position would draw ``A``'s sequence value before ``B``'s and silently reverse
    the pair against SQLite.
    """
    room_id = "!room:localhost"
    thread_id = "$root:localhost"
    first_id = "$a:localhost"
    second_id = "$b:localhost"
    tied_ts = 1000

    await _replace_thread(
        event_cache,
        room_id,
        thread_id,
        [
            _clear_payload(thread_id, body="root", origin_server_ts=tied_ts),
            _clear_payload(first_id, body="a-first", thread_root_id=thread_id, origin_server_ts=tied_ts),
            _clear_payload(second_id, body="b", thread_root_id=thread_id, origin_server_ts=tied_ts),
            _clear_payload(first_id, body="a-last", thread_root_id=thread_id, origin_server_ts=tied_ts),
        ],
    )

    thread_events = await event_cache.get_thread_events(room_id, thread_id)
    assert thread_events is not None
    assert [event["event_id"] for event in thread_events] == [thread_id, second_id, first_id]
    assert thread_events[-1]["content"]["body"] == "a-last"


@pytest.mark.asyncio
async def test_repeated_event_in_one_thread_snapshot_binds_the_thread_once(
    event_cache: ConversationEventCache,
) -> None:
    """A snapshot naming one event twice binds it to the thread exactly once."""
    room_id = "!room:localhost"
    thread_id = "$thread-root:localhost"
    duplicated_id = "$duplicated:localhost"
    root_source = _clear_payload(thread_id, body="root", origin_server_ts=1000)
    reply = _clear_payload(duplicated_id, body="reply", thread_root_id=thread_id, origin_server_ts=1100)

    await _replace_thread(event_cache, room_id, thread_id, [root_source, reply, reply])

    thread_events = await event_cache.get_thread_events(room_id, thread_id)
    assert thread_events is not None
    assert [event["event_id"] for event in thread_events] == [thread_id, duplicated_id]
    assert await event_cache.get_thread_id_for_event(room_id, duplicated_id) == thread_id


@pytest.mark.asyncio
@pytest.mark.parametrize("arrival_order", [("clear", "opaque"), ("opaque", "clear")])
async def test_separate_cache_clients_cannot_downgrade_decrypted_payload(
    event_cache_factory: Callable[[], ConversationEventCache],
    arrival_order: tuple[str, str],
) -> None:
    """Two cache clients on one backing store must converge on the decrypted payload."""
    room_id = "!room:localhost"
    event_id = "$shared:localhost"
    thread_root_id = "$root:localhost"
    decrypting_client = event_cache_factory()
    await decrypting_client.initialize()
    try:
        keyless_client = event_cache_factory()
        await keyless_client.initialize()
        try:
            writers = {"clear": decrypting_client, "opaque": keyless_client}
            payloads = {
                "clear": _clear_payload(event_id, body="decrypted", thread_root_id=thread_root_id),
                "opaque": _opaque_payload(event_id, thread_root_id=thread_root_id),
            }
            for payload_kind in arrival_order:
                await writers[payload_kind].store_event(event_id, room_id, payloads[payload_kind])
            cached_by_decrypting = await decrypting_client.get_event(room_id, event_id)
            cached_by_keyless = await keyless_client.get_event(room_id, event_id)
        finally:
            await keyless_client.close()
    finally:
        await decrypting_client.close()

    for cached_event in (cached_by_decrypting, cached_by_keyless):
        assert cached_event is not None
        assert cached_event["type"] == "m.room.message"
        assert cached_event["content"]["body"] == "decrypted"


@pytest.mark.asyncio
@pytest.mark.parametrize("arrival_order", [("clear", "opaque"), ("opaque", "clear")])
async def test_thread_append_preserves_decrypted_payload_across_arrival_orders(
    event_cache: ConversationEventCache,
    arrival_order: tuple[str, str],
) -> None:
    """Incremental appends must never downgrade an already-decrypted thread snapshot row."""
    room_id = "!room:localhost"
    thread_id = "$root:localhost"
    child_event_id = "$child:localhost"
    await _replace_thread(event_cache, room_id, thread_id, [_clear_payload(thread_id, body="root")])
    payloads = {
        "clear": _clear_payload(
            child_event_id,
            body="decrypted child",
            thread_root_id=thread_id,
            origin_server_ts=2000,
        ),
        "opaque": _opaque_payload(child_event_id, thread_root_id=thread_id, origin_server_ts=2000),
    }

    for payload_kind in arrival_order:
        outcome = await event_cache.apply_thread_mutation_append(
            room_id,
            thread_id,
            payloads[payload_kind],
            append_failed_reason="live_append_failed",
        )
        assert outcome is ThreadAppendOutcome.APPENDED

    thread_events = await event_cache.get_thread_events(room_id, thread_id)
    assert thread_events is not None
    cached_child = next(event for event in thread_events if event["event_id"] == child_event_id)
    assert cached_child["type"] == "m.room.message"
    assert cached_child["content"]["body"] == "decrypted child"
    cached_event = await event_cache.get_event(room_id, child_event_id)
    assert cached_event is not None
    assert cached_event["type"] == "m.room.message"
    assert cached_event["content"]["body"] == "decrypted child"


@pytest.mark.asyncio
async def test_thread_replacement_preserves_decrypted_payload(
    event_cache: ConversationEventCache,
) -> None:
    """A full snapshot replacement must not bypass the clear-payload invariant."""
    room_id = "!room:localhost"
    thread_id = "$root:localhost"
    await _replace_thread(
        event_cache,
        room_id,
        thread_id,
        [_clear_payload(thread_id, body="decrypted root")],
    )

    await _replace_thread(
        event_cache,
        room_id,
        thread_id,
        [_opaque_payload(thread_id)],
    )

    cached_event = await event_cache.get_event(room_id, thread_id)
    thread_events = await event_cache.get_thread_events(room_id, thread_id)
    assert cached_event is not None
    assert cached_event["type"] == "m.room.message"
    assert cached_event["content"]["body"] == "decrypted root"
    assert thread_events is not None
    assert len(thread_events) == 1
    assert thread_events[0]["type"] == "m.room.message"
    assert thread_events[0]["content"]["body"] == "decrypted root"


@pytest.mark.asyncio
async def test_refused_opaque_thread_replacement_preserves_mxc_plaintext(
    event_cache: ConversationEventCache,
) -> None:
    """A refused ciphertext snapshot must retain the clear payload's sidecar ownership."""
    room_id = "!room:localhost"
    thread_id = "$root:localhost"
    mxc_url = "mxc://server/decrypted-sidecar"
    clear_event = _clear_payload(thread_id, body="decrypted root")
    clear_event["content"] = {
        "body": "preview",
        "msgtype": "m.file",
        "url": mxc_url,
        "io.mindroom.long_text": {
            "version": 2,
            "encoding": "matrix_event_content_json",
        },
    }
    await _replace_thread(event_cache, room_id, thread_id, [clear_event])
    assert await event_cache.store_mxc_text(room_id, thread_id, mxc_url, "decrypted sidecar")

    await _replace_thread(
        event_cache,
        room_id,
        thread_id,
        [_opaque_payload(thread_id)],
    )

    assert await event_cache.get_mxc_text(room_id, thread_id, mxc_url) == "decrypted sidecar"


@pytest.mark.asyncio
async def test_refused_opaque_snapshot_still_records_explicit_thread_membership(
    event_cache: ConversationEventCache,
) -> None:
    """Snapshot membership must be indexed even when its opaque payload is refused."""
    room_id = "!room:localhost"
    thread_id = "$root:localhost"
    await event_cache.store_event(
        thread_id,
        room_id,
        _clear_payload(thread_id, body="decrypted root"),
    )
    assert await event_cache.get_thread_id_for_event(room_id, thread_id) is None

    await _replace_thread(
        event_cache,
        room_id,
        thread_id,
        [_opaque_payload(thread_id)],
    )

    cached_event = await event_cache.get_event(room_id, thread_id)
    assert cached_event is not None
    assert cached_event["type"] == "m.room.message"
    assert cached_event["content"]["body"] == "decrypted root"
    assert await event_cache.get_thread_id_for_event(room_id, thread_id) == thread_id


@pytest.mark.asyncio
async def test_refused_opaque_write_keeps_latest_edit_join_readable(
    event_cache: ConversationEventCache,
) -> None:
    """A keyless client's ciphertext for an indexed edit must not corrupt latest-edit reads."""
    room_id = "!room:localhost"
    original_event_id = "$original:localhost"
    edit_event_id = "$edit:localhost"
    await event_cache.store_event(
        original_event_id,
        room_id,
        _clear_payload(original_event_id, body="original"),
    )
    await event_cache.store_event(
        edit_event_id,
        room_id,
        _clear_payload(edit_event_id, body="edited", edit_of=original_event_id, origin_server_ts=2000),
    )

    await event_cache.store_event(edit_event_id, room_id, _opaque_payload(edit_event_id, origin_server_ts=2000))

    latest_edit = await event_cache.get_latest_edit(room_id, original_event_id)
    assert latest_edit is not None
    assert latest_edit["type"] == "m.room.message"
    assert latest_edit["content"]["m.new_content"]["body"] == "edited"


@pytest.mark.asyncio
async def test_redaction_tombstone_survives_clear_and_opaque_rewrites(
    event_cache: ConversationEventCache,
) -> None:
    """The monotonic upsert must not resurrect durably redacted events for any payload quality."""
    room_id = "!room:localhost"
    event_id = "$redacted:localhost"
    await event_cache.store_event(event_id, room_id, _clear_payload(event_id))
    assert await event_cache.redact_event(room_id, event_id)

    await event_cache.store_event(event_id, room_id, _opaque_payload(event_id))
    await event_cache.store_event(event_id, room_id, _clear_payload(event_id))

    assert await event_cache.get_event(room_id, event_id) is None


@pytest.mark.asyncio
async def test_accepted_clear_rewrite_still_moves_thread_index_row(
    event_cache: ConversationEventCache,
) -> None:
    """Accepted clear rewrites must keep last-wins thread index moves working."""
    room_id = "!room:localhost"
    event_id = "$moved:localhost"
    await event_cache.store_events_batch(
        [(event_id, room_id, _clear_payload(event_id, thread_root_id="$root-a:localhost"))],
    )
    assert await event_cache.get_thread_id_for_event(room_id, event_id) == "$root-a:localhost"

    await event_cache.store_events_batch(
        [(event_id, room_id, _clear_payload(event_id, thread_root_id="$root-b:localhost"))],
    )

    assert await event_cache.get_thread_id_for_event(room_id, event_id) == "$root-b:localhost"


@pytest.mark.asyncio
async def test_opaque_payload_remains_retained_and_refreshable(
    event_cache: ConversationEventCache,
) -> None:
    """Opaque events must stay retained and refreshable until clear content improves them."""
    room_id = "!room:localhost"
    event_id = "$opaque-only:localhost"
    thread_root_id = "$root:localhost"
    await event_cache.store_event(event_id, room_id, _opaque_payload(event_id, thread_root_id=thread_root_id))

    cached_event = await event_cache.get_event(room_id, event_id)
    assert cached_event is not None
    assert cached_event["type"] == "m.room.encrypted"
    assert await event_cache.get_thread_id_for_event(room_id, event_id) == thread_root_id
    assert await event_cache.get_thread_id_for_event(room_id, thread_root_id) == thread_root_id

    await event_cache.store_event(event_id, room_id, _opaque_payload(event_id, thread_root_id=thread_root_id))

    refreshed_event = await event_cache.get_event(room_id, event_id)
    assert refreshed_event is not None
    assert refreshed_event["type"] == "m.room.encrypted"


@pytest.mark.asyncio
async def test_event_cache_close_waits_for_in_flight_operation(tmp_path: Path) -> None:
    """Closing the cache should wait for active DB work instead of closing mid-query."""
    cache = SqliteEventCache(tmp_path / "event_cache.db")
    await cache.initialize()
    await cache.store_event(
        "$reply",
        "!room:localhost",
        {
            "event_id": "$reply",
            "sender": "@agent:localhost",
            "origin_server_ts": 2000,
            "type": "m.room.message",
            "content": {"body": "Cached reply", "msgtype": "m.text"},
        },
    )
    operation_started = asyncio.Event()
    allow_operation_finish = asyncio.Event()
    original_load_event = sqlite_event_cache_events.load_event

    async def blocking_load_event(
        db: object,
        *,
        principal_id: str,
        room_id: str,
        event_id: str,
    ) -> dict[str, object] | None:
        operation_started.set()
        await allow_operation_finish.wait()
        return await original_load_event(
            db,
            principal_id=principal_id,
            room_id=room_id,
            event_id=event_id,
        )

    try:
        with patch(
            "mindroom.matrix.cache.sqlite_event_cache_events.load_event",
            new=blocking_load_event,
        ):
            get_task = asyncio.create_task(cache.get_event("!room:localhost", "$reply"))
            await asyncio.wait_for(operation_started.wait(), timeout=1.0)

            close_task = asyncio.create_task(cache.close())
            await asyncio.sleep(0)
            assert close_task.done() is False

            allow_operation_finish.set()
            cached_event = await get_task
            await close_task
    finally:
        if cache.is_initialized:
            await cache.close()

    assert cached_event is not None
    assert cached_event["event_id"] == "$reply"
    assert cache.is_initialized is False


@pytest.mark.asyncio
async def test_event_cache_initialize_clears_half_initialized_connection_on_failure(tmp_path: Path) -> None:
    """Mid-init failures must close and clear the SQLite connection so a later retry can recover."""
    cache = SqliteEventCache(tmp_path / "event_cache.db")
    broken_connection = AsyncMock()
    broken_connection.close = AsyncMock()
    broken_connection.execute = AsyncMock(side_effect=[MagicMock(), RuntimeError("pragma boom")])

    with (
        patch(
            "mindroom.matrix.cache.sqlite_event_cache.aiosqlite.connect",
            AsyncMock(return_value=broken_connection),
        ),
        pytest.raises(RuntimeError, match="pragma boom"),
    ):
        await cache.initialize()

    broken_connection.close.assert_awaited_once()
    assert cache.is_initialized is False


@pytest.mark.asyncio
async def test_individual_event_cache_strips_runtime_timing_marker(event_cache: ConversationEventCache) -> None:
    """Batch event caching should drop in-memory timing objects before serialization."""
    cache = event_cache

    reply_event = _make_text_event(
        event_id="$reply",
        sender="@agent:localhost",
        body="Cached reply",
        server_timestamp=2000,
        source_content={"body": "Cached reply"},
    )
    event_source = _cache_source(reply_event)
    event_source["com.mindroom.dispatch_pipeline_timing"] = DispatchPipelineTiming(
        source_event_id="$reply",
        room_id="!room:localhost",
    )

    try:
        await cache.store_events_batch([("$reply", "!room:localhost", event_source)])
        cached_event = await cache.get_event("!room:localhost", "$reply")
    finally:
        await cache.close()

    assert cached_event is not None
    assert cached_event["event_id"] == "$reply"
    assert "com.mindroom.dispatch_pipeline_timing" not in cached_event


@pytest.mark.asyncio
async def test_thread_cache_store_populates_individual_event_lookup(event_cache: ConversationEventCache) -> None:
    """Thread cache writes should also populate the individual event table."""
    cache = event_cache

    root_event = _make_text_event(
        event_id="$thread_root",
        sender="@user:localhost",
        body="Root message",
        server_timestamp=1000,
        source_content={"body": "Root message"},
    )
    reply_event = _make_text_event(
        event_id="$reply",
        sender="@agent:localhost",
        body="Cached reply",
        server_timestamp=2000,
        source_content={
            "body": "Cached reply",
            "m.relates_to": {"rel_type": "m.thread", "event_id": "$thread_root"},
        },
    )

    try:
        await _seed_thread_cache(
            cache,
            room_id="!room:localhost",
            thread_id="$thread_root",
            events=[_cache_source(root_event), _cache_source(reply_event)],
        )
        cached_event = await cache.get_event("!room:localhost", "$reply")
    finally:
        await cache.close()

    assert cached_event is not None
    assert cached_event["event_id"] == "$reply"
    assert cached_event["content"]["body"] == "Cached reply"


@pytest.mark.asyncio
async def test_thread_event_cache_strips_runtime_timing_marker(event_cache: ConversationEventCache) -> None:
    """Thread cache writes should strip runtime-only timing markers before JSON storage."""
    cache = event_cache

    reply_event = _make_text_event(
        event_id="$reply",
        sender="@agent:localhost",
        body="Reply in thread",
        server_timestamp=2000,
        source_content={
            "body": "Reply in thread",
            "m.relates_to": {"rel_type": "m.thread", "event_id": "$thread_root"},
        },
    )
    event_source = _cache_source(reply_event)
    event_source["com.mindroom.dispatch_pipeline_timing"] = DispatchPipelineTiming(
        source_event_id="$reply",
        room_id="!room:localhost",
    )

    try:
        await _seed_thread_cache(
            cache,
            room_id="!room:localhost",
            thread_id="$thread_root",
            events=[event_source],
        )
        cached_event = await cache.get_event("!room:localhost", "$reply")
        cached_thread_events = await cache.get_thread_events("!room:localhost", "$thread_root")
    finally:
        await cache.close()

    assert cached_event is not None
    assert "com.mindroom.dispatch_pipeline_timing" not in cached_event
    assert cached_thread_events is not None
    assert "com.mindroom.dispatch_pipeline_timing" not in cached_thread_events[0]


@pytest.mark.asyncio
async def test_cached_room_get_event_cache_hit_avoids_network_call(event_cache: ConversationEventCache) -> None:
    """Cached room get event lookups should reconstruct nio responses without I/O."""
    cache = event_cache

    reply_event = _make_text_event(
        event_id="$reply",
        sender="@agent:localhost",
        body="Cached reply",
        server_timestamp=2000,
        source_content={"body": "Cached reply"},
    )
    client = MagicMock()
    client.room_get_event = AsyncMock()

    try:
        await cache.store_event("$reply", "!room:localhost", _cache_source(reply_event))
        response, _ = await _cached_room_get_event(client, cache, "!room:localhost", "$reply")
    finally:
        await cache.close()

    assert isinstance(response, nio.RoomGetEventResponse)
    assert response.event.event_id == "$reply"
    assert response.event.body == "Cached reply"
    client.room_get_event.assert_not_awaited()


@pytest.mark.asyncio
async def test_matrix_conversation_lookup_fill_cannot_cross_leave_and_rejoin(tmp_path: Path) -> None:
    """A point fetch begun before departure must not repopulate the rejoined cache."""
    db_path = tmp_path / "event_cache.db"
    principal_id = "@alice:localhost"
    room_id = "!room:localhost"
    event_id = "$lookup"
    lookup_root = SqliteEventCache(db_path)
    membership_root = SqliteEventCache(db_path)
    await lookup_root.initialize()
    await membership_root.initialize()
    lookup_cache = lookup_root.for_principal(principal_id)
    membership_cache = membership_root.for_principal(principal_id)
    event = _make_text_event(
        event_id=event_id,
        sender="@agent:localhost",
        body="Fetched",
        server_timestamp=1,
        source_content={"body": "Fetched"},
    )
    fetch_started = asyncio.Event()
    release_fetch = asyncio.Event()

    async def room_get_event(_room_id: str, _event_id: str) -> MagicMock:
        fetch_started.set()
        await release_fetch.wait()
        return _make_room_get_event_response(event)

    client = MagicMock()
    client.room_get_event = AsyncMock(side_effect=room_get_event)
    conversation_cache = _conversation_cache_for_thread_reads(tmp_path, lookup_cache, client=client)
    conversation_cache.runtime.event_cache_write_coordinator = EventCacheWriteCoordinator(
        logger=MagicMock(),
        background_task_owner=conversation_cache.runtime,
    )
    lookup_task = asyncio.create_task(conversation_cache.get_event(room_id, event_id))
    try:
        await fetch_started.wait()
        departure_epoch = membership_cache.mark_room_departed(room_id)
        await membership_cache.purge_room(room_id)
        await membership_cache.mark_room_joined(
            room_id,
            expected_departure_epoch=departure_epoch,
        )
        release_fetch.set()

        response = await lookup_task
        assert isinstance(response, nio.RoomGetEventResponse)
        assert await lookup_cache.get_event(room_id, event_id) is None
    finally:
        release_fetch.set()
        if not lookup_task.done():
            await lookup_task
        await membership_root.close()
        await lookup_root.close()


@pytest.mark.asyncio
async def test_cached_room_get_event_cache_hit_returns_latest_visible_edit(
    event_cache: ConversationEventCache,
) -> None:
    """Point-event cache hits should surface the latest edited content for originals."""
    cache = event_cache

    original_event = _make_text_event(
        event_id="$reply",
        sender="@agent:localhost",
        body="Original reply",
        server_timestamp=2000,
        source_content={
            "body": "Original reply",
            "m.relates_to": {"rel_type": "m.thread", "event_id": "$thread_root"},
        },
    )
    edit_event = _make_text_event(
        event_id="$reply_edit",
        sender="@agent:localhost",
        body="* Final reply",
        server_timestamp=3000,
        source_content={
            "body": "* Final reply",
            "m.new_content": {"body": "Final reply", "msgtype": "m.text"},
            "m.relates_to": {"rel_type": "m.replace", "event_id": "$reply"},
        },
    )
    client = MagicMock()
    client.room_get_event = AsyncMock()

    try:
        await cache.store_events_batch(
            [
                ("$reply", "!room:localhost", _cache_source(original_event)),
                ("$reply_edit", "!room:localhost", _cache_source(edit_event)),
            ],
        )
        response, _ = await _cached_room_get_event(client, cache, "!room:localhost", "$reply")
    finally:
        await cache.close()

    assert isinstance(response, nio.RoomGetEventResponse)
    assert response.event.event_id == "$reply"
    assert response.event.body == "Final reply"
    assert response.event.server_timestamp == 3000
    assert EventInfo.from_event(response.event.source).thread_id == "$thread_root"
    client.room_get_event.assert_not_awaited()


@pytest.mark.asyncio
async def test_edit_cache_row_indexes_io_mindroom_tool_approval_edits(
    event_cache: ConversationEventCache,
) -> None:
    """Custom approval-card edits must be visible through the latest-edit index."""
    cache = event_cache

    approval_card = {
        "event_id": "$approval",
        "sender": "@bot:localhost",
        "origin_server_ts": 1000,
        "type": "io.mindroom.tool_approval",
        "content": {
            "approval_id": "approval-1",
            "requester_id": "@user:localhost",
            "status": "pending",
            "tool_name": "read_file",
        },
    }
    approval_edit = {
        "event_id": "$approval_edit",
        "sender": "@bot:localhost",
        "origin_server_ts": 2000,
        "type": "io.mindroom.tool_approval",
        "content": {
            "approval_id": "approval-1",
            "requester_id": "@user:localhost",
            "status": "approved",
            "tool_name": "read_file",
            "m.new_content": {
                "approval_id": "approval-1",
                "requester_id": "@user:localhost",
                "status": "approved",
                "tool_name": "read_file",
            },
            "m.relates_to": {"rel_type": "m.replace", "event_id": "$approval"},
        },
    }

    try:
        await cache.store_events_batch(
            [
                ("$approval", "!room:localhost", approval_card),
                ("$approval_edit", "!room:localhost", approval_edit),
            ],
        )
        latest_edit = await cache.get_latest_edit("!room:localhost", "$approval")
    finally:
        await cache.close()

    assert latest_edit is not None
    assert latest_edit["event_id"] == "$approval_edit"
    assert latest_edit["content"]["m.new_content"]["status"] == "approved"


@pytest.mark.asyncio
async def test_latest_edit_can_be_scoped_to_sender_when_newer_edit_is_untrusted(
    event_cache: ConversationEventCache,
) -> None:
    """Approval lookup should be able to ignore newer edits from other senders."""
    cache = event_cache

    approval_card = {
        "event_id": "$approval",
        "sender": "@bot:localhost",
        "origin_server_ts": 1000,
        "type": "io.mindroom.tool_approval",
        "content": {
            "approval_id": "approval-1",
            "requester_id": "@user:localhost",
            "status": "pending",
            "tool_name": "read_file",
        },
    }
    trusted_edit = {
        "event_id": "$trusted_edit",
        "sender": "@bot:localhost",
        "origin_server_ts": 2000,
        "type": "io.mindroom.tool_approval",
        "content": {
            "approval_id": "approval-1",
            "requester_id": "@user:localhost",
            "status": "approved",
            "tool_name": "read_file",
            "m.new_content": {
                "approval_id": "approval-1",
                "requester_id": "@user:localhost",
                "status": "approved",
                "tool_name": "read_file",
            },
            "m.relates_to": {"rel_type": "m.replace", "event_id": "$approval"},
        },
    }
    untrusted_edit = {
        "event_id": "$untrusted_edit",
        "sender": "@attacker:localhost",
        "origin_server_ts": 3000,
        "type": "io.mindroom.tool_approval",
        "content": {
            "approval_id": "approval-1",
            "requester_id": "@user:localhost",
            "status": "denied",
            "tool_name": "read_file",
            "m.new_content": {
                "approval_id": "approval-1",
                "requester_id": "@user:localhost",
                "status": "denied",
                "tool_name": "read_file",
            },
            "m.relates_to": {"rel_type": "m.replace", "event_id": "$approval"},
        },
    }

    try:
        await cache.store_events_batch(
            [
                ("$approval", "!room:localhost", approval_card),
                ("$trusted_edit", "!room:localhost", trusted_edit),
                ("$untrusted_edit", "!room:localhost", untrusted_edit),
            ],
        )
        latest_edit = await cache.get_latest_edit("!room:localhost", "$approval")
        latest_trusted_edit = await cache.get_latest_edit("!room:localhost", "$approval", sender="@bot:localhost")
    finally:
        await cache.close()

    assert latest_edit is not None
    assert latest_edit["event_id"] == "$untrusted_edit"
    assert latest_trusted_edit is not None
    assert latest_trusted_edit["event_id"] == "$trusted_edit"


@pytest.mark.asyncio
async def test_cached_room_get_event_network_fetch_merges_cached_latest_edit(
    event_cache: ConversationEventCache,
) -> None:
    """Network fetches should still project originals through cached latest edits."""
    cache = event_cache

    original_event = _make_text_event(
        event_id="$reply",
        sender="@agent:localhost",
        body="Original reply",
        server_timestamp=2000,
        source_content={"body": "Original reply"},
    )
    edit_event = _make_text_event(
        event_id="$reply_edit",
        sender="@agent:localhost",
        body="* Final reply",
        server_timestamp=3000,
        source_content={
            "body": "* Final reply",
            "m.new_content": {"body": "Final reply", "msgtype": "m.text"},
            "m.relates_to": {"rel_type": "m.replace", "event_id": "$reply"},
        },
    )
    client = MagicMock()
    client.room_get_event = AsyncMock(return_value=_make_room_get_event_response(original_event))

    try:
        await cache.store_event("$reply_edit", "!room:localhost", _cache_source(edit_event))
        response, _ = await _cached_room_get_event(client, cache, "!room:localhost", "$reply")
    finally:
        await cache.close()

    assert isinstance(response, nio.RoomGetEventResponse)
    assert response.event.event_id == "$reply"
    assert response.event.body == "Final reply"
    client.room_get_event.assert_awaited_once_with("!room:localhost", "$reply")


@pytest.mark.asyncio
async def test_redacting_latest_edit_falls_back_to_previous_cached_edit(event_cache: ConversationEventCache) -> None:
    """Removing the newest edit should expose the previous cached visible state."""
    cache = event_cache

    original_event = _make_text_event(
        event_id="$reply",
        sender="@agent:localhost",
        body="Original reply",
        server_timestamp=1000,
        source_content={"body": "Original reply"},
    )
    older_edit = _make_text_event(
        event_id="$reply_edit_1",
        sender="@agent:localhost",
        body="* Intermediate reply",
        server_timestamp=2000,
        source_content={
            "body": "* Intermediate reply",
            "m.new_content": {"body": "Intermediate reply", "msgtype": "m.text"},
            "m.relates_to": {"rel_type": "m.replace", "event_id": "$reply"},
        },
    )
    newer_edit = _make_text_event(
        event_id="$reply_edit_2",
        sender="@agent:localhost",
        body="* Final reply",
        server_timestamp=3000,
        source_content={
            "body": "* Final reply",
            "m.new_content": {"body": "Final reply", "msgtype": "m.text"},
            "m.relates_to": {"rel_type": "m.replace", "event_id": "$reply"},
        },
    )
    client = MagicMock()
    client.room_get_event = AsyncMock()

    try:
        await cache.store_events_batch(
            [
                ("$reply", "!room:localhost", _cache_source(original_event)),
                ("$reply_edit_1", "!room:localhost", _cache_source(older_edit)),
                ("$reply_edit_2", "!room:localhost", _cache_source(newer_edit)),
            ],
        )
        latest_response, _ = await _cached_room_get_event(client, cache, "!room:localhost", "$reply")
        redacted = await cache.redact_event("!room:localhost", "$reply_edit_2")
        fallback_response, _ = await _cached_room_get_event(client, cache, "!room:localhost", "$reply")
    finally:
        await cache.close()

    assert redacted is True
    assert isinstance(latest_response, nio.RoomGetEventResponse)
    assert latest_response.event.body == "Final reply"
    assert isinstance(fallback_response, nio.RoomGetEventResponse)
    assert fallback_response.event.body == "Intermediate reply"
    client.room_get_event.assert_not_awaited()


@pytest.mark.asyncio
async def test_redaction_removes_individual_event_cache_entry(event_cache: ConversationEventCache) -> None:
    """Redactions should also remove individually cached events."""
    cache = event_cache

    root_event = _make_text_event(
        event_id="$thread_root",
        sender="@user:localhost",
        body="Root message",
        server_timestamp=1000,
        source_content={"body": "Root message"},
    )
    reply_event = _make_text_event(
        event_id="$reply",
        sender="@agent:localhost",
        body="Cached reply",
        server_timestamp=2000,
        source_content={
            "body": "Cached reply",
            "m.relates_to": {"rel_type": "m.thread", "event_id": "$thread_root"},
        },
    )
    try:
        await _seed_thread_cache(
            cache,
            room_id="!room:localhost",
            thread_id="$thread_root",
            events=[_cache_source(root_event), _cache_source(reply_event)],
        )
        assert await cache.get_event("!room:localhost", "$reply") is not None
        redacted = await cache.redact_event("!room:localhost", "$reply")
        cached_event = await cache.get_event("!room:localhost", "$reply")
    finally:
        await cache.close()

    assert redacted is True
    assert cached_event is None


@pytest.mark.asyncio
async def test_redacting_original_removes_dependent_cached_edits_from_thread_history(
    event_cache: ConversationEventCache,
) -> None:
    """Redacting an original must also remove cached edits that would resurrect it."""
    cache = event_cache

    root_event = _make_text_event(
        event_id="$thread_root",
        sender="@user:localhost",
        body="Root message",
        server_timestamp=1000,
        source_content={"body": "Root message"},
    )
    original_event = _make_text_event(
        event_id="$reply",
        sender="@agent:localhost",
        body="Original reply",
        server_timestamp=2000,
        source_content={
            "body": "Original reply",
            "m.relates_to": {"rel_type": "m.thread", "event_id": "$thread_root"},
        },
    )
    edit_event = _make_text_event(
        event_id="$reply_edit",
        sender="@agent:localhost",
        body="* Final reply",
        server_timestamp=3000,
        source_content={
            "body": "* Final reply",
            "m.new_content": {
                "body": "Final reply",
                "msgtype": "m.text",
                "m.relates_to": {"rel_type": "m.thread", "event_id": "$thread_root"},
            },
            "m.relates_to": {"rel_type": "m.replace", "event_id": "$reply"},
        },
    )
    client = MagicMock()
    client.room_get_event = AsyncMock()
    client.room_messages = AsyncMock(return_value=nio.RoomMessagesResponse([], None, None, None))
    client.room_get_event_relations = MagicMock()

    try:
        await _seed_thread_cache(
            cache,
            room_id="!room:localhost",
            thread_id="$thread_root",
            events=[_cache_source(root_event), _cache_source(original_event), _cache_source(edit_event)],
        )
        history_before = await fetch_thread_history(client, "!room:localhost", "$thread_root", event_cache=cache)

        redacted = await cache.redact_event("!room:localhost", "$reply")
        latest_edit = await cache.get_latest_edit("!room:localhost", "$reply")
        cached_edit = await cache.get_event("!room:localhost", "$reply_edit")
        history_after = await fetch_thread_history(client, "!room:localhost", "$thread_root", event_cache=cache)
    finally:
        await cache.close()

    assert redacted is True
    assert [(message.event_id, message.body) for message in history_before] == [
        ("$thread_root", "Root message"),
        ("$reply", "Final reply"),
    ]
    assert latest_edit is None
    assert cached_edit is None
    assert [(message.event_id, message.body) for message in history_after] == [
        ("$thread_root", "Root message"),
    ]


@pytest.mark.asyncio
async def test_invalidate_thread_preserves_separately_cached_latest_edit(
    event_cache: ConversationEventCache,
) -> None:
    """Thread invalidation should not sever edit projection for separately cached edits."""
    cache = event_cache

    root_event = _make_text_event(
        event_id="$thread_root",
        sender="@user:localhost",
        body="Root message",
        server_timestamp=1000,
        source_content={"body": "Root message"},
    )
    original_event = _make_text_event(
        event_id="$reply",
        sender="@agent:localhost",
        body="Original reply",
        server_timestamp=2000,
        source_content={
            "body": "Original reply",
            "m.relates_to": {"rel_type": "m.thread", "event_id": "$thread_root"},
        },
    )
    edit_event = _make_text_event(
        event_id="$reply_edit",
        sender="@agent:localhost",
        body="* Final reply",
        server_timestamp=3000,
        source_content={
            "body": "* Final reply",
            "m.new_content": {"body": "Final reply", "msgtype": "m.text"},
            "m.relates_to": {"rel_type": "m.replace", "event_id": "$reply"},
        },
    )
    client = MagicMock()
    client.room_get_event = AsyncMock(return_value=_make_room_get_event_response(original_event))

    try:
        await _seed_thread_cache(
            cache,
            room_id="!room:localhost",
            thread_id="$thread_root",
            events=[_cache_source(root_event), _cache_source(original_event)],
        )
        await cache.store_event("$reply_edit", "!room:localhost", _cache_source(edit_event))
        await cache.invalidate_thread("!room:localhost", "$thread_root")

        latest_edit = await cache.get_latest_edit("!room:localhost", "$reply")
        response, _ = await _cached_room_get_event(client, cache, "!room:localhost", "$reply")
    finally:
        await cache.close()

    assert latest_edit is not None
    assert latest_edit["event_id"] == "$reply_edit"
    assert isinstance(response, nio.RoomGetEventResponse)
    assert response.event.body == "Final reply"
    client.room_get_event.assert_awaited_once_with("!room:localhost", "$reply")


@pytest.mark.asyncio
async def test_invalidate_thread_removes_event_thread_rows(event_cache: ConversationEventCache) -> None:
    """Thread invalidation must also clear durable event-to-thread mappings."""
    cache = event_cache

    root_event = _make_text_event(
        event_id="$thread_root",
        sender="@user:localhost",
        body="Root message",
        server_timestamp=1000,
        source_content={"body": "Root message"},
    )
    reply_event = _make_text_event(
        event_id="$reply",
        sender="@agent:localhost",
        body="Reply",
        server_timestamp=2000,
        source_content={
            "body": "Reply",
            "m.relates_to": {"rel_type": "m.thread", "event_id": "$thread_root"},
        },
    )

    try:
        await _seed_thread_cache(
            cache,
            room_id="!room:localhost",
            thread_id="$thread_root",
            events=[_cache_source(root_event), _cache_source(reply_event)],
        )
        assert await cache.get_thread_id_for_event("!room:localhost", "$reply") == "$thread_root"

        await cache.invalidate_thread("!room:localhost", "$thread_root")
        thread_id = await cache.get_thread_id_for_event("!room:localhost", "$reply")
    finally:
        await cache.close()

    assert thread_id is None


@pytest.mark.asyncio
async def test_redaction_removes_event_thread_rows_and_blocks_late_edit_resurrection(
    event_cache: ConversationEventCache,
) -> None:
    """Redacting a reply must clear durable thread mapping and ignore late edits for that reply."""
    cache = event_cache

    root_event = _make_text_event(
        event_id="$thread_root",
        sender="@user:localhost",
        body="Root message",
        server_timestamp=1000,
        source_content={"body": "Root message"},
    )
    reply_event = _make_text_event(
        event_id="$reply",
        sender="@agent:localhost",
        body="Reply",
        server_timestamp=2000,
        source_content={
            "body": "Reply",
            "m.relates_to": {"rel_type": "m.thread", "event_id": "$thread_root"},
        },
    )
    late_edit_event = _make_text_event(
        event_id="$reply_edit",
        sender="@agent:localhost",
        body="* Reply edited",
        server_timestamp=3000,
        source_content={
            "body": "* Reply edited",
            "m.new_content": {
                "body": "Reply edited",
                "msgtype": "m.text",
                "m.relates_to": {"rel_type": "m.thread", "event_id": "$thread_root"},
            },
            "m.relates_to": {"rel_type": "m.replace", "event_id": "$reply"},
        },
    )
    client = MagicMock()
    client.room_get_event = AsyncMock()
    client.room_messages = AsyncMock(return_value=nio.RoomMessagesResponse([], None, None, None))
    client.room_get_event_relations = MagicMock()

    try:
        await _seed_thread_cache(
            cache,
            room_id="!room:localhost",
            thread_id="$thread_root",
            events=[_cache_source(root_event), _cache_source(reply_event)],
        )
        assert await cache.get_thread_id_for_event("!room:localhost", "$reply") == "$thread_root"

        redacted = await cache.redact_event("!room:localhost", "$reply")
        await cache.store_events_batch([("$reply_edit", "!room:localhost", _cache_source(late_edit_event))])

        thread_id = await cache.get_thread_id_for_event("!room:localhost", "$reply")
        cached_late_edit = await cache.get_event("!room:localhost", "$reply_edit")
        history = await fetch_thread_history(client, "!room:localhost", "$thread_root", event_cache=cache)
    finally:
        await cache.close()

    assert redacted is True
    assert thread_id is None
    assert cached_late_edit is None
    assert [(message.event_id, message.body) for message in history] == [
        ("$thread_root", "Root message"),
    ]


@pytest.mark.asyncio
async def test_store_events_batch_records_thread_root_self_mapping_from_explicit_thread_child(
    event_cache: ConversationEventCache,
) -> None:
    """Explicit threaded children should also make the root resolve to its own thread id."""
    cache = event_cache

    reply_event = _make_text_event(
        event_id="$reply",
        sender="@user:localhost",
        body="Reply",
        server_timestamp=2000,
        source_content={
            "body": "Reply",
            "m.relates_to": {"rel_type": "m.thread", "event_id": "$thread_root"},
        },
    )

    try:
        await cache.store_events_batch([("$reply", "!room:localhost", _cache_source(reply_event))])
        reply_thread_id = await cache.get_thread_id_for_event("!room:localhost", "$reply")
        root_thread_id = await cache.get_thread_id_for_event("!room:localhost", "$thread_root")
    finally:
        await cache.close()

    assert reply_thread_id == "$thread_root"
    assert root_thread_id == "$thread_root"


@pytest.mark.asyncio
async def test_store_events_batch_rolls_back_on_index_derivation_failure(
    event_cache: ConversationEventCache,
) -> None:
    """Failed batch writes must not leak partial point-lookup rows into later commits."""
    cache = event_cache

    valid_event = _cache_source(
        _make_text_event(
            event_id="$reply",
            sender="@agent:localhost",
            body="Reply",
            server_timestamp=2000,
            source_content={"body": "Reply"},
        ),
    )
    invalid_edit_event = {
        "event_id": "$reply_edit",
        "sender": "@agent:localhost",
        "type": "m.room.message",
        "content": {
            "body": "* Reply edited",
            "m.new_content": {"body": "Reply edited", "msgtype": "m.text"},
            "m.relates_to": {"rel_type": "m.replace", "event_id": "$reply"},
        },
    }
    later_event = _cache_source(
        _make_text_event(
            event_id="$later",
            sender="@agent:localhost",
            body="Later",
            server_timestamp=4000,
            source_content={"body": "Later"},
        ),
    )

    try:
        with pytest.raises(ValueError, match="origin_server_ts"):
            await cache.store_events_batch(
                [
                    ("$reply", "!room:localhost", valid_event),
                    ("$reply_edit", "!room:localhost", invalid_edit_event),
                ],
            )

        await cache.store_events_batch([("$later", "!room:localhost", later_event)])
        cached_reply = await cache.get_event("!room:localhost", "$reply")
        cached_invalid_edit = await cache.get_event("!room:localhost", "$reply_edit")
        cached_later = await cache.get_event("!room:localhost", "$later")
    finally:
        await cache.close()

    assert cached_reply is None
    assert cached_invalid_edit is None
    assert cached_later is not None
    assert cached_later["event_id"] == "$later"


@pytest.mark.asyncio
async def test_initialize_resets_stale_old_cache_schema(tmp_path: Path) -> None:
    """Initialization should discard stale cache DBs instead of migrating them forward."""
    db_path = tmp_path / "event_cache.db"
    original_event = _cache_source(
        _make_text_event(
            event_id="$reply",
            sender="@agent:localhost",
            body="Original reply",
            server_timestamp=2000,
            source_content={"body": "Original reply"},
        ),
    )
    edit_event = _cache_source(
        _make_text_event(
            event_id="$reply_edit",
            sender="@agent:localhost",
            body="* Final reply",
            server_timestamp=3000,
            source_content={
                "body": "* Final reply",
                "m.new_content": {"body": "Final reply", "msgtype": "m.text"},
                "m.relates_to": {"rel_type": "m.replace", "event_id": "$reply"},
            },
        ),
    )

    with closing(sqlite3.connect(db_path)) as db:
        db.execute(
            """
            CREATE TABLE events (
                event_id TEXT PRIMARY KEY,
                room_id TEXT NOT NULL,
                event_json TEXT NOT NULL,
                cached_at REAL NOT NULL
            )
            """,
        )
        db.executemany(
            """
            INSERT INTO events(event_id, room_id, event_json, cached_at)
            VALUES (?, ?, ?, ?)
            """,
            [
                ("$reply", "!room:localhost", json.dumps(original_event, separators=(",", ":")), 1.0),
                ("$reply_edit", "!room:localhost", json.dumps(edit_event, separators=(",", ":")), 1.0),
            ],
        )
        db.commit()

    cache = SqliteEventCache(db_path)
    await cache.initialize()
    try:
        latest_edit = await cache.get_latest_edit("!room:localhost", "$reply")
        cached_original = await cache.get_event("!room:localhost", "$reply")
    finally:
        await cache.close()

    with closing(sqlite3.connect(db_path)) as db:
        schema_version = db.execute("PRAGMA user_version").fetchone()[0]

    assert latest_edit is None
    assert cached_original is None
    assert schema_version == event_cache_module._EVENT_CACHE_SCHEMA_VERSION


@pytest.mark.asyncio
async def test_disabled_event_cache_skips_latest_agent_message_snapshot_reads(
    event_cache: ConversationEventCache,
) -> None:
    """Disabled caches should fail open for latest-agent-message snapshot reads."""
    cache = event_cache
    try:
        await cache.store_events_batch(
            [
                (
                    "$reply",
                    "!room:localhost",
                    {
                        "event_id": "$reply",
                        "sender": "@agent:localhost",
                        "origin_server_ts": 2000,
                        "type": "m.room.message",
                        "content": {"body": "Working...", "msgtype": "m.text"},
                    },
                ),
            ],
        )
        cache.disable("test_disabled")

        snapshot = await cache.get_latest_agent_message_snapshot(
            "!room:localhost",
            None,
            "@agent:localhost",
            runtime_started_at=0.0,
        )
    finally:
        await cache.close()

    assert snapshot is None


@pytest.mark.asyncio
async def test_fetch_thread_history_cache_hit_avoids_full_fetch_calls(event_cache: ConversationEventCache) -> None:
    """Cache hits should bypass the full root-plus-relations fetch path."""
    cache = event_cache

    root_event = _make_text_event(
        event_id="$thread_root",
        sender="@user:localhost",
        body="Root message",
        server_timestamp=1000,
        source_content={"body": "Root message"},
    )
    reply_event = _make_text_event(
        event_id="$reply",
        sender="@agent:localhost",
        body="Cached reply",
        server_timestamp=2000,
        source_content={
            "body": "Cached reply",
            "m.relates_to": {"rel_type": "m.thread", "event_id": "$thread_root"},
        },
    )
    await _seed_thread_cache(
        cache,
        room_id="!room:localhost",
        thread_id="$thread_root",
        events=[_cache_source(root_event), _cache_source(reply_event)],
    )

    client = MagicMock()
    incremental_page = MagicMock(spec=nio.RoomMessagesResponse)
    incremental_page.chunk = [reply_event, root_event]
    incremental_page.end = None
    client.room_messages = AsyncMock(return_value=incremental_page)
    client.room_get_event = AsyncMock()
    client.room_get_event_relations = MagicMock()

    try:
        history = await fetch_thread_history(client, "!room:localhost", "$thread_root", event_cache=cache)
    finally:
        await cache.close()

    assert [message.event_id for message in history] == ["$thread_root", "$reply"]
    client.room_get_event.assert_not_awaited()
    client.room_get_event_relations.assert_not_called()


@pytest.mark.asyncio
async def test_fetch_thread_history_cache_miss_does_full_fetch(event_cache: ConversationEventCache) -> None:
    """Cache misses should scan room history and populate the cache."""
    cache = event_cache

    root_event = _make_text_event(
        event_id="$thread_root",
        sender="@user:localhost",
        body="Root message",
        server_timestamp=1000,
        source_content={"body": "Root message"},
    )
    reply_event = _make_text_event(
        event_id="$reply",
        sender="@agent:localhost",
        body="Reply in thread",
        server_timestamp=2000,
        source_content={
            "body": "Reply in thread",
            "m.relates_to": {"rel_type": "m.thread", "event_id": "$thread_root"},
        },
    )
    client = _make_relations_client(
        root_event=root_event,
        relations={
            _relation_key("$thread_root", RelationshipType.thread): [reply_event],
            _relation_key("$thread_root", RelationshipType.replacement): [],
            _relation_key("$reply", RelationshipType.replacement): [],
        },
    )

    try:
        history = await fetch_thread_history(client, "!room:localhost", "$thread_root", event_cache=cache)
        cached_events = await cache.get_thread_events("!room:localhost", "$thread_root")
    finally:
        await cache.close()

    assert [message.event_id for message in history] == ["$thread_root", "$reply"]
    assert cached_events is not None
    assert [event["event_id"] for event in cached_events] == ["$thread_root", "$reply"]
    client.room_get_event.assert_not_awaited()
    client.room_messages.assert_awaited_once()


@pytest.mark.asyncio
async def test_mxc_text_cache_round_trips_across_event_cache_reopen(
    event_cache_factory: Callable[[], ConversationEventCache],
) -> None:
    """Durable MXC text rows should survive closing and reopening the event cache."""
    cache = event_cache_factory()
    await cache.initialize()
    owner_event = {
        "event_id": "$sidecar-owner",
        "origin_server_ts": 1000,
        "type": "m.room.message",
        "sender": "@agent:localhost",
        "content": {
            "body": "preview",
            "msgtype": "m.file",
            "url": "mxc://server/sidecar",
            "io.mindroom.long_text": {
                "version": 2,
                "encoding": "matrix_event_content_json",
            },
        },
    }

    try:
        await cache.store_event("$sidecar-owner", "!room:localhost", owner_event)
        assert await cache.store_mxc_text(
            "!room:localhost",
            "$sidecar-owner",
            "mxc://server/sidecar",
            "Full text sidecar",
        )
    finally:
        await cache.close()

    reopened_cache = event_cache_factory()
    await reopened_cache.initialize()
    try:
        cached_text = await reopened_cache.get_mxc_text(
            "!room:localhost",
            "$sidecar-owner",
            "mxc://server/sidecar",
        )
    finally:
        await reopened_cache.close()

    assert cached_text == "Full text sidecar"


@pytest.mark.asyncio
async def test_fetch_thread_history_reuses_durable_mxc_text_after_restart(
    event_cache_factory: Callable[[], ConversationEventCache],
) -> None:
    """Cached full-history reads should reuse durable sidecar text after a restart."""
    cache = event_cache_factory()
    await cache.initialize()

    root_event = _make_text_event(
        event_id="$thread_root",
        sender="@user:localhost",
        body="Root message",
        server_timestamp=1000,
        source_content={"body": "Root message"},
    )
    sidecar_reply = _make_text_event(
        event_id="$reply",
        sender="@agent:localhost",
        body="Preview reply",
        server_timestamp=2000,
        source_content={
            "body": "Preview reply",
            "msgtype": "m.file",
            "io.mindroom.long_text": {
                "version": 2,
                "encoding": "matrix_event_content_json",
            },
            "url": "mxc://server/sidecar",
            "m.relates_to": {"rel_type": "m.thread", "event_id": "$thread_root"},
        },
    )
    canonical_sidecar_content = {"body": "Full reply", "msgtype": "m.text"}

    first_client = MagicMock()
    first_client.download = AsyncMock(
        return_value=MagicMock(
            spec=nio.DownloadResponse,
            body=json.dumps(canonical_sidecar_content).encode("utf-8"),
        ),
    )
    first_client.room_get_event = AsyncMock()
    first_client.room_messages = AsyncMock()
    first_client.room_get_event_relations = MagicMock()

    try:
        await _seed_thread_cache(
            cache,
            room_id="!room:localhost",
            thread_id="$thread_root",
            events=[_cache_source(root_event), _cache_source(sidecar_reply)],
        )

        first_history = await fetch_thread_history(
            first_client,
            "!room:localhost",
            "$thread_root",
            event_cache=cache,
        )
    finally:
        await cache.close()

    reopened_cache = event_cache_factory()
    await reopened_cache.initialize()
    second_client = MagicMock()
    second_client.download = AsyncMock(
        return_value=MagicMock(spec=nio.DownloadError),
    )
    second_client.room_get_event = AsyncMock()
    second_client.room_messages = AsyncMock()
    second_client.room_get_event_relations = MagicMock()

    try:
        second_history = await fetch_thread_history(
            second_client,
            "!room:localhost",
            "$thread_root",
            event_cache=reopened_cache,
        )
    finally:
        await reopened_cache.close()

    assert [message.body for message in first_history] == ["Root message", "Full reply"]
    assert [message.body for message in second_history] == ["Root message", "Full reply"]
    first_client.download.assert_awaited_once_with(mxc="mxc://server/sidecar")
    second_client.download.assert_not_awaited()
