"""Facade for Matrix conversation reads and advisory cache notifications.

``MatrixConversationCache`` is the facade for conversation reads and advisory thread bookkeeping; it
composes the read policy (``cache.thread_reads``), the three write policies (``cache.thread_writes``),
and the mutation resolver (``thread_bookkeeping``) over one shared write coordinator.
Bots and tools still read the event cache directly for non-thread point lookups such as agent message
snapshots and recent room events, but all thread reads and thread bookkeeping go through this facade.

Per-turn memoization covers event lookups only. Thread reads are not memoized: the saving was one
re-read per turn, and paying for it meant every caller reasoning about whether a degraded or stale
read might be replayed later in the same turn.
"""

from __future__ import annotations

import asyncio
import time
from contextlib import asynccontextmanager
from contextvars import ContextVar
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any, Protocol, cast

import nio
from nio.responses import RoomGetEventError

from mindroom.background_tasks import create_background_task
from mindroom.entity_resolution import current_internal_sender_ids
from mindroom.logging_config import get_logger
from mindroom.matrix.cache import (
    ConversationEventCache,
    ThreadHistoryResult,
    normalize_nio_event_for_cache,
)
from mindroom.matrix.cache.thread_reads import ThreadReadMode, ThreadReadPolicy
from mindroom.matrix.cache.thread_write_cache_ops import ThreadMutationCacheOps
from mindroom.matrix.cache.thread_writes import ThreadLiveWritePolicy, ThreadOutboundWritePolicy, ThreadSyncWritePolicy
from mindroom.matrix.client_thread_history import (
    BulkThreadRefreshStats,
    bulk_refresh_room_thread_histories,
    fetch_dispatch_thread_history,
    fetch_dispatch_thread_snapshot,
    fetch_thread_history,
    get_room_threads_page,
    log_thread_history_refresh,
    refresh_thread_history_from_source,
    thread_ids_needing_refill,
)
from mindroom.matrix.event_info import EventInfo
from mindroom.matrix.media import (
    is_encrypted_media_event_source,
    parse_matrix_media_event_source,
)
from mindroom.matrix.membership_fence import UNCERTIFIED_MEMBERSHIP_EPOCH
from mindroom.matrix.message_content import extract_edit_body
from mindroom.matrix.thread_bookkeeping import ThreadMutationResolver
from mindroom.matrix.thread_membership import resolve_event_thread_membership
from mindroom.matrix.thread_room_scan import (
    fetch_event_info_for_client,
    lookup_thread_id_from_conversation_cache,
    room_scan_membership_access_for_client,
)
from mindroom.timing import elapsed_ms_since

if TYPE_CHECKING:
    from collections.abc import AsyncIterator, Awaitable, Callable, Collection, Coroutine, Mapping
    from contextlib import AbstractAsyncContextManager

    import structlog

    from mindroom.bot_runtime_view import BotRuntimeView
    from mindroom.matrix.sync_certification import SyncCacheWriteResult


type ThreadReadResult = ThreadHistoryResult
type EventLookupResult = nio.RoomGetEventResponse | RoomGetEventError
type _TurnEventCacheKey = tuple[str, str, int]

logger = get_logger(__name__)


__all__ = [
    "ConversationCacheProtocol",
    "ConversationEventCache",
    "EventLookupResult",
    "MatrixConversationCache",
    "ThreadReadResult",
    "resolve_thread_root_event_id_for_client",
]


_STARTUP_PREWARM_THREAD_LIMIT = 32
_STARTUP_PREWARM_MAX_SCAN_PAGES = 20


async def resolve_thread_root_event_id_for_client(
    client: nio.AsyncClient,
    room_id: str,
    event_id: str,
    *,
    conversation_cache: ConversationCacheProtocol | None = None,
) -> str | None:
    """Resolve one event ID into a canonical thread root when thread membership can prove one."""
    normalized_event_id = event_id.strip() if isinstance(event_id, str) else ""
    if not normalized_event_id:
        return None

    event_info = await fetch_event_info_for_client(
        client,
        room_id,
        normalized_event_id,
        strict=False,
    )
    if event_info is None:
        return await lookup_thread_id_from_conversation_cache(
            conversation_cache,
            room_id,
            normalized_event_id,
        )

    resolution = await resolve_event_thread_membership(
        room_id,
        event_info,
        event_id=normalized_event_id,
        allow_current_root=True,
        access=room_scan_membership_access_for_client(
            client,
            conversation_cache=conversation_cache,
            fetch_event_info=lambda lookup_room_id, lookup_event_id: fetch_event_info_for_client(
                client,
                lookup_room_id,
                lookup_event_id,
                strict=False,
            ),
        ),
    )
    return resolution.thread_id


