"""Test configuration and fixtures for MindRoom tests."""

import asyncio
import os
import re
import shutil
import subprocess
import sys
import time
import uuid
import warnings
from collections.abc import (
    AsyncGenerator,
    AsyncIterator,
    Awaitable,
    Callable,
    Generator,
    Iterable,
    Iterator,
    Mapping,
    MutableMapping,
    Sequence,
)
from contextlib import ExitStack, contextmanager
from dataclasses import replace
from itertools import count
from pathlib import Path
from typing import TYPE_CHECKING, Any, cast
from unittest.mock import AsyncMock, MagicMock, patch

import nio
import pytest
import pytest_asyncio
import structlog
import yaml
from agno.models.base import Model
from agno.models.response import ModelResponse
from aioresponses import aioresponses
from structlog.testing import ReturnLoggerFactory
from structlog.typing import BindableLogger, Context, Processor, WrappedLogger

import mindroom.bot  # noqa: F401
from mindroom.agent_storage import get_agent_session, get_team_session
from mindroom.ai import ResponseTurnContext
from mindroom.bot import AgentBot, TeamBot
from mindroom.coalescing import CoalescingDrainResult
from mindroom.command_turn_executor import CommandTurnExecutor
from mindroom.config.main import Config, load_config
from mindroom.constants import RuntimePaths, resolve_runtime_paths, safe_replace
from mindroom.conversation_resolver import DispatchContextResult, MessageContext
from mindroom.delivery_gateway import DeliveryGateway, EditTextRequest, FinalDeliveryRequest, SendTextRequest
from mindroom.dispatch_source import ScheduledHistoryBudget
from mindroom.edit_regenerator import EditRegenerator
from mindroom.final_delivery import FinalDeliveryOutcome
from mindroom.history.runtime import (
    ScopeSessionContext,
    _resolve_history_scope,
    finalize_history_preparation,
    open_scope_session_context,
    prepare_scope_history,
    resolve_agent_preparation_inputs,
)
from mindroom.history.types import (
    CompactionLifecycle,
    HistoryScope,
    PreparedHistoryState,
    ResolvedHistoryExecutionPlan,
    ResolvedHistorySettings,
)
from mindroom.hooks import EnrichmentItem, MessageEnvelope
from mindroom.ingress_validation import IngressValidator
from mindroom.interactive import InteractiveMetadata
from mindroom.matrix.cache.sqlite_event_cache import SqliteEventCache
from mindroom.matrix.cache.thread_cache_state import ThreadAppendOutcome
from mindroom.matrix.cache.thread_history_result import thread_history_result
from mindroom.matrix.cache.write_coordinator import EventCacheWriteCoordinator
from mindroom.matrix.client import DeliveredMatrixEvent, ResolvedVisibleMessage
from mindroom.matrix.client_delivery import build_edit_event_content
from mindroom.matrix.conversation_cache import ConversationCacheProtocol
from mindroom.matrix.identity import MatrixID
from mindroom.matrix.thread_diagnostics import is_thread_history_degraded
from mindroom.media_fallback import reset_model_media_capability_cache
from mindroom.message_target import MessageTarget
from mindroom.reaction_dispatch import ReactionDispatcher
from mindroom.response_payload_preparation import (
    DispatchPayloadInputs,
    ResponsePayloadPreparation,
    ResponsePayloadPreparer,
)
from mindroom.response_runner import PostLockRequestPreparationError, ResponseRequest, ResponseRunner
from mindroom.runtime_support import StartupThreadPrewarmRegistry
from mindroom.thread_utils import decide_agent_response
from mindroom.turn_controller import TurnController, _DispatchPreparation, _ReplayGuardContext
from mindroom.turn_origin import TurnOrigin, classify_turn_origin
from mindroom.turn_policy import PreparedDispatch, TurnPolicy
from mindroom.turn_store import TurnStore
from mindroom.user_stop_reconciliation import UserStopReconciler
from mindroom.visible_response_reconciliation import VisibleResponseReconciler
from mindroom.visible_voice_echo import VisibleVoiceEchoLifecycle, _reset_visible_voice_echo_barriers
from tests.identity_helpers import persist_entity_accounts

if TYPE_CHECKING:
    from agno.agent import Agent
    from agno.db.base import BaseDb
    from agno.session.agent import AgentSession
    from agno.session.team import TeamSession
    from xdist.workermanage import WorkerController

    from mindroom.config.models import CompactionConfig
    from mindroom.dispatch_handoff import DispatchEvent
    from mindroom.matrix.cache import ConversationEventCache
    from mindroom.matrix_rtc.call_manager import CallManager
    from mindroom.tool_system.worker_routing import ToolExecutionIdentity


_STRUCTLOG_CONFIGURE = structlog.configure
_POSTGRES_CONTAINER_NAME_STASH_KEY = pytest.StashKey[str]()
_POSTGRES_CONTAINER_PREFIX = "mindroom-postgres-cache-test-"
_POSTGRES_STARTUP_TIMEOUT_SECONDS = 30


def _configure_quiet_structlog() -> None:
    """Keep incidental test logging cheap and silent."""
    _STRUCTLOG_CONFIGURE(
        processors=[],
        context_class=dict,
        logger_factory=ReturnLoggerFactory(),
        wrapper_class=structlog.stdlib.BoundLogger,
        cache_logger_on_first_use=False,
    )


def _configure_uncached_structlog(
    processors: Iterable[Processor] | None = None,
    wrapper_class: type[BindableLogger] | None = None,
    context_class: type[Context] | None = None,
    logger_factory: Callable[..., WrappedLogger] | None = None,
    cache_logger_on_first_use: bool | None = None,
) -> None:
    """Prevent logging tests from leaving cached production renderers behind."""
    # Cached proxies outlive one test, so the suite intentionally overrides this request.
    _ = cache_logger_on_first_use
    _STRUCTLOG_CONFIGURE(
        processors=processors,
        wrapper_class=wrapper_class,
        context_class=context_class,
        logger_factory=logger_factory,
        cache_logger_on_first_use=False,
    )


_configure_quiet_structlog()


__all__ = [
    "TEST_ACCESS_TOKEN",
    "TEST_PASSWORD",
    "FakeCredentialsManager",
    "agent_response_should_respond",
    "aioresponse",
    "bind_mock_config_cache",
    "bind_runtime_paths",
    "build_private_template_dir",
    "bypass_authorization",
    "create_mock_room",
    "delivered_matrix_event",
    "delivered_matrix_side_effect",
    "dispatch_context_result",
    "drain_coalescing",
    "event_cache",
    "event_cache_factory",
    "install_call_manager_mock",
    "install_edit_message_mock",
    "install_generate_response_mock",
    "install_runtime_cache_support",
    "install_send_response_mock",
    "install_shutdown_drain_mocks",
    "load_config_yaml",
    "make_conversation_cache_mock",
    "make_event_cache_mock",
    "make_event_cache_write_coordinator_mock",
    "make_matrix_client_mock",
    "make_visible_message",
    "message_origin",
    "normalize_console_output",
    "orchestrator_runtime_paths",
    "patch_response_runner_module",
    "postgres_event_cache_url",
    "prepare_history_for_run_for_test",
    "prepare_payload_via_seam",
    "prepared_dispatch_result",
    "replace_delivery_gateway_deps",
    "replace_edit_regenerator_deps",
    "replace_response_runner_deps",
    "replace_turn_controller_deps",
    "replace_turn_policy_deps",
    "replace_turn_store_deps",
    "request_envelope",
    "requires_linux",
    "runtime_paths_for",
    "sync_bot_runtime_state",
    "test_runtime_paths",
    "unwrap_extracted_collaborator",
    "wrap_extracted_collaborators",
    "write_config_yaml",
]

_TEST_RUNTIME_PATHS_BY_CONFIG_ID: dict[int, RuntimePaths] = {}
_VISIBLE_MESSAGE_IDS = count(1)
_ANSI_RE = re.compile(r"\x1b\[[0-9;]*m")
_SOFT_WRAP_RE = re.compile(r"(?<=\S)\n(?=\S)")
RuntimeBot = AgentBot | TeamBot
TestFunction = Callable[..., object]