class ConversationCacheProtocol(Protocol):
    """Conversation-data reads available to resolver and related callers."""

    def turn_scope(self) -> AbstractAsyncContextManager[None]:
        """Provide per-turn memoization for event lookups."""

    async def get_event(self, room_id: str, event_id: str) -> EventLookupResult:
        """Resolve one Matrix event by ID."""

    async def get_thread_history(
        self,
        room_id: str,
        thread_id: str,
        *,
        caller_label: str = "unknown",
    ) -> ThreadReadResult:
        """Resolve advisory full thread history for one conversation root."""

    async def get_dispatch_thread_snapshot(
        self,
        room_id: str,
        thread_id: str,
        *,
        caller_label: str = "unknown",
    ) -> ThreadReadResult:
        """Resolve strict dispatch thread context using only fresh cache data or a homeserver refill."""

    async def get_dispatch_thread_history(
        self,
        room_id: str,
        thread_id: str,
        *,
        caller_label: str = "unknown",
    ) -> ThreadReadResult:
        """Resolve strict full dispatch thread history using only fresh cache data or a homeserver refill."""

    async def get_strict_thread_history(
        self,
        room_id: str,
        thread_id: str,
        *,
        caller_label: str = "unknown",
    ) -> ThreadReadResult:
        """Resolve strict full thread history without live dispatch timeouts or stale fallback."""

    async def refresh_strict_thread_history_from_source(
        self,
        room_id: str,
        thread_id: str,
        *,
        caller_label: str = "unknown",
    ) -> ThreadReadResult:
        """Refresh strict full history directly from Matrix."""

    async def refresh_startup_thread_history_from_source(
        self,
        room_id: str,
        thread_id: str,
        *,
        caller_label: str = "unknown",
    ) -> ThreadReadResult:
        """Refresh startup-only strict history without entering live cache coordination."""

    async def get_thread_id_for_event(self, room_id: str, event_id: str) -> str | None:
        """Resolve the cached thread root for one event when known."""

    async def purge_rooms(self, room_ids: Collection[str]) -> None:
        """Fence and purge one authoritative batch of departed rooms."""

    async def get_latest_thread_event_id_if_needed(
        self,
        room_id: str,
        thread_id: str | None,
        reply_to_event_id: str | None = None,
        existing_event_id: str | None = None,
        *,
        caller_label: str = "latest_thread_event_lookup",
    ) -> str | None:
        """Resolve the latest visible thread event when MSC3440 fallback needs it."""

    def notify_outbound_message(
        self,
        room_id: str,
        event_id: str | None,
        content: dict[str, Any],
    ) -> None:
        """Schedule one locally sent message or edit for advisory cache bookkeeping.

        This is advisory post-send bookkeeping and must fail open.
        Callers should treat Matrix delivery as complete before this local cache work runs.
        """

    def notify_outbound_event(self, room_id: str, event_source: dict[str, Any]) -> None:
        """Schedule one locally sent outbound event for advisory cache bookkeeping."""

    def notify_outbound_redaction(self, room_id: str, redacted_event_id: str) -> None:
        """Schedule one locally redacted message for advisory cache bookkeeping.

        This is advisory post-redaction bookkeeping and must fail open.
        """

    def reserve_outbound_thread(self, room_id: str, event_id: str, thread_id: str) -> None:
        """Reserve one known outbound response thread for later relation-free edits."""

    def release_outbound_thread(self, room_id: str, event_id: str) -> None:
        """Release one outbound response thread reservation after terminal delivery."""

    async def append_live_event(
        self,
        room_id: str,
        event: nio.RoomMessage,
        *,
        event_info: EventInfo,
    ) -> None:
        """Append one live threaded event into the advisory cache when the thread is known."""


async def _apply_cached_latest_edit(
    event_source: dict[str, Any],
    *,
    room_id: str,
    client: nio.AsyncClient,
    event_cache: ConversationEventCache,
    expected_membership_epoch: int | None = None,
    trusted_sender_ids: Collection[str] = (),
) -> dict[str, Any]:
    """Project one cached original event into its latest visible edited state."""
    if event_source.get("type") != "m.room.message":
        return event_source

    event_info = EventInfo.from_event(event_source)
    event_id = event_source.get("event_id")
    sender = event_source.get("sender")
    if event_info.is_edit or not isinstance(event_id, str) or not event_id or not isinstance(sender, str):
        return event_source

    # Scoped to this event's own sender. A replacement is only legitimate from the sender of the
    # event it replaces, and without this the newest edit from anyone wins - so this path would
    # serve someone else's text under the author's event, while the collapsed thread read of the
    # same cache correctly refuses it.
    latest_edit_source = await event_cache.get_latest_edit(room_id, event_id, sender=sender)
    if latest_edit_source is None:
        return event_source

    edited_body, edited_content = await extract_edit_body(
        latest_edit_source,
        client,
        event_cache=event_cache,
        room_id=room_id,
        expected_membership_epoch=expected_membership_epoch,
        trusted_sender_ids=trusted_sender_ids,
    )
    if edited_body is None or edited_content is None:
        return event_source

    original_content = event_source.get("content", {})
    merged_content = (
        {key: value for key, value in original_content.items() if isinstance(key, str)}
        if isinstance(original_content, dict)
        else {}
    )
    merged_content.update(edited_content)
    merged_content.setdefault("body", edited_body)

    updated_event_source = {key: value for key, value in event_source.items() if isinstance(key, str)}
    updated_event_source["content"] = merged_content

    latest_edit_timestamp = latest_edit_source.get("origin_server_ts")
    if isinstance(latest_edit_timestamp, int) and not isinstance(latest_edit_timestamp, bool):
        updated_event_source["origin_server_ts"] = latest_edit_timestamp
    return updated_event_source


async def _cached_room_get_event_response(
    client: nio.AsyncClient,
    event_cache: ConversationEventCache,
    *,
    room_id: str,
    event_source: dict[str, Any],
    expected_membership_epoch: int | None = None,
    trusted_sender_ids: Collection[str] = (),
) -> nio.RoomGetEventResponse | None:
    """Reconstruct one cached room-get-event response, applying visible edits when present."""
    visible_event_source = await _apply_cached_latest_edit(
        event_source,
        room_id=room_id,
        client=client,
        event_cache=event_cache,
        expected_membership_epoch=expected_membership_epoch,
        trusted_sender_ids=trusted_sender_ids,
    )
    if is_encrypted_media_event_source(visible_event_source):
        parsed_media_event = parse_matrix_media_event_source(visible_event_source)
        if parsed_media_event is None:
            return None
        cached_response = nio.RoomGetEventResponse()
        # nio's response parser also assigns BadEvent to this Event-typed field.
        cached_response.event = cast("nio.Event", parsed_media_event)
        return cached_response
    cached_response = nio.RoomGetEventResponse.from_dict(visible_event_source)
    return cached_response if isinstance(cached_response, nio.RoomGetEventResponse) else None


async def _cached_room_get_event(
    client: nio.AsyncClient,
    event_cache: ConversationEventCache,
    room_id: str,
    event_id: str,
    *,
    expected_membership_epoch: int | None = None,
    trusted_sender_ids: Collection[str] = (),
) -> tuple[nio.RoomGetEventResponse | RoomGetEventError, dict[str, Any] | None]:
    """Return one event through the persistent cache when available."""
    normalized_event_id = event_id.strip()
    if normalized_event_id:
        try:
            cached_event = await event_cache.get_event(room_id, normalized_event_id)
        except Exception as exc:
            logger.warning(
                "Failed to read cached Matrix event",
                room_id=room_id,
                event_id=normalized_event_id,
                error=str(exc),
            )
        else:
            if cached_event is not None:
                cached_response = await _cached_room_get_event_response(
                    client,
                    event_cache,
                    room_id=room_id,
                    event_source=cached_event,
                    expected_membership_epoch=expected_membership_epoch,
                    trusted_sender_ids=trusted_sender_ids,
                )
                if cached_response is not None:
                    return cached_response, None
                logger.warning(
                    "Cached Matrix event could not be reconstructed",
                    room_id=room_id,
                    event_id=normalized_event_id,
                    error=str(cached_response),
                )

    response = await client.room_get_event(room_id, normalized_event_id)
    if not isinstance(response, nio.RoomGetEventResponse):
        return response, None

    event = response.event
    normalized_event_source = normalize_nio_event_for_cache(
        event,
        event_id=normalized_event_id,
    )
    visible_response = await _cached_room_get_event_response(
        client,
        event_cache,
        room_id=room_id,
        event_source=normalized_event_source,
        expected_membership_epoch=expected_membership_epoch,
        trusted_sender_ids=trusted_sender_ids,
    )
    return (visible_response if visible_response is not None else response), normalized_event_source


type _ThreadRefillKey = tuple[str, str, str, bool, bool]


@dataclass(frozen=True, slots=True)
class _ThreadRefillSingleFlightResult:
    """One caller's view of a shared thread refill."""

    result: ThreadReadResult
    wait_ms: float
    shared: bool


@dataclass(slots=True)
class _ThreadRefillSingleFlight:
    """Join concurrent refills of one thread onto a single homeserver scan.

    This is all that survives of the deleted repair registry: no tiering, no speculative fan-out
    gate, no cooldown, no failure backoff. Just "one scan per thread in flight at a time", because
    a full room scan costs seconds under load and N readers of one gapped thread would otherwise
    each run their own.

    Keyed by the caller's whole contract rather than the thread alone. A caller that asked for
    hydrated sidecars, or that allows a stale fallback, must not be handed the result of one that
    did not - sharing across those would return history the caller did not ask for.

    Waiters ``shield`` the shared task, so a caller giving up cannot cancel the scan the others are
    still waiting on. The task is owned so shutdown drains it rather than leaving it pending.
    """

    _owner: object
    _in_flight: dict[_ThreadRefillKey, asyncio.Task[ThreadReadResult]] = field(default_factory=dict, init=False)

    async def run(
        self,
        key: _ThreadRefillKey,
        refill: Callable[[], Coroutine[Any, Any, ThreadReadResult]],
    ) -> _ThreadRefillSingleFlightResult:
        """Run one refill, or join the one already running for this key."""
        in_flight = self._in_flight.get(key)
        shared = in_flight is not None
        wait_started = time.perf_counter() if shared else None
        if in_flight is None:
            in_flight = create_background_task(
                refill(),
                name="matrix_cache_thread_refill",
                owner=self._owner,
                log_exceptions=False,
            )
            self._in_flight[key] = in_flight
            in_flight.add_done_callback(lambda _task: self._in_flight.pop(key, None))
        result = await asyncio.shield(in_flight)
        return _ThreadRefillSingleFlightResult(
            result=result,
            wait_ms=elapsed_ms_since(wait_started, clock=time.perf_counter) if wait_started is not None else 0.0,
            shared=shared,
        )