async def prepare_history_for_run_for_test(
    *,
    agent: "Agent",
    agent_name: str,
    full_prompt: str,
    session_id: str | None,
    runtime_paths: RuntimePaths,
    config: Config,
    execution_identity: "ToolExecutionIdentity | None",
    storage: "BaseDb | None" = None,
    session: "AgentSession | TeamSession | None" = None,
    history_settings: ResolvedHistorySettings | None = None,
    compaction_config: "CompactionConfig | None" = None,
    has_authored_compaction_config: bool | None = None,
    active_model_name: str | None = None,
    active_context_window: int | None = None,
    static_prompt_tokens: int | None = None,
    available_history_budget: int | None = None,
    scope: HistoryScope | None = None,
    execution_plan: ResolvedHistoryExecutionPlan | None = None,
    compaction_lifecycle: CompactionLifecycle | None = None,
) -> PreparedHistoryState:
    """Compose the production history-preparation seams for one test run."""
    resolved_scope = scope or _resolve_history_scope(agent)
    resolved_inputs = resolve_agent_preparation_inputs(
        agent=agent,
        agent_name=agent_name,
        full_prompt=full_prompt,
        config=config,
        history_settings=history_settings,
        compaction_config=compaction_config,
        has_authored_compaction_config=has_authored_compaction_config,
        active_model_name=active_model_name,
        active_context_window=active_context_window,
        static_prompt_tokens=static_prompt_tokens,
        execution_plan=execution_plan,
    )
    if available_history_budget is not None:
        # prepare_scope_history reads its trigger/hard budgets from the execution
        # plan, so express the test budget override through the plan itself.
        resolved_inputs = replace(
            resolved_inputs,
            execution_plan=replace(
                resolved_inputs.execution_plan,
                replay_budget_tokens=available_history_budget,
                hard_replay_budget_tokens=available_history_budget,
            ),
        )
    scope_history_kwargs = {
        "agent": agent,
        "agent_name": agent_name,
        "resolved_inputs": resolved_inputs,
        "runtime_paths": runtime_paths,
        "config": config,
        "scope": resolved_scope,
        "compaction_lifecycle": compaction_lifecycle,
    }
    if storage is not None and resolved_scope is not None and session_id is not None:
        persisted_session = session
        if persisted_session is None:
            persisted_session = (
                get_team_session(storage, session_id)
                if resolved_scope.kind == "team"
                else get_agent_session(storage, session_id)
            )
        scope_context = ScopeSessionContext(
            scope=resolved_scope,
            storage=storage,
            session=persisted_session,
            session_id=session_id,
        )
        prepared_scope_history = await prepare_scope_history(scope_context=scope_context, **scope_history_kwargs)
    else:
        with open_scope_session_context(
            agent=agent,
            agent_name=agent_name,
            session_id=session_id,
            runtime_paths=runtime_paths,
            config=config,
            execution_identity=execution_identity,
            scope=resolved_scope,
        ) as scope_context:
            prepared_scope_history = await prepare_scope_history(scope_context=scope_context, **scope_history_kwargs)
    return finalize_history_preparation(
        prepared_scope_history=prepared_scope_history,
        config=config,
        static_prompt_tokens=static_prompt_tokens,
        available_history_budget=available_history_budget,
    )


def dispatch_context_result(context: MessageContext) -> DispatchContextResult:
    """Wrap a stable message context in the dispatch extraction result shape."""
    return DispatchContextResult(context=context, thread_context=None)


def prepared_dispatch_result(dispatch: PreparedDispatch) -> _DispatchPreparation:
    """Wrap a prepared dispatch in the private turn-controller preparation result shape."""
    return _DispatchPreparation(
        dispatch=dispatch,
        replay_guard=_ReplayGuardContext(
            history=dispatch.context.replay_guard_history,
            degraded=is_thread_history_degraded(dispatch.context.replay_guard_history),
            thread_id=dispatch.target.resolved_thread_id,
        ),
    )


def agent_response_should_respond(
    agent_name: str,
    am_i_mentioned: bool,
    is_thread: bool,
    room: nio.MatrixRoom,
    thread_history: Sequence[ResolvedVisibleMessage],
    config: Config,
    runtime_paths: RuntimePaths,
    mentioned_agents: list[MatrixID] | None = None,
    has_non_agent_mentions: bool = False,
    *,
    sender_id: str,
    available_responders_in_room: list[MatrixID] | None = None,
    agents_in_thread: Sequence[MatrixID] | None = None,
) -> bool:
    """Return the boolean projection of the agent response decision for tests."""
    return decide_agent_response(
        agent_name,
        am_i_mentioned,
        is_thread,
        room,
        thread_history,
        config,
        runtime_paths,
        mentioned_agents,
        has_non_agent_mentions,
        sender_id=sender_id,
        available_responders_in_room=available_responders_in_room,
        agents_in_thread=agents_in_thread,
    ).should_respond


def message_origin(
    *,
    sender_id: str = "@user:localhost",
    requester_id: str | None = None,
    sender_entity_name: str | None = None,
    requester_entity_name: str | None = None,
    source_kind: str = "message",
    original_sender: str | None = None,
    trusted_user_relay: bool = False,
) -> TurnOrigin:
    """Build canonical origin metadata for manually constructed test envelopes."""
    return classify_turn_origin(
        transport_sender_id=sender_id,
        requester_id=requester_id or sender_id,
        sender_entity_name=sender_entity_name,
        requester_entity_name=requester_entity_name,
        source_kind=source_kind,
        original_sender=original_sender,
        trusted_user_relay=trusted_user_relay,
    )


def request_envelope(
    *,
    room_id: str = "!test:localhost",
    reply_to_event_id: str = "$event",
    thread_id: str | None = None,
    prompt: str = "Hello",
    user_id: str | None = "@user:localhost",
    target: MessageTarget | None = None,
    agent_name: str = "test_agent",
    source_kind: str = "message",
    attachment_ids: tuple[str, ...] = (),
) -> MessageEnvelope:
    """Build a canonical response envelope for direct ResponseRequest tests."""
    resolved_user_id = user_id or "@user:localhost"
    resolved_target = target or MessageTarget.resolve(room_id, thread_id, reply_to_event_id)
    return MessageEnvelope(
        source_event_id=reply_to_event_id,
        target=resolved_target,
        body=prompt,
        attachment_ids=attachment_ids,
        mentioned_agents=(),
        agent_name=agent_name,
        origin=message_origin(sender_id=resolved_user_id, requester_id=resolved_user_id, source_kind=source_kind),
    )


def requires_linux(
    *,
    reason: str = "requires Linux",
    timeout: float | None = None,
) -> Callable[[TestFunction], TestFunction]:
    """Return a decorator for tests that only run on Linux."""

    def decorator(test_func: TestFunction) -> TestFunction:
        marked = pytest.mark.skipif(sys.platform != "linux", reason=reason)(test_func)
        if timeout is not None:
            marked = pytest.mark.timeout(timeout)(marked)
        return marked

    return decorator


async def drain_coalescing(*bots: RuntimeBot) -> None:
    """Drain gate batches and detached responses until both are quiescent.

    A detached response settling during the runner drain can release its
    lifecycle lock and flush a busy-conversation backlog into the gate, so a
    single gate-then-runner pass is not a reliable barrier.
    """
    for bot in bots:
        runner = unwrap_extracted_collaborator(bot._response_runner)
        while True:
            # Concurrently: the gate drain may hold a busy conversation's
            # backlog until its detached response goes idle, which only the
            # runner drain settles.
            await asyncio.gather(
                bot._coalescing_gate.drain_all(),
                runner.drain_inbox_responses(),
            )
            if not runner._inbox_response_tasks and not bot._coalescing_gate._gates:
                break


def _wait_for_postgres_container(database_url: str) -> None:
    import psycopg  # noqa: PLC0415

    deadline = time.monotonic() + _POSTGRES_STARTUP_TIMEOUT_SECONDS
    last_error: Exception | None = None
    while time.monotonic() < deadline:
        try:
            with psycopg.connect(database_url, connect_timeout=1):
                return
        except psycopg.OperationalError as exc:
            last_error = exc
            time.sleep(0.25)
    msg = "Postgres test container did not become ready"
    raise RuntimeError(msg) from last_error


def _create_postgres_worker_database(database_url: str, worker_id: str) -> str:
    """Create an isolated database for one worker on the shared Postgres server."""
    import psycopg  # noqa: PLC0415
    from psycopg import sql  # noqa: PLC0415

    database_name = f"mindroom_{worker_id}"
    with psycopg.connect(database_url, autocommit=True) as db:
        db.execute(sql.SQL("CREATE DATABASE {}").format(sql.Identifier(database_name)))
    return f"{database_url.rsplit('/', 1)[0]}/{database_name}"


def _wait_for_postgres_container_port(docker: str, container_name: str) -> str:
    """Wait for Docker to publish a shared container's random host port."""
    deadline = time.monotonic() + _POSTGRES_STARTUP_TIMEOUT_SECONDS
    last_error = ""
    while time.monotonic() < deadline:
        result = subprocess.run(
            [docker, "port", container_name, "5432/tcp"],
            check=False,
            capture_output=True,
            text=True,
        )
        if result.returncode == 0 and result.stdout.strip():
            mapped_port = result.stdout.strip().splitlines()[-1]
            _, port = mapped_port.rsplit(":", 1)
            return f"postgresql://cache:test@127.0.0.1:{port}/mindroom"
        last_error = result.stderr.strip()
        time.sleep(0.05)
    msg = f"Postgres test container did not publish a port: {last_error}"
    raise RuntimeError(msg)


def _postgres_container_name(run_id: str) -> str:
    """Return the deterministic disposable Postgres container name for one test run."""
    return f"{_POSTGRES_CONTAINER_PREFIX}{run_id}"


def _remove_postgres_container(docker: str, container_name: str) -> None:
    """Remove one disposable Postgres container if it exists."""
    result = subprocess.run(
        [docker, "rm", "-f", container_name],
        check=False,
        capture_output=True,
        text=True,
    )
    if result.returncode == 0 or "No such container" in result.stderr:
        return
    msg = f"Could not remove Postgres test container {container_name}: {result.stderr.strip()}"
    raise RuntimeError(msg)


def pytest_configure_node(node: "WorkerController") -> None:
    """Remember the shared Postgres container name in the xdist controller."""
    node.config.stash[_POSTGRES_CONTAINER_NAME_STASH_KEY] = _postgres_container_name(
        node.workerinput["testrunuid"],
    )


def pytest_sessionfinish(session: pytest.Session) -> None:
    """Remove the shared Postgres container after every xdist worker has finished."""
    if hasattr(session.config, "workerinput"):
        return
    container_name = session.config.stash.get(_POSTGRES_CONTAINER_NAME_STASH_KEY, None)
    docker = shutil.which("docker")
    if container_name is not None and docker is not None:
        try:
            _remove_postgres_container(docker, container_name)
        except RuntimeError as exc:
            warnings.warn(pytest.PytestWarning(str(exc)), stacklevel=1)
            session.exitstatus = pytest.ExitCode.TESTS_FAILED


@pytest.fixture(scope="session")
def postgres_event_cache_url(
    worker_id: str,
    testrun_uid: str,
) -> Iterator[str]:
    """Start or reuse one disposable Postgres server for the current test run."""
    docker = shutil.which("docker")
    if docker is None:
        pytest.skip("Docker is required for Postgres event-cache integration tests")

    info_result = subprocess.run(
        [docker, "info"],
        check=False,
        capture_output=True,
        text=True,
    )
    if info_result.returncode != 0:
        pytest.skip("Docker daemon is unavailable for Postgres event-cache integration tests")

    shared_across_workers = worker_id != "master"
    run_id = testrun_uid if shared_across_workers else uuid.uuid4().hex
    container_name = _postgres_container_name(run_id)
    run_result = subprocess.run(
        [
            docker,
            "run",
            "--rm",
            "-d",
            "--name",
            container_name,
            "--label",
            f"mindroom.pytest.run={run_id}",
            "-e",
            "POSTGRES_USER=cache",
            "-e",
            "POSTGRES_PASSWORD=test",
            "-e",
            "POSTGRES_DB=mindroom",
            "-p",
            "127.0.0.1::5432",
            "postgres:15-alpine",
        ],
        check=False,
        capture_output=True,
        text=True,
    )
    if run_result.returncode != 0:
        inspect_result = subprocess.run(
            [docker, "inspect", "--format", "{{.State.Status}}", container_name],
            check=False,
            capture_output=True,
            text=True,
        )
        if not shared_across_workers or inspect_result.returncode != 0:
            pytest.skip(f"Could not start Postgres test container: {run_result.stderr.strip()}")
        if inspect_result.stdout.strip() in {"dead", "exited"}:
            msg = f"Shared Postgres test container is {inspect_result.stdout.strip()}"
            raise RuntimeError(msg)

    try:
        database_url = _wait_for_postgres_container_port(docker, container_name)
        _wait_for_postgres_container(database_url)
        if shared_across_workers:
            database_url = _create_postgres_worker_database(database_url, worker_id)
        yield database_url
    finally:
        if not shared_across_workers:
            _remove_postgres_container(docker, container_name)


@pytest.fixture(params=("sqlite", "postgres"), ids=("sqlite", "postgres"))
def event_cache_factory(
    request: pytest.FixtureRequest,
    tmp_path: Path,
) -> Callable[[], "ConversationEventCache"]:
    """Return a cache factory for backend-neutral event-cache contract tests."""
    backend = str(request.param)
    if backend == "sqlite":
        db_path = tmp_path / "event_cache.db"
        return lambda: SqliteEventCache(db_path)
    if backend == "postgres":
        from mindroom.matrix.cache.postgres_event_cache import PostgresEventCache  # noqa: PLC0415

        database_url = request.getfixturevalue("postgres_event_cache_url")
        namespace = f"test_{uuid.uuid4().hex}"
        return lambda: PostgresEventCache(database_url=database_url, namespace=namespace)
    msg = f"Unsupported event cache backend fixture: {backend}"
    raise AssertionError(msg)


@pytest_asyncio.fixture
async def event_cache(
    event_cache_factory: Callable[[], "ConversationEventCache"],
) -> AsyncGenerator["ConversationEventCache", None]:
    """Return one initialized event cache backend for backend-neutral contract tests."""
    cache = event_cache_factory()
    await cache.initialize()
    try:
        yield cache
    finally:
        await cache.close()


async def _empty_async_iterator() -> AsyncGenerator[object, None]:
    """Yield nothing while preserving async-iterator semantics for nio relations APIs."""
    if False:
        yield None


def _make_room_get_event_response(event_id: str) -> nio.RoomGetEventResponse:
    """Return a minimal RoomGetEventResponse containing one visible text event."""
    event = MagicMock(spec=nio.RoomMessageText)
    event.event_id = event_id
    event.sender = "@user:localhost"
    event.body = event_id
    event.server_timestamp = 0
    event.source = {
        "type": "m.room.message",
        "content": {
            "msgtype": "m.text",
            "body": event_id,
        },
    }
    response = nio.RoomGetEventResponse()
    response.event = event
    return response


def _outcome(
    terminal_status: str,
    event_id: str | None = None,
    is_visible_response: bool = False,
    final_visible_body: str | None = None,
    delivery_kind: str | None = None,
    failure_reason: str | None = None,
    suppressed: bool = False,
    tool_trace: tuple[object, ...] = (),
    extra_content: Mapping[str, object] | None = None,
    option_map: dict[str, str] | None = None,
    options_list: tuple[dict[str, str], ...] | None = None,
) -> FinalDeliveryOutcome:
    """Build one compact terminal outcome for tests."""
    resolved_suppressed = suppressed or (failure_reason == "suppressed_by_hook" and not is_visible_response)
    return FinalDeliveryOutcome(
        terminal_status=terminal_status,
        event_id=event_id,
        is_visible_response=is_visible_response,
        final_visible_body=final_visible_body,
        delivery_kind=delivery_kind,
        failure_reason=failure_reason,
        suppressed=resolved_suppressed,
        tool_trace=tool_trace,
        extra_content=dict(extra_content or {}),
        interactive_metadata=InteractiveMetadata._from_parts(option_map, options_list),
    )


class _AutoRoomCache(MutableMapping[str, nio.MatrixRoom]):
    """Mutable test room cache that lazily vends joined unencrypted rooms."""

    def __init__(self, own_user_id: str) -> None:
        self._own_user_id = own_user_id
        self._rooms: dict[str, nio.MatrixRoom] = {}

    def __getitem__(self, room_id: str) -> nio.MatrixRoom:
        room = self._rooms.get(room_id)
        if room is not None:
            return room
        if not room_id.startswith("!"):
            raise KeyError(room_id)
        room = nio.MatrixRoom(room_id, self._own_user_id)
        self._rooms[room_id] = room
        return room

    def __setitem__(self, room_id: str, room: nio.MatrixRoom) -> None:
        self._rooms[room_id] = room

    def __delitem__(self, room_id: str) -> None:
        del self._rooms[room_id]

    def __iter__(self) -> Iterator[str]:
        yield from self._rooms

    def __len__(self) -> int:
        return len(self._rooms)