@dataclass
class MatrixConversationCache(ConversationCacheProtocol):
    """Own Matrix conversation reads and advisory cache writes for one bot."""

    logger: structlog.stdlib.BoundLogger
    runtime: BotRuntimeView
    _turn_event_cache: ContextVar[dict[_TurnEventCacheKey, EventLookupResult] | None] = field(
        default_factory=lambda: ContextVar("mindroom_turn_event_lookup_cache", default=None),
    )
    _reads: ThreadReadPolicy = field(init=False, repr=False)
    _refill_single_flight: _ThreadRefillSingleFlight = field(init=False, repr=False)
    _write_cache_ops: ThreadMutationCacheOps = field(init=False, repr=False)
    _outbound: ThreadOutboundWritePolicy = field(init=False, repr=False)
    _live: ThreadLiveWritePolicy = field(init=False, repr=False)
    _sync: ThreadSyncWritePolicy = field(init=False, repr=False)

    def __post_init__(self) -> None:
        """Bind extracted read/write collaborators to this facade."""
        self._reads = ThreadReadPolicy(
            logger_getter=lambda: self.logger,
            runtime=self.runtime,
            fetch_thread_history_from_client=self._fetch_thread_history_from_client,
            fetch_dispatch_thread_history_from_client=self._fetch_dispatch_thread_history_from_client,
            fetch_dispatch_thread_snapshot_from_client=self._fetch_dispatch_thread_snapshot_from_client,
            refresh_thread_history_from_source=self._refresh_thread_history_from_client,
        )
        resolver = ThreadMutationResolver(
            logger_getter=lambda: self.logger,
            runtime=self.runtime,
            fetch_event_info_for_thread_resolution=self._event_info_for_thread_resolution,
        )
        self._write_cache_ops = ThreadMutationCacheOps(
            logger_getter=lambda: self.logger,
            runtime=self.runtime,
        )
        self._refill_single_flight = _ThreadRefillSingleFlight(_owner=self)
        self._outbound = ThreadOutboundWritePolicy(
            resolver=resolver,
            cache_ops=self._write_cache_ops,
            require_client=self._require_client,
        )
        self._live = ThreadLiveWritePolicy(
            resolver=resolver,
            cache_ops=self._write_cache_ops,
        )
        self._sync = ThreadSyncWritePolicy(
            resolver=resolver,
            cache_ops=self._write_cache_ops,
        )

    def _require_client(self) -> nio.AsyncClient:
        client = self.runtime.client
        if client is None:
            msg = "Matrix client is not ready for conversation cache"
            raise RuntimeError(msg)
        return client

    def _trusted_sender_ids(self) -> frozenset[str]:
        """Return the exact internal sender IDs allowed to override canonical visible-body reads."""
        return current_internal_sender_ids(self.runtime.config, self.runtime.runtime_paths)

    @asynccontextmanager
    async def turn_scope(self) -> AsyncIterator[None]:
        """Memoize event lookups for the lifetime of one inbound turn."""
        if self._turn_event_cache.get() is not None:
            yield
            return

        event_token = self._turn_event_cache.set({})
        try:
            yield
        finally:
            self._turn_event_cache.reset(event_token)

    async def get_event(
        self,
        room_id: str,
        event_id: str,
    ) -> EventLookupResult:
        """Resolve one event through per-turn memoization and the advisory cache."""
        normalized_event_id = event_id.strip()
        cache_key: _TurnEventCacheKey = (
            room_id,
            normalized_event_id,
            self._write_cache_ops.room_departure_epoch(room_id),
        )
        turn_cache = self._turn_event_cache.get()
        if turn_cache is not None and cache_key in turn_cache:
            return turn_cache[cache_key]

        coordinator = self.runtime.event_cache_write_coordinator
        if coordinator is not None:
            await coordinator.wait_for_prior_room_updates(
                room_id,
                coordination_scope=self.runtime.event_cache.principal_id,
            )

        membership_epoch = await self._capture_membership_epoch(room_id)
        response, fetched_event_source = await _cached_room_get_event(
            self._require_client(),
            self.runtime.event_cache,
            room_id,
            event_id,
            expected_membership_epoch=membership_epoch,
            trusted_sender_ids=self._trusted_sender_ids(),
        )
        if fetched_event_source is not None:
            await self._persist_lookup_fill(
                room_id=room_id,
                event_id=normalized_event_id,
                fetched_event_source=fetched_event_source,
                expected_membership_epoch=membership_epoch,
                queue_write=coordinator is not None,
            )
        if turn_cache is not None:
            turn_cache[cache_key] = response
        return response

    async def _capture_membership_epoch(self, room_id: str) -> int:
        """Return a durable lookup generation or one that rejects every cache write."""
        try:
            membership_epoch = await self.runtime.event_cache.room_membership_epoch(room_id)
        except Exception as exc:
            self.logger.warning(
                "Failed to certify Matrix lookup cache generation; continuing without cache writes",
                room_id=room_id,
                error=str(exc),
            )
            return UNCERTIFIED_MEMBERSHIP_EPOCH
        return UNCERTIFIED_MEMBERSHIP_EPOCH if membership_epoch is None else membership_epoch

    async def _persist_lookup_fill(
        self,
        *,
        room_id: str,
        event_id: str,
        fetched_event_source: dict[str, Any],
        expected_membership_epoch: int,
        queue_write: bool,
    ) -> None:
        """Persist one point-lookup fill without reintroducing same-room barrier deadlocks."""

        async def persist_lookup_event() -> None:
            await self.runtime.event_cache.store_event(
                event_id,
                room_id,
                fetched_event_source,
                expected_membership_epoch=expected_membership_epoch,
            )

        try:
            if queue_write:
                await self.runtime.event_cache_write_coordinator.queue_room_update(
                    room_id,
                    persist_lookup_event,
                    name="matrix_cache_store_room_get_event",
                    coordination_scope=self.runtime.event_cache.principal_id,
                )
            else:
                await persist_lookup_event()
        except Exception as exc:
            self.logger.warning(
                "Failed to cache Matrix event lookup",
                room_id=room_id,
                event_id=event_id,
                error=str(exc),
            )

    async def _event_info_for_thread_resolution(
        self,
        room_id: str,
        event_id: str,
    ) -> EventInfo | None:
        """Resolve one related event without memoizing pre-mutation state in the active turn."""
        membership_epoch = await self._capture_membership_epoch(room_id)
        response, _fetched_event_source = await _cached_room_get_event(
            self._require_client(),
            self.runtime.event_cache,
            room_id,
            event_id,
            expected_membership_epoch=membership_epoch,
            trusted_sender_ids=self._trusted_sender_ids(),
        )
        if not isinstance(response, nio.RoomGetEventResponse):
            return None
        return EventInfo.from_event(response.event.source)

    async def _fetch_thread_history_from_client(
        self,
        room_id: str,
        thread_id: str,
        *,
        caller_label: str,
        coordinator_queue_wait_ms: float,
    ) -> ThreadHistoryResult:
        return await self._fetch_thread_from_client(
            fetch_thread_history,
            room_id,
            thread_id,
            caller_label=caller_label,
            coordinator_queue_wait_ms=coordinator_queue_wait_ms,
            wants_full_history=True,
            allows_stale_fallback=True,
        )

    async def _fetch_dispatch_thread_history_from_client(
        self,
        room_id: str,
        thread_id: str,
        *,
        caller_label: str,
        coordinator_queue_wait_ms: float,
    ) -> ThreadHistoryResult:
        return await self._fetch_thread_from_client(
            fetch_dispatch_thread_history,
            room_id,
            thread_id,
            caller_label=caller_label,
            coordinator_queue_wait_ms=coordinator_queue_wait_ms,
            wants_full_history=True,
            allows_stale_fallback=False,
        )

    async def _fetch_dispatch_thread_snapshot_from_client(
        self,
        room_id: str,
        thread_id: str,
        *,
        caller_label: str,
        coordinator_queue_wait_ms: float,
    ) -> ThreadHistoryResult:
        return await self._fetch_thread_from_client(
            fetch_dispatch_thread_snapshot,
            room_id,
            thread_id,
            caller_label=caller_label,
            coordinator_queue_wait_ms=coordinator_queue_wait_ms,
            wants_full_history=False,
            allows_stale_fallback=False,
        )

    async def _refresh_thread_history_from_client(
        self,
        room_id: str,
        thread_id: str,
        *,
        caller_label: str,
        coordinator_queue_wait_ms: float,
    ) -> ThreadHistoryResult:
        """Refresh one thread from Matrix without accepting a cache hit or stale fallback."""
        post_coordinator_read_started = time.perf_counter()
        result = await self._refill_thread_from_client(
            room_id,
            thread_id,
            cache_reject_diagnostics=None,
            wants_full_history=True,
            allows_stale_fallback=False,
        )
        log_thread_history_refresh(
            room_id=room_id,
            thread_id=thread_id,
            caller_label=caller_label,
            mode="full_scan",
            diagnostics=result.diagnostics,
            coordinator_queue_wait_ms=coordinator_queue_wait_ms,
            post_coordinator_read_started=post_coordinator_read_started,
        )
        return result

    async def _refill_thread_from_client(
        self,
        room_id: str,
        thread_id: str,
        *,
        cache_reject_diagnostics: Mapping[str, str | int | float | bool] | None,
        wants_full_history: bool,
        allows_stale_fallback: bool,
    ) -> ThreadHistoryResult:
        """Rebuild one thread from Matrix and reinstall its snapshot.

        Matching readers join one single-flight, and its fetch-plus-store owns the same-thread write
        lane. Mutations queued before the refill run first; mutations arriving while it runs wait
        until the snapshot is installed, then extend that snapshot instead of repeatedly re-gapping
        a snapshotless thread.

        The 126 ms figure quoted elsewhere in this subsystem prices a warm *cache* read and does not
        apply here.
        """
        principal_id = self.runtime.event_cache.principal_id

        async def refill() -> ThreadReadResult:
            result = await self._write_cache_ops.queue_thread_cache_update(
                room_id,
                thread_id,
                lambda: refresh_thread_history_from_source(
                    self._require_client(),
                    room_id,
                    thread_id,
                    self.runtime.event_cache,
                    hydrate_sidecars=wants_full_history,
                    allow_stale_fallback=allows_stale_fallback,
                    cache_reject_diagnostics=cache_reject_diagnostics,
                    trusted_sender_ids=self._trusted_sender_ids(),
                ),
                name="matrix_cache_thread_refill",
                emit_timing=True,
            )
            return cast("ThreadReadResult", result)

        refill_result = await self._refill_single_flight.run(
            (principal_id, room_id, thread_id, wants_full_history, allows_stale_fallback),
            refill,
        )
        return ThreadHistoryResult(
            messages=list(refill_result.result),
            is_full_history=refill_result.result.is_full_history,
            diagnostics={
                **refill_result.result.diagnostics,
                "refill_singleflight_wait_ms": refill_result.wait_ms,
                "refill_singleflight_shared": refill_result.shared,
            },
        )

    async def _fetch_thread_from_client(
        self,
        fetcher: Callable[..., Awaitable[ThreadHistoryResult]],
        room_id: str,
        thread_id: str,
        *,
        caller_label: str,
        coordinator_queue_wait_ms: float,
        wants_full_history: bool,
        allows_stale_fallback: bool,
    ) -> ThreadHistoryResult:
        post_coordinator_read_started = time.perf_counter()

        async def refill(
            cache_reject_diagnostics: Mapping[str, str | int | float | bool] | None,
        ) -> ThreadHistoryResult:
            return await self._refill_thread_from_client(
                room_id,
                thread_id,
                cache_reject_diagnostics=cache_reject_diagnostics,
                wants_full_history=wants_full_history,
                allows_stale_fallback=allows_stale_fallback,
            )

        return await fetcher(
            self._require_client(),
            room_id,
            thread_id,
            event_cache=self.runtime.event_cache,
            trusted_sender_ids=self._trusted_sender_ids(),
            caller_label=caller_label,
            coordinator_queue_wait_ms=coordinator_queue_wait_ms,
            post_coordinator_read_started=post_coordinator_read_started,
            refill=refill,
        )

    async def _bulk_refresh_startup_threads(
        self,
        room_id: str,
        thread_ids: Collection[str],
    ) -> BulkThreadRefreshStats:
        """Refresh startup threads without occupying the live write coordinator during the scan."""
        return await bulk_refresh_room_thread_histories(
            self._require_client(),
            room_id,
            self.runtime.event_cache,
            thread_root_ids=thread_ids,
            caller_label="startup_thread_prewarm",
            max_scan_pages=_STARTUP_PREWARM_MAX_SCAN_PAGES,
        )

    def _log_startup_thread_prewarm_complete(
        self,
        room_id: str,
        *,
        started_at: float,
        threads_warmed: int,
        threads_failed: int,
    ) -> None:
        self.logger.info(
            "startup_thread_prewarm_complete",
            room_id=room_id,
            threads_warmed=threads_warmed,
            threads_failed=threads_failed,
            elapsed_ms=elapsed_ms_since(started_at, clock=time.perf_counter),
        )

    async def _startup_thread_prewarm_ids(
        self,
        room_id: str,
    ) -> list[str] | None:
        """Return startup-prewarm thread IDs using local recency first and /threads as a top-up.

        Tuwunel does not currently order /threads by latest thread activity, so the local cache is the
        best available recency signal for startup prewarm. /threads is only used to fill any remaining
        slots when we have fewer than the target number of locally known threads.
        """
        thread_ids = await self.runtime.event_cache.get_recent_room_thread_ids(
            room_id,
            limit=_STARTUP_PREWARM_THREAD_LIMIT,
        )
        if len(thread_ids) >= _STARTUP_PREWARM_THREAD_LIMIT:
            return thread_ids
        try:
            thread_roots, _next_batch = await get_room_threads_page(
                self._require_client(),
                room_id,
                limit=_STARTUP_PREWARM_THREAD_LIMIT,
            )
        except Exception as exc:
            self.logger.warning(
                "startup_thread_prewarm_room_threads_failed",
                room_id=room_id,
                error=str(exc),
                local_thread_count=len(thread_ids),
            )
            # Partial local prewarm is still useful here because /threads is only a best-effort top-up.
            return thread_ids or None

        for thread_root in thread_roots:
            thread_id = thread_root.event_id.strip()
            if thread_id and thread_id not in thread_ids:
                thread_ids.append(thread_id)
            if len(thread_ids) >= _STARTUP_PREWARM_THREAD_LIMIT:
                break
        return thread_ids

    async def prewarm_recent_room_threads(
        self,
        room_id: str,
        *,
        is_shutting_down: Callable[[], bool],
    ) -> bool:
        """Warm one room's recent thread roots and report whether the room-level pass finished."""
        if not self.runtime.event_cache.durable_writes_available:
            self.logger.warning(
                "startup_thread_prewarm_skipped",
                room_id=room_id,
                reason="event_cache_writes_unavailable",
            )
            return False
        started_at = time.perf_counter()
        thread_ids = await self._startup_thread_prewarm_ids(room_id)
        if thread_ids is None or is_shutting_down() or not self.runtime.event_cache.durable_writes_available:
            return False
        try:
            thread_ids_to_refill = await thread_ids_needing_refill(
                self.runtime.event_cache,
                room_id,
                thread_ids,
            )
        except Exception as exc:
            self.logger.warning(
                "startup_thread_prewarm_cache_probe_failed",
                room_id=room_id,
                thread_count=len(thread_ids),
                error=str(exc),
                error_type=type(exc).__name__,
                exc_info=True,
            )
            thread_ids_to_refill = None
        if thread_ids_to_refill is None or is_shutting_down() or not self.runtime.event_cache.durable_writes_available:
            return False
        already_warm = len(thread_ids) - len(thread_ids_to_refill)
        if not thread_ids_to_refill:
            self._log_startup_thread_prewarm_complete(
                room_id,
                started_at=started_at,
                threads_warmed=already_warm,
                threads_failed=0,
            )
            return True

        try:
            stats = await self._bulk_refresh_startup_threads(
                room_id,
                thread_ids_to_refill,
            )
        except Exception as exc:
            self.logger.warning(
                "startup_thread_prewarm_bulk_failed",
                room_id=room_id,
                thread_count=len(thread_ids_to_refill),
                error=str(exc),
            )
            return False

        threads_warmed = already_warm + stats.usable_threads
        threads_failed = len(thread_ids_to_refill) - stats.usable_threads
        self._log_startup_thread_prewarm_complete(
            room_id,
            started_at=started_at,
            threads_warmed=threads_warmed,
            threads_failed=threads_failed,
        )
        return not is_shutting_down() and self.runtime.event_cache.durable_writes_available

    async def get_thread_history(
        self,
        room_id: str,
        thread_id: str,
        *,
        caller_label: str = "unknown",
    ) -> ThreadReadResult:
        """Resolve advisory full thread history for one conversation root."""
        return await self._reads.read_thread(
            room_id,
            thread_id,
            mode=ThreadReadMode.ADVISORY_FULL,
            caller_label=caller_label,
        )

    async def get_dispatch_thread_snapshot(
        self,
        room_id: str,
        thread_id: str,
        *,
        caller_label: str = "unknown",
    ) -> ThreadReadResult:
        """Resolve strict dispatch thread context using only fresh cache data or a homeserver refill."""
        return await self._reads.read_thread(
            room_id,
            thread_id,
            mode=ThreadReadMode.DISPATCH_SNAPSHOT,
            caller_label=caller_label,
        )

    async def get_dispatch_thread_history(
        self,
        room_id: str,
        thread_id: str,
        *,
        caller_label: str = "unknown",
    ) -> ThreadReadResult:
        """Resolve strict full dispatch thread history using only fresh cache data or a homeserver refill."""
        return await self._reads.read_thread(
            room_id,
            thread_id,
            mode=ThreadReadMode.DISPATCH_FULL,
            caller_label=caller_label,
        )

    async def get_strict_thread_history(
        self,
        room_id: str,
        thread_id: str,
        *,
        caller_label: str = "unknown",
    ) -> ThreadReadResult:
        """Resolve strict full thread history without live dispatch timeouts or stale fallback."""
        return await self._reads.read_thread(
            room_id,
            thread_id,
            mode=ThreadReadMode.STRICT_FULL,
            caller_label=caller_label,
        )

    async def refresh_strict_thread_history_from_source(
        self,
        room_id: str,
        thread_id: str,
        *,
        caller_label: str = "unknown",
    ) -> ThreadReadResult:
        """Refresh strict full history directly from Matrix."""
        return await self._reads.read_thread(
            room_id,
            thread_id,
            mode=ThreadReadMode.STRICT_SOURCE_REFRESH,
            caller_label=caller_label,
        )

    async def refresh_startup_thread_history_from_source(
        self,
        room_id: str,
        thread_id: str,
        *,
        caller_label: str = "unknown",
    ) -> ThreadReadResult:
        """Refresh startup-only strict history without entering live cache coordination."""
        return await refresh_thread_history_from_source(
            self._require_client(),
            room_id,
            thread_id,
            self.runtime.event_cache,
            hydrate_sidecars=True,
            allow_stale_fallback=False,
            trusted_sender_ids=self._trusted_sender_ids(),
            caller_label=caller_label,
        )

    async def get_thread_id_for_event(self, room_id: str, event_id: str) -> str | None:
        """Resolve the cached thread root for one event when known."""
        try:
            return await self.runtime.event_cache.get_thread_id_for_event(room_id, event_id)
        except Exception as error:
            logger.warning(
                "Conversation cache thread lookup failed; continuing without cached thread id",
                room_id=room_id,
                event_id=event_id,
                error=str(error),
            )
            return None

    async def get_latest_thread_event_id_if_needed(
        self,
        room_id: str,
        thread_id: str | None,
        reply_to_event_id: str | None = None,
        existing_event_id: str | None = None,
        *,
        caller_label: str = "latest_thread_event_lookup",
    ) -> str | None:
        """Resolve the latest visible thread event when MSC3440 fallback needs it."""
        return await self._reads.get_latest_thread_event_id_if_needed(
            room_id,
            thread_id,
            reply_to_event_id=reply_to_event_id,
            existing_event_id=existing_event_id,
            caller_label=caller_label,
        )

    def notify_outbound_message(
        self,
        room_id: str,
        event_id: str | None,
        content: dict[str, Any],
    ) -> None:
        """Schedule one locally sent message or edit for advisory cache bookkeeping."""
        self._evict_turn_event_lookups_for_outbound_event(
            room_id,
            event_id=event_id,
            event_info=EventInfo.from_event({"type": "m.room.message", "content": content}),
        )
        self._outbound.notify_outbound_message(room_id, event_id, content)

    def notify_outbound_event(
        self,
        room_id: str,
        event_source: dict[str, Any],
    ) -> None:
        """Schedule one locally sent outbound event for advisory cache bookkeeping."""
        event_id = event_source.get("event_id")
        self._evict_turn_event_lookups_for_outbound_event(
            room_id,
            event_id=event_id if isinstance(event_id, str) else None,
            event_info=EventInfo.from_event(event_source),
        )
        self._outbound.notify_outbound_event(room_id, event_source)

    def notify_outbound_redaction(self, room_id: str, redacted_event_id: str) -> None:
        """Schedule one locally redacted message for advisory cache bookkeeping."""
        self._evict_turn_event_lookups_for_room(room_id)
        self._outbound.notify_outbound_redaction(room_id, redacted_event_id)

    def reserve_outbound_thread(self, room_id: str, event_id: str, thread_id: str) -> None:
        """Reserve one known outbound response thread for later relation-free edits."""
        self._outbound.reserve_thread_response(room_id, event_id, thread_id)

    def release_outbound_thread(self, room_id: str, event_id: str) -> None:
        """Release one outbound response thread reservation after terminal delivery."""
        self._outbound.release_thread_response(room_id, event_id)

    def _evict_turn_event_lookup(self, room_id: str, event_id: str) -> None:
        """Discard point-read memoization invalidated by one successful outbound mutation."""
        turn_cache = self._turn_event_cache.get()
        if turn_cache is None:
            return
        normalized_event_id = event_id.strip()
        for cache_key in tuple(turn_cache):
            if cache_key[:2] == (room_id, normalized_event_id):
                turn_cache.pop(cache_key)

    def _evict_turn_event_lookups_for_room(self, room_id: str) -> None:
        """Discard point reads that one outbound redaction could change indirectly."""
        turn_cache = self._turn_event_cache.get()
        if turn_cache is None:
            return
        for cache_key in tuple(turn_cache):
            if cache_key[0] == room_id:
                turn_cache.pop(cache_key)

    def _evict_turn_event_lookups_for_outbound_event(
        self,
        room_id: str,
        *,
        event_id: str | None,
        event_info: EventInfo,
    ) -> None:
        """Discard point reads changed by one outbound event or edit."""
        for changed_event_id in (event_id, event_info.original_event_id):
            if isinstance(changed_event_id, str):
                self._evict_turn_event_lookup(room_id, changed_event_id)

    async def append_live_event(
        self,
        room_id: str,
        event: nio.RoomMessage,
        *,
        event_info: EventInfo,
    ) -> None:
        """Append one live threaded event into the advisory cache when the thread is known."""
        await self._live.append_live_event(room_id, event, event_info=event_info)

    async def apply_redaction(self, room_id: str, event: nio.RedactionEvent) -> None:
        """Apply one redaction to the advisory cache when the affected thread is known."""
        await self._live.apply_redaction(room_id, event)

    async def purge_rooms(self, room_ids: Collection[str]) -> None:
        """Fence an entire authoritative leave batch before awaiting any purge."""
        departed_room_ids = tuple(dict.fromkeys(room_ids))
        for room_id in departed_room_ids:
            self._write_cache_ops.mark_room_departed(room_id)
        tasks = tuple(
            self._write_cache_ops.queue_room_cache_update(
                room_id,
                lambda room_id=room_id: self._write_cache_ops.purge_room(room_id),
                name="matrix_cache_purge_departed_room",
            )
            for room_id in departed_room_ids
        )
        if tasks:
            await asyncio.gather(*tasks)

    async def mark_room_joined(self, room_id: str) -> None:
        """Allow principal-owned caching again after an authoritative rejoin."""
        expected_departure_epoch = self._write_cache_ops.room_departure_epoch(room_id)
        task = self._write_cache_ops.queue_room_cache_update(
            room_id,
            lambda: self._write_cache_ops.mark_room_joined(
                room_id,
                expected_departure_epoch=expected_departure_epoch,
            ),
            name="matrix_cache_mark_room_joined",
        )
        await task

    async def cache_historical_event(
        self,
        room: nio.MatrixRoom,
        event: nio.Event,
    ) -> None:
        """Durably cache one nio-recovered history event before admission."""
        await self._sync.cache_historical_event(room.room_id, event)

    def limited_sync_timeline_room_ids(
        self,
        response: nio.SyncResponse,
    ) -> tuple[tuple[str, ...], tuple[BaseException, ...]]:
        """Return limited joined-room IDs or validation errors for one sync response."""
        return self._sync.limited_sync_timeline_room_ids(response)

    def cache_sync_timeline(
        self,
        response: nio.SyncResponse,
        *,
        raise_on_cache_write_failure: bool = False,
    ) -> list[asyncio.Task[object]]:
        """Queue sync timeline persistence through the room-ordered cache barrier."""
        return self._sync.cache_sync_timeline(
            response,
            raise_on_cache_write_failure=raise_on_cache_write_failure,
        )

    async def cache_sync_timeline_for_certification(
        self,
        response: nio.SyncResponse,
    ) -> SyncCacheWriteResult:
        """Durably persist sync timeline events and report cache-certification status."""
        return await self._sync.cache_sync_timeline_for_certification(response)