def make_matrix_client_mock(*, user_id: str = "@mindroom_test:example.com") -> AsyncMock:
    """Return an AsyncClient-shaped mock with safe defaults for sync nio APIs."""
    client = AsyncMock(spec=nio.AsyncClient)
    client.user_id = user_id
    client.rooms = _AutoRoomCache(user_id)
    client.next_batch = "s_test_token"
    client.loaded_sync_token = ""
    client.has_uncommitted_classic_sync_state = False
    presence_response = MagicMock()
    presence_response.presence = "offline"
    presence_response.last_active_ago = 3_600_000
    room_messages_response = nio.RoomMessagesResponse(room_id="!test:localhost", chunk=[], start="", end=None)
    client.add_event_callback = MagicMock()
    client.add_response_callback = MagicMock()
    client.get_presence = AsyncMock(return_value=presence_response)
    client.room_get_event = AsyncMock(side_effect=lambda _room_id, event_id: _make_room_get_event_response(event_id))
    client.room_get_event_relations = MagicMock(return_value=_empty_async_iterator())
    client.room_messages = AsyncMock(return_value=room_messages_response)
    client.joined_rooms = AsyncMock(return_value=nio.JoinedRoomsResponse(rooms=[]))

    async def reset_classic_sync_state() -> None:
        client.next_batch = ""
        client.loaded_sync_token = ""
        client.rooms.clear()
        client.has_uncommitted_classic_sync_state = False

    def acknowledge_classic_sync(_next_batch: str) -> None:
        client.has_uncommitted_classic_sync_state = False

    client.clear_persisted_sync_recovery = MagicMock()
    client.acknowledge_classic_sync = MagicMock(side_effect=acknowledge_classic_sync)
    client.reset_classic_sync_state.side_effect = reset_classic_sync_state
    return client


def delivered_matrix_event(
    event_id: str,
    content: Mapping[str, object] | None = None,
) -> DeliveredMatrixEvent:
    """Return one delivered Matrix event using the exact content payload seen by the helper."""
    return DeliveredMatrixEvent(
        event_id=event_id,
        content_sent={} if content is None else dict(content),
    )


def delivered_matrix_side_effect(event_id: str) -> Callable[..., Awaitable[DeliveredMatrixEvent]]:
    """Build one async mock side effect that mirrors send/edit helpers returning delivered events."""

    async def _deliver(*args: object, **kwargs: object) -> DeliveredMatrixEvent:
        if "content" in kwargs:
            content = kwargs["content"]
            content_mapping = content if isinstance(content, Mapping) else None
            return delivered_matrix_event(event_id, content_mapping)
        if "new_content" in kwargs:
            new_content = kwargs["new_content"]
            new_text = kwargs.get("new_text")
            if isinstance(new_content, Mapping) and isinstance(new_text, str):
                return delivered_matrix_event(
                    event_id,
                    build_edit_event_content(
                        event_id=str(args[2]) if len(args) > 2 else "",
                        new_content=dict(new_content),
                        new_text=new_text,
                    ),
                )
            content = new_content
            content_mapping = content if isinstance(content, Mapping) else None
            return delivered_matrix_event(event_id, content_mapping)
        if len(args) > 4 and isinstance(args[3], Mapping) and isinstance(args[4], str):
            return delivered_matrix_event(
                event_id,
                build_edit_event_content(
                    event_id=str(args[2]),
                    new_content=dict(args[3]),
                    new_text=args[4],
                ),
            )
        content_index = 2 if len(args) <= 3 else 3
        content = args[content_index] if len(args) > content_index else None
        content_mapping = content if isinstance(content, Mapping) else None
        return delivered_matrix_event(event_id, content_mapping)

    return _deliver


def make_event_cache_mock() -> AsyncMock:
    """Return an async mock shaped like the event cache protocol."""
    event_cache = AsyncMock(spec=SqliteEventCache)
    event_cache.principal_id = "@mindroom_test:localhost"
    event_cache.cache_generation = "test-cache-generation"
    event_cache.durable_writes_available = True
    event_cache.get_event.return_value = None
    event_cache.get_latest_edit.return_value = None
    event_cache.get_mxc_text.return_value = None
    event_cache.get_mxc_texts.return_value = {}
    event_cache.get_recent_room_events.return_value = []
    event_cache.get_recent_room_thread_ids.return_value = []
    event_cache.get_thread_events.return_value = None

    async def has_thread_snapshot(room_id: str, thread_id: str) -> bool:
        return await event_cache.get_thread_events(room_id, thread_id) is not None

    event_cache.has_thread_snapshot.side_effect = has_thread_snapshot
    event_cache.get_thread_cache_gap.return_value = None
    event_cache.get_thread_id_for_event.return_value = None
    event_cache.get_latest_agent_message_snapshot.return_value = None
    event_cache.pending_durable_write_room_ids.return_value = ()
    event_cache.runtime_diagnostics.return_value = {"cache_backend": "mock"}
    departure_epochs: dict[str, int] = {}

    def mark_room_departed(room_id: str) -> int:
        epoch = departure_epochs.get(room_id, 0) + 1
        departure_epochs[room_id] = epoch
        return epoch

    event_cache.mark_room_departed.side_effect = mark_room_departed
    event_cache.room_departure_epoch.side_effect = lambda room_id: departure_epochs.get(room_id, 0)
    event_cache.room_membership_epoch.return_value = 0
    event_cache.flush_pending_durable_writes.return_value = None
    event_cache.apply_thread_mutation_append.return_value = ThreadAppendOutcome.APPENDED
    event_cache.redact_event.return_value = False
    event_cache.store_mxc_text.return_value = True
    event_cache.replace_thread.return_value = True
    return event_cache


def make_conversation_cache_mock() -> AsyncMock:
    """Return an async mock shaped like the conversation cache protocol."""
    conversation_cache = AsyncMock(spec=ConversationCacheProtocol)
    conversation_cache.get_event = AsyncMock(
        side_effect=lambda _room_id, event_id: _make_room_get_event_response(event_id),
    )
    conversation_cache.get_thread_history = AsyncMock(
        return_value=thread_history_result([], is_full_history=True),
    )
    conversation_cache.get_dispatch_thread_snapshot = AsyncMock(
        return_value=thread_history_result([], is_full_history=False),
    )
    conversation_cache.get_dispatch_thread_history = AsyncMock(
        return_value=thread_history_result([], is_full_history=True),
    )
    conversation_cache.get_strict_thread_history = AsyncMock(
        return_value=thread_history_result([], is_full_history=True),
    )

    conversation_cache.get_thread_id_for_event = AsyncMock(return_value=None)
    conversation_cache.get_latest_thread_event_id_if_needed = AsyncMock(return_value=None)
    conversation_cache.append_live_event = AsyncMock()
    conversation_cache.notify_outbound_message = MagicMock()
    conversation_cache.notify_outbound_event = MagicMock()
    conversation_cache.notify_outbound_redaction = MagicMock()
    return conversation_cache


def make_event_cache_write_coordinator_mock(*, owner: object | None = None) -> EventCacheWriteCoordinator:
    """Return a coordinator-shaped runtime helper with the real synchronous queue contract."""
    return EventCacheWriteCoordinator(
        logger=MagicMock(),
        background_task_owner=object() if owner is None else owner,
    )


def install_runtime_cache_support(bot: RuntimeBot) -> RuntimeBot:
    """Attach required cache runtime support to one test bot."""
    if bot._runtime_view.event_cache is None:
        bot.event_cache = make_event_cache_mock()
    if bot._runtime_view.event_cache_write_coordinator is None:
        bot.event_cache_write_coordinator = make_event_cache_write_coordinator_mock(owner=bot._runtime_view)
    if bot._runtime_view.startup_thread_prewarm_registry is None:
        bot.startup_thread_prewarm_registry = StartupThreadPrewarmRegistry()
    sync_bot_runtime_state(bot)
    return bot


def install_call_manager_mock(bot: RuntimeBot, call_manager: object | None) -> None:
    """Install a call-manager fake through the shared test seam."""
    bot._call_manager = cast("CallManager | None", call_manager)


def normalize_console_output(text: str) -> str:
    """Collapse wrapped console output for stable substring assertions."""
    return " ".join(_SOFT_WRAP_RE.sub("", _ANSI_RE.sub("", text)).split())


class _ExtractedCollaboratorProxy[CollaboratorT]:
    """Mutable proxy that keeps real collaborator attributes visible to tests."""

    def __init__(self, wrapped: CollaboratorT) -> None:
        object.__setattr__(self, "_wrapped", wrapped)

    def __getattr__(self, name: str) -> object:
        proxy_dict = object.__getattribute__(self, "__dict__")
        if name in proxy_dict:
            return proxy_dict[name]
        return getattr(object.__getattribute__(self, "_wrapped"), name)

    def __setattr__(self, name: str, value: object) -> None:
        object.__getattribute__(self, "__dict__")[name] = value

    def __delattr__(self, name: str) -> None:
        proxy_dict = object.__getattribute__(self, "__dict__")
        if name in proxy_dict:
            del proxy_dict[name]
            return
        msg = f"{type(self).__name__!s} has no attribute {name!r}"
        raise AttributeError(msg)


class FakeModel(Model):
    """Minimal model returning one canned response, for deterministic agent tests."""

    def invoke(self, *_args: object, **_kwargs: object) -> ModelResponse:
        """Return one successful fake response."""
        return ModelResponse(content="ok")

    async def ainvoke(self, *_args: object, **_kwargs: object) -> ModelResponse:
        """Return one successful fake async response."""
        return ModelResponse(content="ok")

    def invoke_stream(self, *_args: object, **_kwargs: object) -> Iterator[ModelResponse]:
        """Yield one successful fake streaming response."""
        yield ModelResponse(content="ok")

    async def ainvoke_stream(self, *_args: object, **_kwargs: object) -> AsyncIterator[ModelResponse]:
        """Yield one successful fake async streaming response."""
        yield ModelResponse(content="ok")

    def _parse_provider_response(self, response: ModelResponse, *_args: object, **_kwargs: object) -> ModelResponse:
        return response

    def _parse_provider_response_delta(
        self,
        response: ModelResponse,
        *_args: object,
        **_kwargs: object,
    ) -> ModelResponse:
        return response


class FakeCredentialsManager:
    """Stub credentials manager for tests that need credential lookup."""

    def __init__(
        self,
        credentials_by_service: dict[str, dict[str, object]],
        worker_managers: dict[str, "FakeCredentialsManager"] | None = None,
        *,
        storage_root: Path | None = None,
        current_worker_key: str | None = None,
        current_worker_root: Path | None = None,
    ) -> None:
        self._credentials_by_service = credentials_by_service
        self._worker_managers = worker_managers or {}
        self.storage_root = storage_root or Path("/var/empty/mindroom-fake-storage")
        self.base_path = self.storage_root / "credentials"
        self.shared_base_path = self.base_path
        self.current_worker_key = current_worker_key
        self.current_worker_root = current_worker_root

    def load_credentials(self, service: str) -> dict[str, object]:
        """Return stored credentials for *service*, or empty dict."""
        return self._credentials_by_service.get(service, {})

    def for_worker(self, worker_key: str) -> "FakeCredentialsManager":
        """Return a worker-scoped credentials manager."""
        return self._worker_managers.get(
            worker_key,
            FakeCredentialsManager(
                {},
                storage_root=self.storage_root / "workers" / worker_key,
                current_worker_key=worker_key,
                current_worker_root=self.storage_root / "workers" / worker_key,
            ),
        )

    def shared_manager(self) -> "FakeCredentialsManager":
        """Return the shared credential layer for this fake manager."""
        return self


def pytest_collection_modifyitems(items: list[pytest.Item]) -> None:
    """Skip tests marked with requires_matrix unless MATRIX_SERVER_URL is set."""
    if os.environ.get("MATRIX_SERVER_URL"):
        # Matrix server available, don't skip
        return

    skip_marker = pytest.mark.skip(reason="requires_matrix: no MATRIX_SERVER_URL set")
    for item in items:
        if "requires_matrix" in item.keywords:
            item.add_marker(skip_marker)


# Test credentials constants - not real credentials, safe for testing
TEST_PASSWORD = "mock_test_password"  # noqa: S105
TEST_ACCESS_TOKEN = "mock_test_token"  # noqa: S105


def test_runtime_paths(tmp_root: Path) -> RuntimePaths:
    """Create an isolated runtime context for one test config."""
    tmp_root.mkdir(parents=True, exist_ok=True)
    config_path = tmp_root / "config.yaml"
    config_path.write_text("router:\n  model: default\n", encoding="utf-8")
    return resolve_runtime_paths(
        config_path=config_path,
        storage_path=tmp_root / "mindroom_data",
        process_env={
            "MATRIX_HOMESERVER": "http://localhost:8008",
            "MINDROOM_NAMESPACE": "",
        },
    )


test_runtime_paths.__test__ = False


def orchestrator_runtime_paths(
    storage_path: Path,
    *,
    config_path: Path | None = None,
) -> RuntimePaths:
    """Build an explicit runtime context for orchestrator tests.

    Default the config path to an isolated file under the provided test root so
    callers never fall back to the tracked repo-root config.yaml.
    """
    if config_path is None:
        config_path = storage_path / "config.yaml"
    return resolve_runtime_paths(
        config_path=config_path,
        storage_path=storage_path,
        process_env={
            "MATRIX_HOMESERVER": "http://localhost:8008",
            "MINDROOM_NAMESPACE": "",
        },
    )


def load_config_yaml(config_path: Path) -> Config:
    """Load a config YAML file through the production runtime-aware loader."""
    return load_config(resolve_runtime_paths(config_path=Path(config_path).expanduser().resolve()))


def write_config_yaml(config: Config, config_path: Path) -> None:
    """Write a test config using the authored YAML representation."""
    path = Path(config_path)
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp_path = path.with_suffix(path.suffix + ".tmp")
    with tmp_path.open("w", encoding="utf-8") as f:
        yaml.dump(
            config.authored_model_dump(),
            f,
            default_flow_style=False,
            sort_keys=True,
            allow_unicode=True,
            width=120,
        )
    safe_replace(tmp_path, path)


def bind_runtime_paths(
    config: Config,
    runtime_paths: RuntimePaths,
) -> Config:
    """Return a runtime-bound copy of a test config."""
    bound = Config.validate_with_runtime(config.authored_model_dump(), runtime_paths)
    _persist_bound_entity_accounts(bound, runtime_paths)
    authored_coalescing = config.defaults.coalescing
    if "debounce_ms" not in authored_coalescing.model_fields_set:
        bound.defaults.coalescing.debounce_ms = 0
    _TEST_RUNTIME_PATHS_BY_CONFIG_ID[id(bound)] = runtime_paths
    return bound


def _persist_bound_entity_accounts(config: Config, runtime_paths: RuntimePaths) -> None:
    """Prepare managed Matrix accounts for tests that bind runtime config."""
    persist_entity_accounts(config, runtime_paths)


def bind_mock_config_cache(mock_config: MagicMock, runtime_root: Path) -> Path:
    """Give a config mock the cache path contract used by orchestrator init."""
    cache_path = runtime_root / "event_cache.db"
    mock_config.cache.backend = "sqlite"
    mock_config.cache.resolve_db_path.return_value = cache_path
    return cache_path


def runtime_paths_for(config: Config) -> RuntimePaths:
    """Return the explicit runtime context previously bound to a test config."""
    runtime_paths = _TEST_RUNTIME_PATHS_BY_CONFIG_ID.get(id(config))
    if runtime_paths is None:
        msg = "Test config is missing bound RuntimePaths"
        raise KeyError(msg)
    return runtime_paths


def create_mock_room(
    room_id: str = "!test:localhost",
    agents: list[str] | None = None,
    config: Config | None = None,
) -> MagicMock:
    """Create a mock room with specified agents."""
    room = MagicMock()
    room.room_id = room_id
    if agents:
        domain = config.get_domain(runtime_paths_for(config)) if config is not None else "localhost"
        room.users = {f"@mindroom_{agent}:{domain}": None for agent in agents}
    else:
        room.users = {}
    return room


def make_turn_context(
    entity_label: str = "test_agent",
    *,
    session_id: str | None = "test_session",
    run_id: str | None = None,
    correlation_id: str = "corr-test",
    reply_to_event_id: str | None = None,
    room_id: str | None = None,
    thread_id: str | None = None,
    requester_id: str | None = None,
    matrix_run_metadata: dict[str, Any] | None = None,
    active_event_ids: frozenset[str] = frozenset(),
    transient_enrichment_items: tuple[EnrichmentItem, ...] = (),
    system_enrichment_items: tuple[EnrichmentItem, ...] = (),
    scheduled_history_budget: ScheduledHistoryBudget | None = None,
) -> ResponseTurnContext:
    """Build one response-turn context with test defaults."""
    return ResponseTurnContext(
        entity_label=entity_label,
        session_id=session_id,
        run_id=run_id,
        correlation_id=correlation_id,
        reply_to_event_id=reply_to_event_id,
        room_id=room_id,
        thread_id=thread_id,
        requester_id=requester_id,
        matrix_run_metadata=matrix_run_metadata,
        active_event_ids=active_event_ids,
        transient_enrichment_items=transient_enrichment_items,
        system_enrichment_items=system_enrichment_items,
        scheduled_history_budget=scheduled_history_budget,
    )


def make_visible_message(
    *,
    sender: str = "@user:localhost",
    body: str = "",
    event_id: str | None = None,
    timestamp: int = 0,
    content: dict[str, object] | None = None,
    thread_id: str | None = None,
) -> ResolvedVisibleMessage:
    """Build one typed visible message for thread/history tests."""
    resolved_content = dict(content) if isinstance(content, dict) else {}
    if "body" not in resolved_content and body:
        resolved_content["body"] = body
    return ResolvedVisibleMessage.synthetic(
        sender=sender,
        body=body,
        event_id=event_id or f"$visible-{next(_VISIBLE_MESSAGE_IDS)}",
        timestamp=timestamp,
        content=resolved_content or None,
        thread_id=thread_id,
    )


def unwrap_extracted_collaborator[T](collaborator: T) -> T:
    """Return the real extracted collaborator behind one test wrapper."""
    if isinstance(collaborator, MagicMock):
        wrapped = collaborator._mock_wraps
        if wrapped is not None:
            return wrapped
    wrapped = getattr(collaborator, "_wrapped", None)
    if wrapped is not None:
        return wrapped
    return collaborator


def wrap_extracted_collaborators(bot: RuntimeBot, *names: str) -> RuntimeBot:
    """Wrap frozen extracted collaborators so tests can patch their methods."""
    sync_bot_runtime_state(bot)
    collaborator_names = names or (
        "_turn_policy",
        "_delivery_gateway",
        "_response_runner",
        "_turn_store",
        "_visible_voice_echo",
        "_edit_regenerator",
        "_inbound_turn_normalizer",
        "_conversation_resolver",
        "_conversation_state_writer",
    )
    for name in collaborator_names:
        collaborator = getattr(bot, name)
        if isinstance(collaborator, MagicMock | _ExtractedCollaboratorProxy):
            continue
        setattr(bot, name, _ExtractedCollaboratorProxy(collaborator))
    _sync_request_payload_preparer(bot)
    return bot


def _sync_request_payload_preparer(bot: RuntimeBot) -> None:
    """Repoint the response runner's payload preparer at the current collaborators.

    The preparer captures the normalizer and ingress hook runner; tests swap
    those for proxies after construction, so rebuild the preparer to track them.
    """
    runner = unwrap_extracted_collaborator(bot._response_runner)
    preparer = ResponsePayloadPreparer(
        normalizer=bot._inbound_turn_normalizer,
        ingress_hook_runner=bot._ingress_hook_runner,
        agent_name=runner.deps.agent_name,
        logger=runner.deps.logger,
    )
    bot._request_payload_preparer = preparer
    runner.deps = replace(runner.deps, request_preparer=preparer)


async def prepare_payload_via_seam(bot: RuntimeBot, execute_args: tuple[object, ...]) -> None:
    """Drive the execution-side payload preparation from captured dispatch args."""
    event = cast("DispatchEvent", execute_args[1])
    dispatch = cast("PreparedDispatch", execute_args[2])
    payload_inputs = cast("DispatchPayloadInputs", execute_args[4])
    await bot._request_payload_preparer.prepare(
        ResponseRequest(
            thread_history=dispatch.context.thread_history,
            prompt=event.body,
            response_envelope=dispatch.envelope,
            payload_preparation=ResponsePayloadPreparation(
                dispatch=dispatch,
                prompt=event.body,
                action_kind="individual",
                payload_inputs=payload_inputs,
                target_member_names=None,
                dispatch_started_at=0.0,
                context_ready_monotonic=0.0,
            ),
        ),
    )


def sync_bot_runtime_state(bot: RuntimeBot) -> None:
    """Update the extracted runtime state after tests mutate bot internals."""
    runtime = bot._runtime_view
    client = bot.client
    if client is not None and getattr(client, "user_id", None) is None:
        client.user_id = bot.matrix_id.full_id
    runtime.client = bot.client
    runtime.config = bot.config
    runtime.enable_streaming = bot.enable_streaming
    runtime.orchestrator = bot.orchestrator


def replace_turn_policy_deps(bot: RuntimeBot, **changes: object) -> TurnPolicy:
    """Rebuild the turn policy after swapping collaborators captured at construction."""
    sync_bot_runtime_state(bot)
    policy = unwrap_extracted_collaborator(bot._turn_policy)
    policy_field_names = set(policy.deps.__dataclass_fields__)
    policy_changes = {name: value for name, value in changes.items() if name in policy_field_names}
    rebuilt = TurnPolicy(replace(policy.deps, **policy_changes)) if policy_changes else policy
    bot._turn_policy = rebuilt
    wrap_extracted_collaborators(bot, "_turn_policy")
    store_field_names = set(unwrap_extracted_collaborator(bot._turn_store).deps.__dataclass_fields__)
    store_changes = {name: value for name, value in changes.items() if name in store_field_names}
    if store_changes:
        replace_turn_store_deps(bot, **store_changes)
    controller = unwrap_extracted_collaborator(bot._turn_controller)
    controller_field_names = set(controller.deps.__dataclass_fields__)
    controller_changes = {name: value for name, value in changes.items() if name in controller_field_names}
    if policy_changes:
        controller_changes["turn_policy"] = bot._turn_policy
    if store_changes:
        controller_changes["turn_store"] = bot._turn_store
    if controller_changes:
        replace_turn_controller_deps(bot, **controller_changes)
    return rebuilt


def replace_turn_store_deps(bot: RuntimeBot, **changes: object) -> TurnStore:
    """Rebuild the turn store after swapping collaborators captured at construction."""
    sync_bot_runtime_state(bot)
    store = unwrap_extracted_collaborator(bot._turn_store)
    rebuilt = TurnStore(replace(store.deps, **changes))
    bot._turn_store = rebuilt
    wrap_extracted_collaborators(bot, "_turn_store")
    return rebuilt


def replace_delivery_gateway_deps(bot: RuntimeBot, **changes: object) -> DeliveryGateway:
    """Rebuild the delivery gateway after swapping captured collaborators."""
    sync_bot_runtime_state(bot)
    gateway = unwrap_extracted_collaborator(bot._delivery_gateway)
    rebuilt = DeliveryGateway(replace(gateway.deps, **changes))
    bot._delivery_gateway = rebuilt
    wrap_extracted_collaborators(bot, "_delivery_gateway")
    replace_turn_controller_deps(bot, delivery_gateway=bot._delivery_gateway)
    replace_response_runner_deps(bot, delivery_gateway=bot._delivery_gateway)
    return rebuilt


def replace_response_runner_deps(bot: RuntimeBot, **changes: object) -> ResponseRunner:
    """Rebuild the response runner after swapping captured collaborators."""
    sync_bot_runtime_state(bot)
    coordinator = unwrap_extracted_collaborator(bot._response_runner)
    rebuilt = ResponseRunner(replace(coordinator.deps, **changes))
    bot._response_runner = rebuilt
    wrap_extracted_collaborators(bot, "_response_runner")
    replace_turn_controller_deps(bot, response_runner=bot._response_runner)
    return rebuilt


def replace_edit_regenerator_deps(bot: RuntimeBot, **changes: object) -> EditRegenerator:
    """Rebuild the edit regenerator after swapping captured collaborators."""
    install_runtime_cache_support(bot)
    regenerator = unwrap_extracted_collaborator(bot._edit_regenerator)
    regenerator_field_names = set(regenerator.deps.__dataclass_fields__)
    rebuilt_changes = {
        name: value for name, value in changes.items() if name in regenerator_field_names or name == "logger"
    }
    if "logger" in rebuilt_changes:
        logger = rebuilt_changes.pop("logger")
        rebuilt_changes["get_logger"] = lambda logger=logger: logger
    if "receipt_order" not in rebuilt_changes:
        receipt_orders = count(1)

        async def next_receipt_order() -> int:
            return next(receipt_orders)

        rebuilt_changes["receipt_order"] = next_receipt_order
    store_field_names = set(unwrap_extracted_collaborator(bot._turn_store).deps.__dataclass_fields__)
    store_changes = {name: value for name, value in changes.items() if name in store_field_names}
    if store_changes:
        replace_turn_store_deps(bot, **store_changes)
        rebuilt_changes["turn_store"] = bot._turn_store
    rebuilt = EditRegenerator(replace(regenerator.deps, **rebuilt_changes))
    bot._edit_regenerator = rebuilt
    wrap_extracted_collaborators(bot, "_edit_regenerator")
    replace_turn_controller_deps(bot, edit_regenerator=bot._edit_regenerator)
    return rebuilt


def replace_reaction_dispatcher_deps(bot: RuntimeBot, **changes: object) -> ReactionDispatcher:
    """Rebuild reaction dispatch after swapping collaborators captured at construction."""
    rebuilt = ReactionDispatcher(replace(bot._reaction_dispatcher.deps, **changes))
    bot._reaction_dispatcher = rebuilt
    return rebuilt


def replace_turn_controller_deps(bot: RuntimeBot, **changes: object) -> TurnController:
    """Rebuild the turn controller after swapping collaborators captured at construction."""
    sync_bot_runtime_state(bot)
    controller = unwrap_extracted_collaborator(bot._turn_controller)
    controller_field_names = set(controller.deps.__dataclass_fields__)
    rebuilt_changes = {name: value for name, value in changes.items() if name in controller_field_names}
    default_collaborators = {
        "conversation_cache": "_conversation_cache",
        "resolver": "_conversation_resolver",
        "normalizer": "_inbound_turn_normalizer",
        "command_executor": "_command_turn_executor",
        "turn_policy": "_turn_policy",
        "ingress_hook_runner": "_ingress_hook_runner",
        "response_runner": "_response_runner",
        "delivery_gateway": "_delivery_gateway",
        "tool_runtime": "_tool_runtime_support",
        "turn_store": "_turn_store",
        "edit_regenerator": "_edit_regenerator",
        "visible_responses": "_visible_responses",
    }
    for field_name, attr_name in default_collaborators.items():
        if field_name in rebuilt_changes:
            continue
        rebuilt_changes[field_name] = getattr(bot, attr_name)
    store_field_names = set(unwrap_extracted_collaborator(bot._turn_store).deps.__dataclass_fields__)
    store_changes = {name: value for name, value in changes.items() if name in store_field_names}
    if store_changes:
        replace_turn_store_deps(bot, **store_changes)
        rebuilt_changes["turn_store"] = bot._turn_store
    if "edit_regenerator" not in rebuilt_changes:
        rebuilt_changes["edit_regenerator"] = bot._edit_regenerator
    if "ingress" not in rebuilt_changes:
        rebuilt_changes["ingress"] = IngressValidator(
            replace(
                bot._ingress_validator.deps,
                turn_store=rebuilt_changes["turn_store"],
                turn_policy=rebuilt_changes["turn_policy"],
            ),
        )
    bot._ingress_validator = rebuilt_changes["ingress"]
    visible_voice_echo = unwrap_extracted_collaborator(bot._visible_voice_echo)
    bot._visible_voice_echo = VisibleVoiceEchoLifecycle(
        replace(
            visible_voice_echo.deps,
            runtime=rebuilt_changes.get("runtime", controller.deps.runtime),
            logger=rebuilt_changes.get("logger", controller.deps.logger),
            agent_name=rebuilt_changes.get("agent_name", controller.deps.agent_name),
            delivery_gateway=rebuilt_changes["delivery_gateway"],
            turn_store=rebuilt_changes["turn_store"],
            ingress=rebuilt_changes["ingress"],
        ),
    )
    wrap_extracted_collaborators(bot, "_visible_voice_echo")
    rebuilt_changes["visible_voice_echo"] = bot._visible_voice_echo
    visible_responses = unwrap_extracted_collaborator(bot._visible_responses)
    visible_response_changes = {
        name: value
        for name, value in changes.items()
        if name in visible_responses.deps.__dataclass_fields__
        and name not in {"runtime", "logger", "turn_store", "delivery_gateway"}
    }
    bot._visible_responses = VisibleResponseReconciler(
        replace(
            visible_responses.deps,
            runtime=rebuilt_changes.get("runtime", controller.deps.runtime),
            logger=rebuilt_changes.get("logger", controller.deps.logger),
            turn_store=rebuilt_changes["turn_store"],
            delivery_gateway=rebuilt_changes["delivery_gateway"],
            **visible_response_changes,
        ),
    )
    wrap_extracted_collaborators(bot, "_visible_responses")
    rebuilt_changes["visible_responses"] = bot._visible_responses
    command_executor = unwrap_extracted_collaborator(bot._command_turn_executor)
    command_changes = {
        name: value
        for name, value in changes.items()
        if name in command_executor.deps.__dataclass_fields__
        and name
        not in {
            "runtime",
            "logger",
            "runtime_paths",
            "agent_name",
            "normalizer",
            "conversation_cache",
            "turn_policy",
            "turn_store",
            "visible_responses",
        }
    }
    bot._command_turn_executor = CommandTurnExecutor(
        replace(
            command_executor.deps,
            runtime=rebuilt_changes.get("runtime", controller.deps.runtime),
            logger=rebuilt_changes.get("logger", controller.deps.logger),
            runtime_paths=rebuilt_changes.get("runtime_paths", controller.deps.runtime_paths),
            agent_name=rebuilt_changes.get("agent_name", controller.deps.agent_name),
            normalizer=rebuilt_changes["normalizer"],
            conversation_cache=rebuilt_changes["conversation_cache"],
            turn_policy=rebuilt_changes["turn_policy"],
            turn_store=rebuilt_changes["turn_store"],
            visible_responses=rebuilt_changes["visible_responses"],
            **command_changes,
        ),
    )
    wrap_extracted_collaborators(bot, "_command_turn_executor")
    rebuilt_changes["command_executor"] = bot._command_turn_executor
    user_stop_reconciler = unwrap_extracted_collaborator(bot._user_stop_reconciler)
    bot._user_stop_reconciler = UserStopReconciler(
        replace(
            user_stop_reconciler.deps,
            turn_store=rebuilt_changes["turn_store"],
            response_runner=rebuilt_changes["response_runner"],
            delivery_gateway=rebuilt_changes["delivery_gateway"],
        ),
    )
    wrap_extracted_collaborators(bot, "_user_stop_reconciler")
    rebuilt = TurnController(replace(controller.deps, **rebuilt_changes))
    bot._turn_controller = rebuilt
    reaction_dispatcher = bot._reaction_dispatcher
    replace_reaction_dispatcher_deps(
        bot,
        runtime=rebuilt.deps.runtime,
        logger=rebuilt.deps.logger,
        runtime_paths=rebuilt.deps.runtime_paths,
        agent_name=rebuilt.deps.agent_name,
        turn_policy=rebuilt.deps.turn_policy,
        turn_store=rebuilt.deps.turn_store,
        conversation_cache=rebuilt.deps.conversation_cache,
        user_stop_reconciler=bot._user_stop_reconciler,
        ingress=rebuilt.deps.ingress,
        stop_manager=bot.stop_manager,
        reserve_prompt_ingress_order=rebuilt.reserve_prompt_ingress_order,
        handle_interactive_selection=rebuilt.handle_interactive_selection,
        config_confirmation=replace(
            reaction_dispatcher.deps.config_confirmation,
            runtime=rebuilt.deps.runtime,
            runtime_paths=rebuilt.deps.runtime_paths,
            build_message_target=rebuilt.deps.resolver.build_message_target,
            delivery_gateway=rebuilt.deps.delivery_gateway,
        ),
    )
    edit_changes = {
        name: value
        for name, value in changes.items()
        if name in unwrap_extracted_collaborator(bot._edit_regenerator).deps.__dataclass_fields__
    }
    if edit_changes:
        replace_edit_regenerator_deps(bot, **edit_changes)
    return rebuilt


@contextmanager
def patch_response_runner_module(**changes: object) -> Generator[None, None, None]:
    """Patch module-level response coordinator seams on the real current owner."""
    with ExitStack() as stack:
        for name, replacement in changes.items():
            module_name = (
                "mindroom.response_lifecycle" if name == "apply_post_response_effects" else "mindroom.response_runner"
            )
            stack.enter_context(patch(f"{module_name}.{name}", new=replacement))
        yield


def install_shutdown_drain_mocks(
    bot: RuntimeBot,
    *,
    coalescing_drain_result: CoalescingDrainResult,
    responses_drained: bool,
    response_recovery_complete: bool,
) -> None:
    """Install exact shutdown drain outcomes through stable collaborator seams."""
    wrap_extracted_collaborators(bot, "_coalescing_gate", "_response_runner")
    bot._coalescing_gate.drain_all = AsyncMock(
        return_value=coalescing_drain_result,
    )
    bot._response_runner.drain_inbox_responses = AsyncMock(return_value=responses_drained)
    unwrap_extracted_collaborator(
        bot._response_runner,
    )._incomplete_inbox_responses_recoverable = response_recovery_complete


def install_send_response_mock(bot: RuntimeBot, send_response: AsyncMock) -> None:
    """Route visible delivery through one target-explicit send-response mock."""
    wrap_extracted_collaborators(bot, "_delivery_gateway")

    async def _send_text(request: SendTextRequest) -> str | None:
        return await send_response(
            target=request.target,
            response_text=request.response_text,
            skip_mentions=request.skip_mentions,
            tool_trace=request.tool_trace,
            extra_content=request.extra_content,
        )

    bot._delivery_gateway.send_text = AsyncMock(side_effect=_send_text)

    async def _deliver_final(request: FinalDeliveryRequest) -> FinalDeliveryOutcome:
        event_id = await send_response(
            target=request.target,
            response_text=request.response_text,
            skip_mentions=request.skip_mentions,
            tool_trace=request.tool_trace,
            extra_content=request.extra_content,
        )
        delivery_kind = "edited" if request.existing_event_id is not None else "sent"
        if event_id is None:
            if request.existing_event_id is not None:
                return _outcome(
                    terminal_status="error",
                    event_id=request.existing_event_id,
                    is_visible_response=True,
                    final_visible_body=request.response_text,
                    failure_reason="test_mock_no_visible_response",
                    extra_content=request.extra_content,
                )
            return _outcome(
                terminal_status="error",
                failure_reason="test_mock_no_visible_response",
            )
        return _outcome(
            terminal_status="completed",
            event_id=event_id,
            is_visible_response=True,
            final_visible_body=request.response_text,
            delivery_kind=delivery_kind,
            extra_content=request.extra_content,
        )

    bot._delivery_gateway.deliver_final = AsyncMock(side_effect=_deliver_final)
    replace_turn_controller_deps(bot, delivery_gateway=bot._delivery_gateway)
    replace_response_runner_deps(bot, delivery_gateway=bot._delivery_gateway)


def install_generate_response_mock(bot: RuntimeBot, generate_response: AsyncMock) -> None:
    """Route response execution through one envelope-explicit generate-response mock."""
    wrap_extracted_collaborators(bot, "_response_runner")

    def _resolved_event_id_from_test_result(
        result: FinalDeliveryOutcome | str | None,
    ) -> str | None:
        if isinstance(result, FinalDeliveryOutcome):
            return result.final_visible_event_id
        return result

    async def _generate(request: ResponseRequest) -> str | None:
        if request.payload_preparation is not None:
            try:
                request = await bot._request_payload_preparer.prepare(request)
            except Exception as exc:
                raise PostLockRequestPreparationError from exc
        attachment_ids = list(request.attachment_ids) if request.attachment_ids is not None else None
        result = await generate_response(
            prompt=request.prompt,
            thread_history=request.thread_history,
            existing_event_id=request.existing_event_id,
            existing_event_is_placeholder=request.existing_event_is_placeholder,
            user_id=request.user_id,
            media=request.media,
            attachment_ids=attachment_ids,
            model_prompt=request.model_prompt,
            transient_enrichment_items=request.transient_enrichment_items,
            system_enrichment_items=request.system_enrichment_items,
            response_envelope=request.response_envelope,
            correlation_id=request.correlation_id,
            matrix_run_metadata=request.matrix_run_metadata,
        )
        return _resolved_event_id_from_test_result(result)

    bot._response_runner.generate_response = AsyncMock(side_effect=_generate)
    replace_turn_controller_deps(bot, response_runner=bot._response_runner)


def install_edit_message_mock(bot: RuntimeBot, edit_message: AsyncMock) -> None:
    """Route Matrix edits through one argument-expanded edit-message mock."""
    wrap_extracted_collaborators(bot, "_delivery_gateway")

    async def _edit_text(request: EditTextRequest) -> bool:
        return await edit_message(
            request.target.room_id,
            request.event_id,
            request.new_text,
            request.target.resolved_thread_id,
            tool_trace=request.tool_trace,
            extra_content=request.extra_content,
        )

    bot._delivery_gateway.edit_text = AsyncMock(side_effect=_edit_text)
    replace_turn_controller_deps(bot, delivery_gateway=bot._delivery_gateway)
    replace_response_runner_deps(bot, delivery_gateway=bot._delivery_gateway)


@pytest.fixture
def build_private_template_dir(tmp_path: Path) -> Callable[..., Path]:
    """Return a helper that creates a local private-instance template directory."""

    def _build(
        name: str = "private_template",
        *,
        files: dict[str, str] | None = None,
    ) -> Path:
        template_dir = tmp_path / name
        template_dir.mkdir(parents=True, exist_ok=True)
        template_files = files or {
            "SOUL.md": "Template soul.\n",
            "USER.md": "Template user.\n",
            "MEMORY.md": "# Memory\n",
            "memory/notes.md": "Private note.\n",
        }
        for relative_path, content in template_files.items():
            destination = template_dir / relative_path
            destination.parent.mkdir(parents=True, exist_ok=True)
            destination.write_text(content, encoding="utf-8")
        return template_dir

    return _build


@pytest_asyncio.fixture
async def aioresponse() -> AsyncGenerator[aioresponses, None]:
    """Async fixture for mocking HTTP responses in tests."""
    # Based on https://github.com/matrix-nio/matrix-nio/blob/main/tests/conftest_async.py
    with aioresponses() as m:
        yield m


@pytest.fixture(autouse=True)
def _isolate_structlog_configuration(
    monkeypatch: pytest.MonkeyPatch,
    request: pytest.FixtureRequest,
) -> Generator[None, None, None]:
    """Silence incidental logs and prevent global logging configuration leaks."""
    _configure_quiet_structlog()
    if request.node.path.name != "test_logging_config.py":
        monkeypatch.setattr(structlog, "configure", _configure_uncached_structlog)
    yield
    _configure_quiet_structlog()


@pytest.fixture(autouse=True)
def _pin_matrix_homeserver(monkeypatch: pytest.MonkeyPatch) -> None:
    """Keep test runtime defaults isolated from shell-level runtime overrides.

    Tests use ':localhost' Matrix IDs and non-namespaced localparts unless they
    explicitly opt into a different runtime context.
    """
    monkeypatch.delenv("MATRIX_HOMESERVER", raising=False)
    monkeypatch.delenv("MATRIX_SERVER_NAME", raising=False)
    monkeypatch.delenv("MINDROOM_NAMESPACE", raising=False)
    monkeypatch.delenv("MINDROOM_CONFIG_PATH", raising=False)
    monkeypatch.delenv("MINDROOM_STORAGE_PATH", raising=False)


@pytest.fixture(autouse=True)
def _reset_runtime_paths() -> Generator[None, None, None]:
    """Restore process env and bound test runtime mappings after each test."""
    original_env = os.environ.copy()
    original_bound_configs = dict(_TEST_RUNTIME_PATHS_BY_CONFIG_ID)
    yield
    os.environ.clear()
    os.environ.update(original_env)
    _TEST_RUNTIME_PATHS_BY_CONFIG_ID.clear()
    _TEST_RUNTIME_PATHS_BY_CONFIG_ID.update(original_bound_configs)


@pytest.fixture(autouse=True)
def _reset_model_media_capabilities() -> Generator[None, None, None]:
    """Keep process-local learned media support isolated per test."""
    reset_model_media_capability_cache()
    yield
    reset_model_media_capability_cache()


@pytest.fixture(autouse=True)
def _reset_voice_echo_barriers() -> Generator[None, None, None]:
    """Keep cross-bot voice echo ordering state, and its loop bindings, per test."""
    _reset_visible_voice_echo_barriers()
    yield
    _reset_visible_voice_echo_barriers()


@pytest.fixture(autouse=True)
def bypass_authorization(request: pytest.FixtureRequest) -> Generator[None, None, None]:
    """Bypass authorization checks in tests by default.

    This allows test users like @user:example.com to interact with agents
    without needing to be in the authorized_users list.

    Tests in test_authorization.py are excluded since they test authorization itself.
    """
    # Don't bypass authorization for tests that are specifically testing it
    if "test_authorization" in request.node.parent.name:
        yield
    else:
        with (
            patch("mindroom.bot.is_authorized_sender", return_value=True),
            patch("mindroom.ingress_validation.is_authorized_sender", return_value=True),
            patch("mindroom.reaction_dispatch.is_authorized_sender", return_value=True),
        ):
            yield
