"""Tests for config auto-reload and room membership updates."""

from __future__ import annotations

import asyncio
import json
import sys
import tempfile
from contextlib import suppress
from dataclasses import replace
from pathlib import Path
from typing import TYPE_CHECKING
from unittest.mock import AsyncMock, MagicMock, patch

import pytest
import yaml

import mindroom.orchestrator as orchestrator_module
import mindroom.tool_system.plugin_imports as plugin_module
from mindroom.bot import AgentBot
from mindroom.config.agent import AgentConfig, CultureConfig, RoomConfig, TeamConfig
from mindroom.config.calls import CallsConfig, RealtimeCallProfile
from mindroom.config.main import Config
from mindroom.config.models import ModelConfig, RouterConfig
from mindroom.constants import ROUTER_AGENT_NAME, STREAM_STATUS_KEY, STREAM_STATUS_PENDING
from mindroom.file_watcher import _tree_snapshot
from mindroom.hooks import EVENT_MESSAGE_RECEIVED, HookRegistry
from mindroom.matrix.client_room_admin import RoomJoinOutcome
from mindroom.matrix.state import MatrixState
from mindroom.matrix.users import AgentMatrixUser
from mindroom.message_target import MessageTarget
from mindroom.orchestration.config_updates import ConfigUpdatePlan, _get_changed_agents, build_config_update_plan
from mindroom.orchestration.plugin_watch import (
    _collect_plugin_root_changes,
    _drop_unconfigured_plugin_root_snapshots,
    watch_plugins_task,
)
from mindroom.orchestration.runtime import log_startup_phase_finished, log_startup_phase_started
from mindroom.orchestrator import _MultiAgentOrchestrator, _watch_skills_task
from mindroom.response_admission import ResponseAdmissionGate, ResponseAdmissionRefusedError
from mindroom.response_runner import ResponseRequest
from mindroom.runtime_shutdown import SYNC_RESTART_SHUTDOWN
from mindroom.startup_errors import PermanentStartupError
from mindroom.tool_system.plugins import PluginReloadResult
from mindroom.tool_system.skills import _get_plugin_skill_roots, set_plugin_skill_roots
from tests.conftest import (
    TEST_PASSWORD,
    bind_runtime_paths,
    install_send_response_mock,
    make_event_cache_mock,
    make_event_cache_write_coordinator_mock,
    orchestrator_runtime_paths,
    request_envelope,
    runtime_paths_for,
    test_runtime_paths,
    unwrap_extracted_collaborator,
)

if TYPE_CHECKING:
    from collections.abc import Iterable


def _calls_for(
    agent_name: str,
    *,
    enabled: bool = True,
    model: str = "gpt-realtime",
    voice: str = "marin",
) -> CallsConfig:
    return CallsConfig(
        enabled=enabled,
        profiles={
            "voice": RealtimeCallProfile(
                backend="realtime",
                model=model,
                credentials_service="openai",
                voice=voice,
            ),
        },
        agents={agent_name: "voice"},
    )


def _runtime_bound_config(config: Config, runtime_root: Path | None = None) -> Config:
    """Return a runtime-bound config for reload tests."""
    runtime_paths = test_runtime_paths(runtime_root or Path(tempfile.mkdtemp()))
    return bind_runtime_paths(config, runtime_paths)


def setup_test_bot(bot: AgentBot, mock_client: AsyncMock) -> None:
    """Helper to setup a test bot with required attributes."""
    bot.client = mock_client
    bot.event_cache = make_event_cache_mock()
    bot.event_cache_write_coordinator = make_event_cache_write_coordinator_mock()


def test_startup_phase_logging_helpers_emit_structured_timing(monkeypatch: pytest.MonkeyPatch) -> None:
    """Startup phase logging helpers should emit one stable structured event pair."""
    logger_mock = MagicMock()
    monkeypatch.setattr("mindroom.orchestration.runtime.logger", logger_mock)

    with patch("mindroom.orchestration.runtime.time.monotonic", side_effect=[10.0, 10.1234]):
        started_at = log_startup_phase_started("phase.name")
        log_startup_phase_finished("phase.name", started_at, status="failed")

    assert started_at == 10.0
    logger_mock.info.assert_any_call("startup_phase_started", phase="phase.name")
    logger_mock.info.assert_any_call(
        "startup_phase_finished",
        phase="phase.name",
        status="failed",
        elapsed_ms=123.4,
    )


@pytest.mark.asyncio
async def test_watch_skills_task_snapshots_off_event_loop(monkeypatch: pytest.MonkeyPatch) -> None:
    """The skills watcher should not run recursive skill snapshots on the asyncio loop."""

    class DummyOrchestrator:
        running = True

    dummy_orchestrator = DummyOrchestrator()
    snapshot_function_calls: list[object] = []
    clear_calls = 0

    def direct_snapshot_call() -> object:
        raise AssertionError

    async def fake_to_thread(function: object, *args: object, **kwargs: object) -> tuple[tuple[str, int, int], ...]:
        del args, kwargs
        snapshot_function_calls.append(function)
        if len(snapshot_function_calls) == 1:
            return (("before", 1, 1),)
        dummy_orchestrator.running = False
        return (("after", 2, 2),)

    async def fake_sleep(delay: float) -> None:
        del delay

    def fake_clear_skill_cache() -> None:
        nonlocal clear_calls
        clear_calls += 1

    monkeypatch.setattr(orchestrator_module, "get_skill_snapshot", direct_snapshot_call)
    monkeypatch.setattr(orchestrator_module.asyncio, "to_thread", fake_to_thread)
    monkeypatch.setattr(orchestrator_module.asyncio, "sleep", fake_sleep)
    monkeypatch.setattr(orchestrator_module, "clear_skill_cache", fake_clear_skill_cache)

    await _watch_skills_task(dummy_orchestrator)  # type: ignore[arg-type]

    assert snapshot_function_calls == [direct_snapshot_call, direct_snapshot_call]
    assert clear_calls == 1


def _write_plugin_removal_test_files(tmp_path: Path) -> Path:
    """Create one minimal plugin used by config-reload teardown tests."""
    plugin_root = tmp_path / "plugins" / "removed-task-plugin"
    plugin_root.mkdir(parents=True)
    (plugin_root / "mindroom.plugin.json").write_text(
        json.dumps({"name": "removed-task-plugin", "hooks_module": "hooks.py", "skills": []}),
        encoding="utf-8",
    )
    hooks_path = plugin_root / "hooks.py"
    hooks_path.write_text(
        "from mindroom.hooks import hook\n"
        "\n"
        "@hook('message:received')\n"
        "async def audit(ctx):\n"
        "    del ctx\n"
        "    return 'ok'\n",
        encoding="utf-8",
    )
    return hooks_path


def _write_plugin_removal_test_config(tmp_path: Path, *, with_plugin: bool) -> None:
    """Write one minimal config for config-reload plugin teardown tests."""
    config_data = {
        "models": {"default": {"provider": "anthropic", "id": "claude-sonnet-4-6"}},
        "router": {"model": "default"},
        "agents": {
            "assistant": {
                "display_name": "Assistant",
                "role": "Helpful",
                "model": "default",
                "rooms": ["lobby"],
            },
        },
        "authorization": {"global_users": ["@owner:localhost"]},
        "plugins": [{"path": "./plugins/removed-task-plugin"}] if with_plugin else [],
    }
    (tmp_path / "config.yaml").write_text(
        yaml.safe_dump(config_data, sort_keys=False),
        encoding="utf-8",
    )


async def _noop_prepare_user_account(
    self: _MultiAgentOrchestrator,
    config: Config,
    *,
    update_runtime_state: bool,
) -> None:
    del self, config, update_runtime_state


async def _noop_prepare_entity_accounts(
    self: _MultiAgentOrchestrator,
    config: Config,
    entity_names: Iterable[str],
) -> dict[str, AgentMatrixUser]:
    del self
    return {
        entity_name: AgentMatrixUser(
            agent_name=entity_name,
            user_id=f"@actual_{entity_name}:localhost",
            display_name=(
                "RouterAgent"
                if entity_name == ROUTER_AGENT_NAME
                else config.agents[entity_name].display_name
                if entity_name in config.agents
                else config.teams[entity_name].display_name
            ),
            password=TEST_PASSWORD,
        )
        for entity_name in entity_names
    }


async def _noop_sync_mcp_manager(
    self: _MultiAgentOrchestrator,
    config: Config,
) -> set[str]:
    del self, config
    return set()


async def _noop_sync_event_cache_service(
    self: _MultiAgentOrchestrator,
    config: Config,
) -> None:
    del self, config


async def _noop_sync_runtime_support_services(
    self: _MultiAgentOrchestrator,
    config: Config,
    *,
    start_watcher: bool,
    previous_config: Config | None = None,
) -> None:
    del self, config, start_watcher, previous_config


async def _noop_setup_rooms_and_memberships(
    self: _MultiAgentOrchestrator,
    bots: list[AgentBot],
) -> None:
    del self, bots


async def _noop_emit_config_reloaded(
    self: _MultiAgentOrchestrator,
    *,
    new_config: Config,
    changed_entities: set[str],
    added_entities: set[str],
    removed_entities: set[str],
    plugin_changes: tuple[str, ...],
) -> None:
    del self, new_config, changed_entities, added_entities, removed_entities, plugin_changes


def _noop_start_sync_task(
    self: _MultiAgentOrchestrator,
    entity_name: str,
    bot: AgentBot,
) -> None:
    del self, entity_name, bot


async def _noop_try_start(self: AgentBot) -> bool:
    self.running = True
    return True


async def _noop_stop(self: AgentBot, *, shutdown_intent: object | None = None) -> None:
    del shutdown_intent
    self.running = False


def _patch_orchestrator_plugin_update_test_runtime(monkeypatch: pytest.MonkeyPatch) -> None:
    """Patch bot and orchestrator runtime helpers unrelated to plugin teardown."""
    monkeypatch.setattr(
        "mindroom.orchestrator._MultiAgentOrchestrator._prepare_user_account",
        _noop_prepare_user_account,
    )
    monkeypatch.setattr(
        "mindroom.orchestrator._MultiAgentOrchestrator._prepare_entity_accounts",
        _noop_prepare_entity_accounts,
    )
    monkeypatch.setattr(
        "mindroom.orchestrator._MultiAgentOrchestrator._sync_mcp_manager",
        _noop_sync_mcp_manager,
    )
    monkeypatch.setattr(
        "mindroom.orchestrator._MultiAgentOrchestrator._sync_event_cache_service",
        _noop_sync_event_cache_service,
    )
    monkeypatch.setattr(
        "mindroom.orchestrator._MultiAgentOrchestrator._sync_runtime_support_services",
        _noop_sync_runtime_support_services,
    )
    monkeypatch.setattr(
        "mindroom.orchestrator._MultiAgentOrchestrator._setup_rooms_and_memberships",
        _noop_setup_rooms_and_memberships,
    )
    monkeypatch.setattr(
        "mindroom.orchestrator._MultiAgentOrchestrator._emit_config_reloaded",
        _noop_emit_config_reloaded,
    )
    monkeypatch.setattr(
        "mindroom.orchestrator._MultiAgentOrchestrator._start_sync_task",
        _noop_start_sync_task,
    )
    monkeypatch.setattr("mindroom.bot.AgentBot.try_start", _noop_try_start)
    monkeypatch.setattr("mindroom.bot.TeamBot.try_start", _noop_try_start)
    monkeypatch.setattr("mindroom.bot.AgentBot.stop", _noop_stop)
    monkeypatch.setattr("mindroom.bot.TeamBot.stop", _noop_stop)


@pytest.mark.asyncio
async def test_plugin_watcher_debounces_changes_and_ignores_unconfigured_roots(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Plugin watcher should coalesce local edits and ignore unconfigured plugin roots."""
    monkeypatch.setattr("mindroom.file_watcher._WATCH_SCAN_INTERVAL_SECONDS", 0.01)
    monkeypatch.setattr("mindroom.file_watcher._WATCH_TREE_DEBOUNCE_SECONDS", 0.01)

    plugin_root = tmp_path / "plugins" / "demo"
    plugin_root.mkdir(parents=True)
    (plugin_root / "mindroom.plugin.json").write_text(
        '{"name": "demo", "hooks_module": "hooks.py", "skills": []}',
        encoding="utf-8",
    )
    (plugin_root / "hooks.py").write_text("VALUE = 1\n", encoding="utf-8")
    ignored_root = tmp_path / "plugins" / "ignored"
    ignored_root.mkdir(parents=True)
    (ignored_root / "hooks.py").write_text("VALUE = 1\n", encoding="utf-8")

    config = _runtime_bound_config(Config(plugins=["./plugins/demo"]), tmp_path)
    orchestrator = _MultiAgentOrchestrator(runtime_paths_for(config))
    orchestrator.config = config
    orchestrator.running = True

    reload_calls: list[tuple[str, tuple[Path, ...]]] = []
    reload_seen = asyncio.Event()

    async def record_reload(*, source: str, changed_paths: tuple[Path, ...] = (), **_: int) -> PluginReloadResult:
        reload_calls.append((source, changed_paths))
        reload_seen.set()
        return PluginReloadResult(HookRegistry.empty(), (), 0)

    orchestrator.reload_plugins_now = AsyncMock(side_effect=record_reload)
    watcher_task = asyncio.create_task(watch_plugins_task(orchestrator))
    try:
        await asyncio.sleep(0.05)
        (ignored_root / "hooks.py").write_text("VALUE = 2\n", encoding="utf-8")
        await asyncio.sleep(0.08)
        assert reload_calls == []

        (plugin_root / "hooks.py").write_text("VALUE = 2\n", encoding="utf-8")
        await asyncio.sleep(0.005)
        (plugin_root / "helper.py").write_text("VALUE = 3\n", encoding="utf-8")
        await asyncio.wait_for(reload_seen.wait(), timeout=1)
    finally:
        watcher_task.cancel()
        with suppress(asyncio.CancelledError):
            await watcher_task

    assert len(reload_calls) == 1
    source, changed_paths = reload_calls[0]
    assert source == "watcher"
    assert (plugin_root / "hooks.py") in changed_paths
    assert (plugin_root / "helper.py") in changed_paths
    assert all(path.is_relative_to(plugin_root) for path in changed_paths)


def test_plugin_watcher_drops_unconfigured_root_snapshots(tmp_path: Path) -> None:
    """Plugin watcher snapshot pruning should remove only unconfigured roots."""
    configured_root = tmp_path / "plugins" / "configured"
    removed_root = tmp_path / "plugins" / "removed"
    snapshots = {
        configured_root: {configured_root / "hooks.py": 1},
        removed_root: {removed_root / "hooks.py": 1},
    }

    _drop_unconfigured_plugin_root_snapshots((configured_root,), snapshots)

    assert snapshots == {configured_root: {configured_root / "hooks.py": 1}}


@pytest.mark.asyncio
async def test_plugin_watcher_tracks_configured_absolute_root_outside_config_dir(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Plugin watcher should follow configured absolute roots, not just config_dir/plugins."""
    monkeypatch.setattr("mindroom.file_watcher._WATCH_SCAN_INTERVAL_SECONDS", 0.01)
    monkeypatch.setattr("mindroom.file_watcher._WATCH_TREE_DEBOUNCE_SECONDS", 0.01)

    runtime_root = tmp_path / "runtime"
    external_plugin_root = tmp_path / "external-plugin"
    external_plugin_root.mkdir(parents=True)
    (external_plugin_root / "mindroom.plugin.json").write_text(
        '{"name": "external-demo", "hooks_module": "hooks.py", "skills": []}',
        encoding="utf-8",
    )
    hooks_path = external_plugin_root / "hooks.py"
    hooks_path.write_text("VALUE = 1\n", encoding="utf-8")

    config = _runtime_bound_config(Config(plugins=[str(external_plugin_root)]), runtime_root)
    orchestrator = _MultiAgentOrchestrator(runtime_paths_for(config))
    orchestrator.config = config
    orchestrator.running = True

    reload_calls: list[tuple[str, tuple[Path, ...]]] = []
    reload_seen = asyncio.Event()

    async def record_reload(*, source: str, changed_paths: tuple[Path, ...] = (), **_: int) -> PluginReloadResult:
        reload_calls.append((source, changed_paths))
        reload_seen.set()
        return PluginReloadResult(HookRegistry.empty(), (), 0)

    orchestrator.reload_plugins_now = AsyncMock(side_effect=record_reload)
    watcher_task = asyncio.create_task(watch_plugins_task(orchestrator))
    try:
        await asyncio.sleep(0.05)
        hooks_path.write_text("VALUE = 2\n", encoding="utf-8")
        await asyncio.wait_for(reload_seen.wait(), timeout=1)
    finally:
        watcher_task.cancel()
        with suppress(asyncio.CancelledError):
            await watcher_task

    assert len(reload_calls) == 1
    source, changed_paths = reload_calls[0]
    assert source == "watcher"
    assert changed_paths == (hooks_path,)


@pytest.mark.asyncio
async def test_plugin_watcher_catches_first_save_after_startup(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """A save immediately after watcher startup should still trigger one reload."""
    monkeypatch.setattr("mindroom.file_watcher._WATCH_SCAN_INTERVAL_SECONDS", 0.05)
    monkeypatch.setattr("mindroom.file_watcher._WATCH_TREE_DEBOUNCE_SECONDS", 0.01)

    plugin_root = tmp_path / "plugins" / "demo"
    plugin_root.mkdir(parents=True)
    (plugin_root / "mindroom.plugin.json").write_text(
        '{"name": "demo", "hooks_module": "hooks.py", "skills": []}',
        encoding="utf-8",
    )
    hooks_path = plugin_root / "hooks.py"
    hooks_path.write_text("VALUE = 1\n", encoding="utf-8")

    config = _runtime_bound_config(Config(plugins=["./plugins/demo"]), tmp_path)
    orchestrator = _MultiAgentOrchestrator(runtime_paths_for(config))
    orchestrator.config = config
    orchestrator.running = True

    reload_calls: list[tuple[str, tuple[Path, ...]]] = []
    reload_seen = asyncio.Event()

    async def record_reload(*, source: str, changed_paths: tuple[Path, ...] = (), **_: int) -> PluginReloadResult:
        reload_calls.append((source, changed_paths))
        reload_seen.set()
        return PluginReloadResult(HookRegistry.empty(), (), 0)

    orchestrator.reload_plugins_now = AsyncMock(side_effect=record_reload)
    watcher_task = asyncio.create_task(watch_plugins_task(orchestrator))
    try:
        await asyncio.sleep(0.01)
        hooks_path.write_text("VALUE = 2\n", encoding="utf-8")
        await asyncio.wait_for(reload_seen.wait(), timeout=1)
    finally:
        watcher_task.cancel()
        with suppress(asyncio.CancelledError):
            await watcher_task

    assert reload_calls == [("watcher", (hooks_path,))]


@pytest.mark.asyncio
async def test_plugin_watcher_catches_first_save_after_config_switch(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """A save under a newly configured plugin root should not be absorbed into its first baseline."""
    monkeypatch.setattr("mindroom.file_watcher._WATCH_SCAN_INTERVAL_SECONDS", 0.05)
    monkeypatch.setattr("mindroom.file_watcher._WATCH_TREE_DEBOUNCE_SECONDS", 0.01)

    first_root = tmp_path / "plugins" / "first"
    first_root.mkdir(parents=True)
    (first_root / "mindroom.plugin.json").write_text(
        '{"name": "first", "hooks_module": "hooks.py", "skills": []}',
        encoding="utf-8",
    )
    (first_root / "hooks.py").write_text("VALUE = 1\n", encoding="utf-8")

    second_root = tmp_path / "plugins" / "second"
    second_root.mkdir(parents=True)
    (second_root / "mindroom.plugin.json").write_text(
        '{"name": "second", "hooks_module": "hooks.py", "skills": []}',
        encoding="utf-8",
    )
    second_hooks_path = second_root / "hooks.py"
    second_hooks_path.write_text("VALUE = 1\n", encoding="utf-8")

    first_config = _runtime_bound_config(Config(plugins=["./plugins/first"]), tmp_path)
    second_config = _runtime_bound_config(Config(plugins=["./plugins/second"]), tmp_path)
    orchestrator = _MultiAgentOrchestrator(runtime_paths_for(first_config))
    orchestrator.config = first_config
    orchestrator.running = True

    reload_calls: list[tuple[str, tuple[Path, ...]]] = []
    reload_seen = asyncio.Event()

    async def record_reload(*, source: str, changed_paths: tuple[Path, ...] = (), **_: int) -> PluginReloadResult:
        reload_calls.append((source, changed_paths))
        reload_seen.set()
        return PluginReloadResult(HookRegistry.empty(), (), 0)

    orchestrator.reload_plugins_now = AsyncMock(side_effect=record_reload)
    watcher_task = asyncio.create_task(watch_plugins_task(orchestrator))
    try:
        await asyncio.sleep(0.08)
        orchestrator.config = second_config
        orchestrator.plugin_watch.sync_roots(second_config)
        await asyncio.sleep(0.01)
        second_hooks_path.write_text("VALUE = 2\n", encoding="utf-8")
        await asyncio.wait_for(reload_seen.wait(), timeout=1)
    finally:
        watcher_task.cancel()
        with suppress(asyncio.CancelledError):
            await watcher_task

    assert len(reload_calls) == 1
    source, changed_paths = reload_calls[0]
    assert source == "watcher"
    assert second_hooks_path in changed_paths
    assert all(path.is_relative_to(second_root) for path in changed_paths)


@pytest.mark.asyncio
async def test_plugin_watcher_does_not_reload_on_config_switch_without_plugin_edit(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Switching configured plugin roots should not trigger a watcher reload on its own."""
    monkeypatch.setattr("mindroom.file_watcher._WATCH_SCAN_INTERVAL_SECONDS", 0.05)
    monkeypatch.setattr("mindroom.file_watcher._WATCH_TREE_DEBOUNCE_SECONDS", 0.01)

    first_root = tmp_path / "plugins" / "first"
    first_root.mkdir(parents=True)
    (first_root / "mindroom.plugin.json").write_text(
        '{"name": "first", "hooks_module": "hooks.py", "skills": []}',
        encoding="utf-8",
    )
    (first_root / "hooks.py").write_text("VALUE = 1\n", encoding="utf-8")

    second_root = tmp_path / "plugins" / "second"
    second_root.mkdir(parents=True)
    (second_root / "mindroom.plugin.json").write_text(
        '{"name": "second", "hooks_module": "hooks.py", "skills": []}',
        encoding="utf-8",
    )
    (second_root / "hooks.py").write_text("VALUE = 1\n", encoding="utf-8")

    first_config = _runtime_bound_config(Config(plugins=["./plugins/first"]), tmp_path)
    second_config = _runtime_bound_config(Config(plugins=["./plugins/second"]), tmp_path)
    orchestrator = _MultiAgentOrchestrator(runtime_paths_for(first_config))
    orchestrator.config = first_config
    orchestrator.running = True

    reload_calls: list[tuple[str, tuple[Path, ...]]] = []

    async def record_reload(*, source: str, changed_paths: tuple[Path, ...] = (), **_: int) -> PluginReloadResult:
        reload_calls.append((source, changed_paths))
        return PluginReloadResult(HookRegistry.empty(), (), 0)

    orchestrator.reload_plugins_now = AsyncMock(side_effect=record_reload)
    watcher_task = asyncio.create_task(watch_plugins_task(orchestrator))
    try:
        await asyncio.sleep(0.08)
        orchestrator.config = second_config
        orchestrator.plugin_watch.sync_roots(second_config)
        await asyncio.sleep(0.25)
    finally:
        watcher_task.cancel()
        with suppress(asyncio.CancelledError):
            await watcher_task

    assert reload_calls == []


@pytest.mark.asyncio
async def test_plugin_watcher_ignores_cache_artifacts_created_during_reload(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Plugin watcher should not self-trigger on cache artifacts created during reload."""
    monkeypatch.setattr("mindroom.file_watcher._WATCH_SCAN_INTERVAL_SECONDS", 0.01)
    monkeypatch.setattr("mindroom.file_watcher._WATCH_TREE_DEBOUNCE_SECONDS", 0.01)

    plugin_root = tmp_path / "plugins" / "demo"
    plugin_root.mkdir(parents=True)
    (plugin_root / "mindroom.plugin.json").write_text(
        '{"name": "demo", "hooks_module": "hooks.py", "skills": []}',
        encoding="utf-8",
    )
    hooks_path = plugin_root / "hooks.py"
    hooks_path.write_text("VALUE = 1\n", encoding="utf-8")

    config = _runtime_bound_config(Config(plugins=["./plugins/demo"]), tmp_path)
    orchestrator = _MultiAgentOrchestrator(runtime_paths_for(config))
    orchestrator.config = config
    orchestrator.running = True

    reload_calls: list[tuple[str, tuple[Path, ...]]] = []
    reload_seen = asyncio.Event()

    async def record_reload(*, source: str, changed_paths: tuple[Path, ...] = (), **_: int) -> PluginReloadResult:
        reload_calls.append((source, changed_paths))
        if len(reload_calls) == 1:
            pycache_dir = plugin_root / "__pycache__"
            pycache_dir.mkdir(exist_ok=True)
            (pycache_dir / "hooks.cpython-312.pyc").write_bytes(b"compiled")
            ruff_cache_dir = plugin_root / ".ruff_cache"
            ruff_cache_dir.mkdir(exist_ok=True)
            (ruff_cache_dir / "hooks.json").write_text("{}", encoding="utf-8")
            mypy_cache_dir = plugin_root / ".mypy_cache"
            mypy_cache_dir.mkdir(exist_ok=True)
            (mypy_cache_dir / "meta.json").write_text("{}", encoding="utf-8")
            pytest_cache_dir = plugin_root / ".pytest_cache" / "v" / "cache"
            pytest_cache_dir.mkdir(parents=True, exist_ok=True)
            (pytest_cache_dir / "nodeids").write_text("[]", encoding="utf-8")
            reload_seen.set()
        return PluginReloadResult(HookRegistry.empty(), (), 0)

    orchestrator.reload_plugins_now = AsyncMock(side_effect=record_reload)
    watcher_task = asyncio.create_task(watch_plugins_task(orchestrator))
    try:
        await asyncio.sleep(0.05)
        hooks_path.write_text("VALUE = 2\n", encoding="utf-8")
        await asyncio.wait_for(reload_seen.wait(), timeout=1)
        await asyncio.sleep(0.08)
    finally:
        watcher_task.cancel()
        with suppress(asyncio.CancelledError):
            await watcher_task

    assert len(reload_calls) == 1
    source, changed_paths = reload_calls[0]
    assert source == "watcher"
    assert hooks_path in changed_paths
    assert all("__pycache__" not in path.parts for path in changed_paths)
    assert all(".ruff_cache" not in path.parts for path in changed_paths)
    assert all(".mypy_cache" not in path.parts for path in changed_paths)
    assert all(".pytest_cache" not in path.parts for path in changed_paths)


@pytest.mark.asyncio
async def test_manual_plugin_reload_consumes_pending_watcher_changes(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """A manual reload should consume the current dirty save so the watcher does not reload it again."""
    monkeypatch.setattr("mindroom.file_watcher._WATCH_SCAN_INTERVAL_SECONDS", 0.01)
    monkeypatch.setattr("mindroom.file_watcher._WATCH_TREE_DEBOUNCE_SECONDS", 0.05)

    plugin_root = tmp_path / "plugins" / "demo"
    plugin_root.mkdir(parents=True)
    (plugin_root / "mindroom.plugin.json").write_text(
        '{"name": "demo", "hooks_module": "hooks.py", "skills": []}',
        encoding="utf-8",
    )
    hooks_path = plugin_root / "hooks.py"
    hooks_path.write_text("VALUE = 1\n", encoding="utf-8")

    config = _runtime_bound_config(Config(plugins=["./plugins/demo"]), tmp_path)
    orchestrator = _MultiAgentOrchestrator(runtime_paths_for(config))
    orchestrator.config = config
    orchestrator.running = True

    reload_call_count = 0

    def record_reload(config: Config, runtime_paths: object) -> PluginReloadResult:
        nonlocal reload_call_count
        del config, runtime_paths
        reload_call_count += 1
        return PluginReloadResult(HookRegistry.empty(), ("demo",), 0)

    watcher_task = asyncio.create_task(watch_plugins_task(orchestrator))
    try:
        await asyncio.sleep(0.05)
        with patch("mindroom.orchestrator.reload_plugins", side_effect=record_reload):
            hooks_path.write_text("VALUE = 2\n", encoding="utf-8")
            await asyncio.sleep(0.02)
            await orchestrator.reload_plugins_now(source="command")
            await asyncio.sleep(0.12)
    finally:
        watcher_task.cancel()
        with suppress(asyncio.CancelledError):
            await watcher_task

    assert reload_call_count == 1


def _plugin_watcher_race_runtime(tmp_path: Path) -> tuple[_MultiAgentOrchestrator, Config, Path, Path]:
    plugin_root = tmp_path / "plugin"
    plugin_root.mkdir()
    (plugin_root / "mindroom.plugin.json").write_text('{"name":"demo","skills":[]}', encoding="utf-8")
    hooks_path = plugin_root / "hooks.py"
    hooks_path.write_text("VALUE = 1\n", encoding="utf-8")
    config = _runtime_bound_config(Config(plugins=[str(plugin_root)]), tmp_path)
    orchestrator = _MultiAgentOrchestrator(runtime_paths_for(config))
    orchestrator.config = config
    orchestrator.running = True
    return orchestrator, config, plugin_root, hooks_path


@pytest.mark.asyncio
async def test_plugin_watcher_drops_pending_changes_when_reload_commits_during_scan(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """A reload committing mid-scan should consume pending edits."""
    monkeypatch.setattr("mindroom.file_watcher._WATCH_SCAN_INTERVAL_SECONDS", 0.0)
    monkeypatch.setattr("mindroom.file_watcher._WATCH_TREE_DEBOUNCE_SECONDS", 0.0)
    orchestrator, config, plugin_root, hooks_path = _plugin_watcher_race_runtime(tmp_path)
    second_scan_started = asyncio.Event()
    release_second_scan = asyncio.Event()
    collect_count = 0

    async def collect(roots: tuple[Path, ...], _snapshots: dict[Path, dict[Path, int]]) -> set[Path]:
        nonlocal collect_count
        assert roots == (plugin_root,)
        collect_count += 1
        if collect_count == 1:
            return {hooks_path}
        second_scan_started.set()
        await release_second_scan.wait()
        return set()

    monkeypatch.setattr("mindroom.orchestration.plugin_watch._collect_plugin_root_changes", collect)
    reload_mock = AsyncMock()
    orchestrator.reload_plugins_now = reload_mock
    task = asyncio.create_task(watch_plugins_task(orchestrator))
    try:
        await asyncio.wait_for(second_scan_started.wait(), timeout=1)
        orchestrator.plugin_watch.refresh(config)
        orchestrator.running = False
        release_second_scan.set()
        await task
    finally:
        task.cancel()
        release_second_scan.set()
    reload_mock.assert_not_awaited()


@pytest.mark.asyncio
async def test_manual_plugin_reload_keeps_edit_landing_while_source_loads(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """An edit during source loading should remain visible to the watcher."""
    orchestrator, _, plugin_root, hooks_path = _plugin_watcher_race_runtime(tmp_path)
    source_loaded = False

    def snapshot(_root: Path) -> dict[Path, int]:
        return {hooks_path: 2 if source_loaded else 1}

    def load_source(_config: Config, _runtime_paths: object) -> PluginReloadResult:
        nonlocal source_loaded
        source_loaded = True
        return PluginReloadResult(HookRegistry.empty(), ("demo",), 0)

    monkeypatch.setattr("mindroom.orchestration.plugin_watch.file_watcher._tree_snapshot", snapshot)
    with patch("mindroom.orchestrator.reload_plugins", side_effect=load_source):
        await orchestrator.reload_plugins_now(source="command")
    changed_paths = await _collect_plugin_root_changes((plugin_root,), orchestrator.plugin_watch.last_snapshot_by_root)
    assert changed_paths == {hooks_path}


@pytest.mark.asyncio
async def test_watcher_reload_drops_stale_revision_after_waiting_for_config_lock(tmp_path: Path) -> None:
    """A newer publication while lock-blocked should consume watcher work."""
    orchestrator, config, _, hooks_path = _plugin_watcher_race_runtime(tmp_path)
    orchestrator.plugin_watch.sync_roots(config)
    await orchestrator._config_update_lock.acquire()
    task = asyncio.create_task(
        orchestrator.reload_plugins_now(
            source="watcher",
            changed_paths=(hooks_path,),
            expected_revision=orchestrator.plugin_watch.revision,
        ),
    )
    await asyncio.sleep(0)
    orchestrator.plugin_watch.refresh(config)
    orchestrator._config_update_lock.release()
    assert await task is None


def test_plugin_tree_snapshot_ignores_git_metadata(tmp_path: Path) -> None:
    """Plugin tree snapshots should ignore Git metadata files inside watched repos."""
    plugin_root = tmp_path / "plugins" / "demo"
    plugin_root.mkdir(parents=True)
    hooks_path = plugin_root / "hooks.py"
    hooks_path.write_text("VALUE = 1\n", encoding="utf-8")
    git_dir = plugin_root / ".git"
    git_dir.mkdir()
    git_head = git_dir / "HEAD"
    git_head.write_text("ref: refs/heads/main\n", encoding="utf-8")
    git_ref = git_dir / "refs" / "heads" / "main"
    git_ref.parent.mkdir(parents=True)
    git_ref.write_text("deadbeef\n", encoding="utf-8")

    snapshot = _tree_snapshot(plugin_root)

    assert hooks_path in snapshot
    assert git_head not in snapshot
    assert git_ref not in snapshot


@pytest.mark.asyncio
async def test_plugin_watcher_does_not_retry_failed_reload_without_new_change(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """One broken save should trigger one failed reload attempt until another change arrives."""
    monkeypatch.setattr("mindroom.file_watcher._WATCH_SCAN_INTERVAL_SECONDS", 0.01)
    monkeypatch.setattr("mindroom.file_watcher._WATCH_TREE_DEBOUNCE_SECONDS", 0.01)

    plugin_root = tmp_path / "plugins" / "demo"
    plugin_root.mkdir(parents=True)
    (plugin_root / "mindroom.plugin.json").write_text(
        '{"name": "demo", "hooks_module": "hooks.py", "skills": []}',
        encoding="utf-8",
    )
    hooks_path = plugin_root / "hooks.py"
    hooks_path.write_text("VALUE = 1\n", encoding="utf-8")

    config = _runtime_bound_config(Config(plugins=["./plugins/demo"]), tmp_path)
    orchestrator = _MultiAgentOrchestrator(runtime_paths_for(config))
    orchestrator.config = config
    orchestrator.running = True

    reload_calls: list[tuple[str, tuple[Path, ...]]] = []
    first_reload_seen = asyncio.Event()
    second_reload_seen = asyncio.Event()
    error_message = "broken plugin"

    async def record_reload(*, source: str, changed_paths: tuple[Path, ...] = (), **_: int) -> PluginReloadResult:
        reload_calls.append((source, changed_paths))
        if len(reload_calls) == 1:
            first_reload_seen.set()
        if len(reload_calls) == 2:
            second_reload_seen.set()
        raise RuntimeError(error_message)

    orchestrator.reload_plugins_now = AsyncMock(side_effect=record_reload)
    watcher_task = asyncio.create_task(watch_plugins_task(orchestrator))
    try:
        await asyncio.sleep(0.05)
        hooks_path.write_text("VALUE = 2\n", encoding="utf-8")
        await asyncio.wait_for(first_reload_seen.wait(), timeout=1)
        await asyncio.sleep(0.08)

        assert len(reload_calls) == 1

        hooks_path.write_text("VALUE = 3\n", encoding="utf-8")
        await asyncio.wait_for(second_reload_seen.wait(), timeout=1)
    finally:
        watcher_task.cancel()
        with suppress(asyncio.CancelledError):
            await watcher_task

    assert len(reload_calls) == 2
    assert reload_calls[0][0] == "watcher"
    assert reload_calls[1][0] == "watcher"


@pytest.mark.asyncio
async def test_reload_plugins_now_deactivates_broken_plugin_after_failure(tmp_path: Path) -> None:
    """A failed explicit reload should deactivate the broken plugin instead of leaving old hooks live."""
    plugin_root = tmp_path / "plugins" / "broken-reload"
    plugin_root.mkdir(parents=True)
    (plugin_root / "mindroom.plugin.json").write_text(
        '{"name": "broken-reload", "hooks_module": "hooks.py", "skills": []}',
        encoding="utf-8",
    )
    hooks_path = (plugin_root / "hooks.py").resolve()
    hooks_path.write_text(
        "from mindroom.hooks import hook\n"
        "\n"
        "@hook('message:received')\n"
        "async def audit(ctx):\n"
        "    del ctx\n"
        "    return 'ok'\n",
        encoding="utf-8",
    )
    config = _runtime_bound_config(Config(plugins=["./plugins/broken-reload"]), tmp_path)
    orchestrator = _MultiAgentOrchestrator(runtime_paths_for(config))
    orchestrator.config = config
    orchestrator.running = True

    original_plugin_roots = _get_plugin_skill_roots()
    original_plugin_cache = plugin_module._PLUGIN_CACHE.copy()
    original_module_cache = plugin_module._MODULE_IMPORT_CACHE.copy()
    original_modules = set(sys.modules)
    shared_task = asyncio.create_task(asyncio.Event().wait())
    try:
        initial = await orchestrator.reload_plugins_now(source="test")
        assert initial.active_plugin_names == ("broken-reload",)
        assert len(orchestrator.hook_registry.hooks_for(EVENT_MESSAGE_RECEIVED)) == 1

        hooks_module = plugin_module._MODULE_IMPORT_CACHE[hooks_path].module
        hooks_module._AUTO_POKE_TASK = shared_task

        hooks_path.unlink()

        with pytest.raises(plugin_module.PluginValidationError, match="Plugin hooks module not found"):
            await orchestrator.reload_plugins_now(source="test")
        await asyncio.sleep(0)

        assert shared_task.cancelled()
        assert orchestrator.hook_registry.hooks_for(EVENT_MESSAGE_RECEIVED) == ()
    finally:
        await asyncio.gather(shared_task, return_exceptions=True)
        plugin_module._PLUGIN_CACHE.clear()
        plugin_module._PLUGIN_CACHE.update(original_plugin_cache)
        plugin_module._MODULE_IMPORT_CACHE.clear()
        plugin_module._MODULE_IMPORT_CACHE.update(original_module_cache)
        set_plugin_skill_roots(original_plugin_roots)
        for module_name in set(sys.modules) - original_modules:
            if module_name.startswith("mindroom_plugin_"):
                sys.modules.pop(module_name, None)


@pytest.mark.asyncio
async def test_reload_plugins_now_deactivates_all_plugins_when_degraded_reload_still_fails(
    tmp_path: Path,
) -> None:
    """Unrecoverable reload failures should clear the live plugin registry."""
    first_root = tmp_path / "plugins" / "first"
    first_root.mkdir(parents=True)
    (first_root / "mindroom.plugin.json").write_text(
        '{"name": "first", "hooks_module": "hooks.py", "skills": ["skills"]}',
        encoding="utf-8",
    )
    (first_root / "hooks.py").write_text(
        "from mindroom.hooks import hook\n"
        "\n"
        "@hook('message:received')\n"
        "async def audit(ctx):\n"
        "    del ctx\n"
        "    return 'first'\n",
        encoding="utf-8",
    )
    first_skills = first_root / "skills" / "first-skill"
    first_skills.mkdir(parents=True)
    (first_skills / "SKILL.md").write_text(
        "---\nname: first-skill\ndescription: test\n---\n",
        encoding="utf-8",
    )

    second_root = tmp_path / "plugins" / "second"
    second_root.mkdir(parents=True)
    second_manifest_path = second_root / "mindroom.plugin.json"
    second_manifest_path.write_text(
        '{"name": "second", "hooks_module": "hooks.py", "skills": []}',
        encoding="utf-8",
    )
    (second_root / "hooks.py").write_text(
        "from mindroom.hooks import hook\n"
        "\n"
        "@hook('message:received')\n"
        "async def audit(ctx):\n"
        "    del ctx\n"
        "    return 'second'\n",
        encoding="utf-8",
    )

    config = _runtime_bound_config(Config(plugins=["./plugins/first", "./plugins/second"]), tmp_path)
    orchestrator = _MultiAgentOrchestrator(runtime_paths_for(config))
    orchestrator.config = config
    orchestrator.running = True

    original_plugin_roots = _get_plugin_skill_roots()
    original_plugin_cache = plugin_module._PLUGIN_CACHE.copy()
    original_module_cache = plugin_module._MODULE_IMPORT_CACHE.copy()
    original_modules = set(sys.modules)
    try:
        initial = await orchestrator.reload_plugins_now(source="test")
        assert initial.active_plugin_names == ("first", "second")
        assert [hook.plugin_name for hook in orchestrator.hook_registry.hooks_for(EVENT_MESSAGE_RECEIVED)] == [
            "first",
            "second",
        ]
        assert _get_plugin_skill_roots() == [first_root / "skills"]

        second_manifest_path.write_text(
            '{"name": "first", "hooks_module": "hooks.py", "skills": []}',
            encoding="utf-8",
        )

        with pytest.raises(plugin_module.PluginValidationError, match="Duplicate plugin manifest names configured"):
            await orchestrator.reload_plugins_now(source="test")

        assert orchestrator.hook_registry.hooks_for(EVENT_MESSAGE_RECEIVED) == ()
        assert _get_plugin_skill_roots() == []
    finally:
        plugin_module._PLUGIN_CACHE.clear()
        plugin_module._PLUGIN_CACHE.update(original_plugin_cache)
        plugin_module._MODULE_IMPORT_CACHE.clear()
        plugin_module._MODULE_IMPORT_CACHE.update(original_module_cache)
        set_plugin_skill_roots(original_plugin_roots)
        for module_name in set(sys.modules) - original_modules:
            if module_name.startswith("mindroom_plugin_"):
                sys.modules.pop(module_name, None)


@pytest.mark.asyncio
async def test_update_config_keeps_current_config_when_new_entity_account_preparation_fails(tmp_path: Path) -> None:
    """Hot reload should not publish config until new entity accounts are prepared."""
    current_config = _runtime_bound_config(
        Config(
            agents={"general": {"display_name": "GeneralAgent", "model": "default"}},
            models={"default": {"provider": "test", "id": "test-model"}},
        ),
        tmp_path,
    )
    new_config = _runtime_bound_config(
        Config(
            agents={
                "general": {"display_name": "GeneralAgent", "model": "default"},
                "writer": {"display_name": "WriterAgent", "model": "default"},
            },
            models={"default": {"provider": "test", "id": "test-model"}},
        ),
        tmp_path,
    )
    orchestrator = _MultiAgentOrchestrator(runtime_paths_for(current_config))
    orchestrator.config = current_config
    orchestrator.agent_bots = {ROUTER_AGENT_NAME: MagicMock(), "general": MagicMock()}
    orchestrator.running = True
    account_error = PermanentStartupError("configured entities share a Matrix ID")

    with (
        patch("mindroom.orchestration.config_lifecycle.load_config", return_value=new_config),
        patch.object(orchestrator, "_prepare_entity_accounts", new=AsyncMock(side_effect=account_error)) as prepare,
        pytest.raises(PermanentStartupError, match="share a Matrix ID"),
    ):
        await orchestrator.config_reload._update_config()

    assert orchestrator.config is current_config
    prepare.assert_awaited_once_with(new_config, {"writer"})


@pytest.mark.asyncio
async def test_update_config_keeps_current_config_when_restarted_entity_account_check_fails(tmp_path: Path) -> None:
    """Hot reload should not publish config until restarted entity accounts validate."""
    current_config = _runtime_bound_config(
        Config(
            agents={"general": {"display_name": "GeneralAgent", "model": "default"}},
            models={"default": {"provider": "test", "id": "test-model"}},
        ),
        tmp_path,
    )
    new_config = _runtime_bound_config(
        Config(
            agents={"general": {"display_name": "RenamedGeneralAgent", "model": "default"}},
            models={"default": {"provider": "test", "id": "test-model"}},
        ),
        tmp_path,
    )
    orchestrator = _MultiAgentOrchestrator(runtime_paths_for(current_config))
    orchestrator.config = current_config
    orchestrator.agent_bots = {ROUTER_AGENT_NAME: MagicMock(), "general": MagicMock()}
    orchestrator.running = True
    account_error = PermanentStartupError("configured entities share a Matrix ID")

    with (
        patch("mindroom.orchestration.config_lifecycle.load_config", return_value=new_config),
        patch.object(orchestrator, "_prepare_entity_accounts", new=AsyncMock(side_effect=account_error)) as prepare,
        pytest.raises(PermanentStartupError, match="share a Matrix ID"),
    ):
        await orchestrator.config_reload._update_config()

    assert orchestrator.config is current_config
    prepare.assert_awaited_once_with(new_config, {"general"})


@pytest.mark.asyncio
async def test_initialize_keeps_config_unpublished_when_entity_account_preparation_fails(tmp_path: Path) -> None:
    """Startup should not publish runtime config before entity accounts validate."""
    config = _runtime_bound_config(
        Config(
            agents={"general": {"display_name": "GeneralAgent", "model": "default"}},
            models={"default": {"provider": "test", "id": "test-model"}},
        ),
        tmp_path,
    )
    orchestrator = _MultiAgentOrchestrator(runtime_paths_for(config))
    account_error = PermanentStartupError("configured entities share a Matrix ID")

    with (
        patch("mindroom.orchestrator.load_config", return_value=config),
        patch.object(orchestrator, "_prepare_user_account", new=AsyncMock()),
        patch.object(orchestrator, "_prepare_entity_accounts", new=AsyncMock(side_effect=account_error)),
        pytest.raises(PermanentStartupError, match="share a Matrix ID"),
    ):
        await orchestrator.initialize()

    assert orchestrator.config is None


@pytest.mark.asyncio
async def test_update_config_validates_internal_user_collision_before_publish(tmp_path: Path) -> None:
    """Internal-user reloads must recheck entity identity collisions before publishing config."""
    current_config = _runtime_bound_config(
        Config(
            agents={"general": {"display_name": "GeneralAgent", "model": "default"}},
            models={"default": {"provider": "test", "id": "test-model"}},
            mindroom_user={"username": "mindroom_user", "display_name": "Old MindRoom User"},
        ),
        tmp_path,
    )
    new_config = _runtime_bound_config(
        Config(
            agents={"general": {"display_name": "GeneralAgent", "model": "default"}},
            models={"default": {"provider": "test", "id": "test-model"}},
            mindroom_user={"username": "mindroom_user", "display_name": "New MindRoom User"},
        ),
        tmp_path,
    )
    runtime_paths = runtime_paths_for(current_config)
    state = MatrixState.load(runtime_paths=runtime_paths)
    state.add_account("agent_router", "actual_router", TEST_PASSWORD, domain="localhost")
    state.add_account("agent_general", "shared_actual", TEST_PASSWORD, domain="localhost")
    state.add_account(
        "agent_user",
        "shared_actual",
        TEST_PASSWORD,
        requested_username="mindroom_user",
        domain="localhost",
    )
    state.save(runtime_paths=runtime_paths)
    orchestrator = _MultiAgentOrchestrator(runtime_paths)
    orchestrator.config = current_config
    orchestrator.agent_bots = {ROUTER_AGENT_NAME: MagicMock(), "general": MagicMock()}
    orchestrator.running = True

    with (
        patch("mindroom.orchestration.config_lifecycle.load_config", return_value=new_config),
        patch.object(orchestrator, "_prepare_user_account", new=AsyncMock()),
        patch.object(orchestrator, "_prepare_entity_accounts", new=AsyncMock()) as prepare_entities,
        pytest.raises(PermanentStartupError, match="internal user Matrix ID"),
    ):
        await orchestrator.config_reload._update_config()

    assert orchestrator.config is current_config
    prepare_entities.assert_not_awaited()


@pytest.mark.asyncio
async def test_prepare_entity_accounts_retries_transient_create_agent_user_failure(tmp_path: Path) -> None:
    """Account preparation should retry transient Matrix provisioning failures."""
    config = _runtime_bound_config(
        Config(
            agents={"general": {"display_name": "GeneralAgent", "model": "default"}},
            models={"default": {"provider": "test", "id": "test-model"}},
        ),
        tmp_path,
    )
    runtime_paths = runtime_paths_for(config)
    orchestrator = _MultiAgentOrchestrator(runtime_paths)
    calls: list[str] = []
    test_paths = runtime_paths

    async def flaky_user(
        _homeserver: str,
        entity_name: str,
        display_name: str,
        *,
        runtime_paths: object,
        username: str | None = None,
    ) -> AgentMatrixUser:
        del runtime_paths, username
        calls.append(entity_name)
        if len(calls) == 1:
            msg = "temporary Matrix provisioning failure"
            raise RuntimeError(msg)
        state = MatrixState.load(runtime_paths=test_paths)
        actual_username = f"actual_{entity_name}"
        state.add_account(f"agent_{entity_name}", actual_username, TEST_PASSWORD, domain="localhost")
        state.save(runtime_paths=test_paths)
        return AgentMatrixUser(
            agent_name=entity_name,
            user_id=f"@{actual_username}:localhost",
            display_name=display_name,
            password=TEST_PASSWORD,
        )

    with (
        patch("mindroom.orchestration.runtime.retry_delay_seconds", return_value=0.0),
        patch("mindroom.orchestrator.create_agent_user", new=flaky_user),
    ):
        users = await orchestrator._prepare_entity_accounts(config, [ROUTER_AGENT_NAME, "general"])

    assert calls == [ROUTER_AGENT_NAME, ROUTER_AGENT_NAME, "general"]
    assert users[ROUTER_AGENT_NAME].user_id == "@actual_router:localhost"
    assert users["general"].user_id == "@actual_general:localhost"


@pytest.mark.asyncio
async def test_prepare_entity_accounts_rejects_duplicate_persisted_matrix_ids(tmp_path: Path) -> None:
    """Account preparation should fail permanently when persisted entity IDs are ambiguous."""
    config = _runtime_bound_config(
        Config(
            agents={
                "general": {"display_name": "GeneralAgent", "model": "default"},
                "writer": {"display_name": "WriterAgent", "model": "default"},
            },
            models={"default": {"provider": "test", "id": "test-model"}},
        ),
        tmp_path,
    )
    runtime_paths = runtime_paths_for(config)
    state = MatrixState.load(runtime_paths=runtime_paths)
    state.add_account("agent_router", "actual_router", TEST_PASSWORD, domain="localhost")
    state.add_account("agent_general", "shared_bot", TEST_PASSWORD, domain="localhost")
    state.add_account("agent_writer", "shared_bot", TEST_PASSWORD, domain="localhost")
    state.save(runtime_paths=runtime_paths)
    orchestrator = _MultiAgentOrchestrator(runtime_paths)
    calls: list[str] = []

    async def existing_user(
        _homeserver: str,
        entity_name: str,
        display_name: str,
        *,
        runtime_paths: object,
        username: str | None = None,
    ) -> AgentMatrixUser:
        del runtime_paths, username
        calls.append(entity_name)
        return AgentMatrixUser(
            agent_name=entity_name,
            user_id="@shared_bot:localhost" if entity_name != ROUTER_AGENT_NAME else "@actual_router:localhost",
            display_name=display_name,
            password=TEST_PASSWORD,
        )

    with (
        patch("mindroom.orchestrator.create_agent_user", new=existing_user),
        pytest.raises(PermanentStartupError, match="shared_bot"),
    ):
        await orchestrator._prepare_entity_accounts(config, [ROUTER_AGENT_NAME, "general", "writer"])

    assert calls == [ROUTER_AGENT_NAME, "general", "writer"]


@pytest.mark.asyncio
async def test_prepare_entity_accounts_rejects_internal_user_entity_id_collision(tmp_path: Path) -> None:
    """Account preparation should fail when the internal user shares an entity Matrix ID."""
    config = _runtime_bound_config(
        Config(
            agents={"general": {"display_name": "GeneralAgent", "model": "default"}},
            models={"default": {"provider": "test", "id": "test-model"}},
            mindroom_user={"username": "mindroom_user", "display_name": "MindRoomUser"},
        ),
        tmp_path,
    )
    runtime_paths = runtime_paths_for(config)
    state = MatrixState.load(runtime_paths=runtime_paths)
    state.add_account("agent_router", "actual_router", TEST_PASSWORD, domain="localhost")
    state.add_account("agent_general", "shared_actual", TEST_PASSWORD, domain="localhost")
    state.add_account(
        "agent_user",
        "shared_actual",
        TEST_PASSWORD,
        requested_username="mindroom_user",
        domain="localhost",
    )
    state.save(runtime_paths=runtime_paths)
    orchestrator = _MultiAgentOrchestrator(runtime_paths)

    async def existing_user(
        _homeserver: str,
        entity_name: str,
        display_name: str,
        *,
        runtime_paths: object,
        username: str | None = None,
    ) -> AgentMatrixUser:
        del runtime_paths, username
        user_id = "@actual_router:localhost" if entity_name == ROUTER_AGENT_NAME else "@shared_actual:localhost"
        return AgentMatrixUser(
            agent_name=entity_name,
            user_id=user_id,
            display_name=display_name,
            password=TEST_PASSWORD,
        )

    with (
        patch("mindroom.orchestrator.create_agent_user", new=existing_user),
        pytest.raises(PermanentStartupError, match="internal user Matrix ID"),
    ):
        await orchestrator._prepare_entity_accounts(config, [ROUTER_AGENT_NAME, "general"])


@pytest.mark.asyncio
async def test_update_config_cancels_tasks_for_removed_plugins(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Config-driven plugin removal should tear down the old live plugin runtime."""
    hooks_path = _write_plugin_removal_test_files(tmp_path)
    original_plugin_roots = _get_plugin_skill_roots()
    original_plugin_cache = plugin_module._PLUGIN_CACHE.copy()
    original_module_cache = plugin_module._MODULE_IMPORT_CACHE.copy()
    original_modules = set(sys.modules)
    task = asyncio.create_task(asyncio.Event().wait())
    try:
        _write_plugin_removal_test_config(tmp_path, with_plugin=True)
        _patch_orchestrator_plugin_update_test_runtime(monkeypatch)
        orchestrator = _MultiAgentOrchestrator(runtime_paths=orchestrator_runtime_paths(tmp_path))
        await orchestrator.initialize()

        hooks_module = plugin_module._MODULE_IMPORT_CACHE[hooks_path.resolve()].module
        package_root = plugin_module._MODULE_IMPORT_CACHE[hooks_path.resolve()].module_name.split(".", 1)[0]
        hooks_module._AUTO_POKE_TASK = task

        _write_plugin_removal_test_config(tmp_path, with_plugin=False)
        updated = await orchestrator.config_reload._update_config()
        await asyncio.sleep(0)

        assert updated is True
        assert task.cancelled()
        assert hooks_path.resolve() not in plugin_module._MODULE_IMPORT_CACHE
        assert not any(
            module_name == package_root or module_name.startswith(f"{package_root}.") for module_name in sys.modules
        )
        assert not orchestrator.hook_registry.has_hooks(EVENT_MESSAGE_RECEIVED)
    finally:
        if not task.done():
            task.cancel()
        await asyncio.gather(task, return_exceptions=True)
        plugin_module._PLUGIN_CACHE.clear()
        plugin_module._PLUGIN_CACHE.update(original_plugin_cache)
        plugin_module._MODULE_IMPORT_CACHE.clear()
        plugin_module._MODULE_IMPORT_CACHE.update(original_module_cache)
        set_plugin_skill_roots(original_plugin_roots)
        for module_name in set(sys.modules) - original_modules:
            if module_name.startswith("mindroom_plugin_"):
                sys.modules.pop(module_name, None)


@pytest.mark.asyncio
async def test_update_config_serializes_live_plugin_reload_against_staged_plugin_commit(
    tmp_path: Path,
) -> None:
    """A concurrent live reload must wait until the staged config reload commit finishes."""
    plugin_root = tmp_path / "plugins" / "demo"
    plugin_root.mkdir(parents=True)
    hooks_path = plugin_root / "hooks.py"
    (plugin_root / "mindroom.plugin.json").write_text(
        '{"name":"demo","hooks_module":"hooks.py","skills":[]}',
        encoding="utf-8",
    )
    hooks_path.write_text("VALUE = 1\n", encoding="utf-8")

    other_root = tmp_path / "plugins" / "other"
    other_root.mkdir(parents=True)
    (other_root / "mindroom.plugin.json").write_text(
        '{"name":"other","hooks_module":"hooks.py","skills":[]}',
        encoding="utf-8",
    )
    (other_root / "hooks.py").write_text("OTHER = 1\n", encoding="utf-8")

    current_config = _runtime_bound_config(
        Config(
            agents={"general": {"display_name": "GeneralAgent", "model": "default"}},
            models={"default": {"provider": "test", "id": "test-model"}},
            plugins=["./plugins/demo"],
        ),
        tmp_path,
    )
    new_config = _runtime_bound_config(
        Config(
            agents={"general": {"display_name": "GeneralAgent", "model": "default"}},
            models={"default": {"provider": "test", "id": "test-model"}},
            plugins=["./plugins/demo", "./plugins/other"],
        ),
        tmp_path,
    )

    orchestrator = _MultiAgentOrchestrator(runtime_paths_for(current_config))
    orchestrator.config = current_config
    orchestrator.running = True

    plan = ConfigUpdatePlan(
        new_config=new_config,
        changed_mcp_servers=set(),
        configured_entities=set(),
        entities_to_restart=set(),
        new_entities=set(),
        removed_entities=set(),
        mindroom_user_changed=False,
        matrix_room_access_changed=False,
        matrix_space_changed=False,
        authorization_changed=False,
    )

    original_plugin_roots = _get_plugin_skill_roots()
    original_plugin_cache = plugin_module._PLUGIN_CACHE.copy()
    original_module_cache = plugin_module._MODULE_IMPORT_CACHE.copy()
    original_modules = set(sys.modules)
    reload_task: asyncio.Task[PluginReloadResult] | None = None
    try:
        await orchestrator.reload_plugins_now(source="initial")
        assert plugin_module._MODULE_IMPORT_CACHE[hooks_path.resolve()].module.VALUE == 1

        async def start_blocked_reload(*_args: object, **_kwargs: object) -> set[str]:
            nonlocal reload_task
            hooks_path.write_text("VALUE = 2\n", encoding="utf-8")
            reload_task = asyncio.create_task(orchestrator.reload_plugins_now(source="interleaved"))
            await asyncio.sleep(0)
            assert reload_task is not None
            assert not reload_task.done()
            return set()

        with (
            patch("mindroom.orchestration.config_lifecycle.load_config", return_value=new_config),
            patch("mindroom.orchestration.config_lifecycle.build_config_update_plan", return_value=plan),
            patch.object(
                orchestrator,
                "_stop_entities_before_mcp_sync",
                new=AsyncMock(side_effect=start_blocked_reload),
            ),
            patch.object(orchestrator, "_sync_mcp_manager", new=AsyncMock(return_value=set())),
            patch.object(orchestrator, "_sync_event_cache_service", new=AsyncMock()),
            patch.object(orchestrator, "_sync_runtime_support_services", new=AsyncMock()),
            patch.object(orchestrator, "_emit_config_reloaded", new=AsyncMock()),
        ):
            updated = await orchestrator.config_reload._update_config()

        assert updated is False
        assert reload_task is not None
        result = await reload_task

        assert result.active_plugin_names == ("demo", "other")
        assert plugin_module._MODULE_IMPORT_CACHE[hooks_path.resolve()].module.VALUE == 2
    finally:
        if reload_task is not None:
            await asyncio.gather(reload_task, return_exceptions=True)
        plugin_module._PLUGIN_CACHE.clear()
        plugin_module._PLUGIN_CACHE.update(original_plugin_cache)
        plugin_module._MODULE_IMPORT_CACHE.clear()
        plugin_module._MODULE_IMPORT_CACHE.update(original_module_cache)
        set_plugin_skill_roots(original_plugin_roots)
        for module_name in set(sys.modules) - original_modules:
            if module_name.startswith("mindroom_plugin_"):
                sys.modules.pop(module_name, None)


@pytest.mark.asyncio
async def test_config_update_serializes_manual_plugin_reload_and_mcp_catalog_change(
    tmp_path: Path,
) -> None:
    """Manual plugin reloads and MCP restarts must wait for config application."""
    config = _runtime_bound_config(
        Config(
            agents={"agent1": AgentConfig(display_name="Agent 1", tools=["mcp_demo"])},
            mcp_servers={"demo": {"transport": "stdio", "command": "npx"}},
        ),
        tmp_path,
    )
    orchestrator = _MultiAgentOrchestrator(runtime_paths_for(config))
    orchestrator.config = config
    orchestrator.running = True
    apply_started = asyncio.Event()
    finish_apply = asyncio.Event()

    async def apply_update_plan(
        _current_config: Config,
        _plan: ConfigUpdatePlan,
        _plugin_changes: tuple[str, ...],
    ) -> bool:
        apply_started.set()
        await finish_apply.wait()
        return False

    reload_result = PluginReloadResult(HookRegistry.empty(), (), 0)
    with (
        patch("mindroom.orchestration.config_lifecycle.load_config", return_value=config),
        patch.object(orchestrator.config_reload, "apply_update_plan", new=apply_update_plan),
        patch("mindroom.orchestrator.reload_plugins", return_value=reload_result) as reload_plugins_mock,
    ):
        config_task = asyncio.create_task(orchestrator.config_reload._update_config())
        await asyncio.wait_for(apply_started.wait(), timeout=1)
        plugin_task = asyncio.create_task(orchestrator.reload_plugins_now(source="test"))
        mcp_task = asyncio.create_task(orchestrator._handle_mcp_catalog_change("demo"))
        await asyncio.sleep(0)

        assert not plugin_task.done()
        assert not mcp_task.done()
        reload_plugins_mock.assert_not_called()

        # The authoritative under-lock recheck must observe the config current
        # when the queued MCP replacement finally applies.
        replacement_config = _runtime_bound_config(Config(), tmp_path)
        orchestrator.config = replacement_config
        finish_apply.set()
        assert await config_task is False
        assert await plugin_task is reload_result
        await mcp_task

    reload_plugins_mock.assert_called_once_with(replacement_config, orchestrator.runtime_paths)


@pytest.mark.asyncio
async def test_queued_config_reload_waits_for_in_flight_response_without_event_id(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
    mock_agent_users: dict[str, AgentMatrixUser],
) -> None:
    """Queued reloads should wait for tracked responses even without a Matrix event ID."""
    monkeypatch.setattr("mindroom.orchestration.config_lifecycle._CONFIG_RELOAD_DEBOUNCE_SECONDS", 0.01)
    monkeypatch.setattr("mindroom.orchestration.config_lifecycle._REPLACEMENT_DRAIN_IDLE_POLL_SECONDS", 0.01)

    config = _runtime_bound_config(
        Config(
            agents={"agent1": AgentConfig(display_name="Agent 1")},
            router=RouterConfig(model="default"),
        ),
        tmp_path,
    )
    bot = AgentBot(
        agent_user=mock_agent_users["agent1"],
        storage_path=tmp_path,
        config=config,
        runtime_paths=runtime_paths_for(config),
    )
    setup_test_bot(bot, AsyncMock())
    send_response = AsyncMock(return_value=None)
    install_send_response_mock(bot, send_response)

    response_started = asyncio.Event()
    release_response = asyncio.Event()

    async def response_function(message_id: str | None) -> None:
        assert message_id is None
        response_started.set()
        await release_response.wait()

    runner = unwrap_extracted_collaborator(bot._response_runner)
    request = ResponseRequest(
        thread_history=(),
        prompt="Hello",
        response_envelope=request_envelope(
            room_id="!room:localhost",
            reply_to_event_id="$reply",
            agent_name=bot.agent_name,
        ),
    )

    async def run_response(
        target: MessageTarget,
        _early_placeholder: object,
    ) -> str | None:
        return await runner._run_cancellable_response(
            target=target,
            response_function=response_function,
            thinking_message="Thinking...",
        )

    orchestrator = _MultiAgentOrchestrator(runtime_paths=orchestrator_runtime_paths(tmp_path))
    orchestrator.running = True
    orchestrator.agent_bots["agent1"] = bot
    # The orchestrator owns one shared gate that every managed bot admits through.
    bot.admission_gate = orchestrator.config_reload.response_admission_gate
    orchestrator.config_reload._update_config = AsyncMock(return_value=True)

    response_task = asyncio.create_task(
        runner._run_locked_response_lifecycle(
            request,
            response_kind="ai",
            locked_operation=run_response,
        ),
    )

    try:
        await asyncio.wait_for(response_started.wait(), timeout=1)
        send_response.assert_awaited_once()
        assert bot.in_flight_response_count == 1

        orchestrator.config_reload.request_reload()
        task = orchestrator.config_reload._reload_task
        assert task is not None

        await asyncio.sleep(0.05)
        orchestrator.config_reload._update_config.assert_not_awaited()

        release_response.set()
        await asyncio.wait_for(response_task, timeout=1)
        await asyncio.wait_for(task, timeout=1)

        orchestrator.config_reload._update_config.assert_awaited_once()
    finally:
        release_response.set()
        await asyncio.gather(response_task, return_exceptions=True)
        for cleanup_task in bot.stop_manager.cleanup_tasks:
            cleanup_task.cancel()
        await asyncio.gather(*bot.stop_manager.cleanup_tasks, return_exceptions=True)
        await orchestrator.config_reload.cancel()


def test_get_changed_agents_detects_culture_config_updates() -> None:
    """Agent restarts should trigger when their culture mode/assignment changes."""
    old_config = _runtime_bound_config(
        Config(
            agents={
                "agent1": AgentConfig(display_name="Agent 1"),
            },
            cultures={
                "engineering": CultureConfig(
                    description="Engineering standards",
                    agents=["agent1"],
                    mode="automatic",
                ),
            },
        ),
    )
    new_config = _runtime_bound_config(
        Config(
            agents={
                "agent1": AgentConfig(display_name="Agent 1"),
            },
            cultures={
                "engineering": CultureConfig(
                    description="Engineering standards",
                    agents=["agent1"],
                    mode="agentic",
                ),
            },
        ),
    )

    changed = _get_changed_agents(old_config, new_config, agent_bots={"agent1": AsyncMock()})
    assert changed == {"agent1"}


def test_get_changed_agents_detects_tool_override_updates() -> None:
    """Agent restarts should trigger when authored tool overrides change."""
    old_config = _runtime_bound_config(
        Config(
            agents={
                "agent1": AgentConfig(
                    display_name="Agent 1",
                    tools=[{"shell": {"enable_run_shell_command": False}}],
                ),
            },
        ),
    )
    new_config = _runtime_bound_config(
        Config(
            agents={
                "agent1": AgentConfig(
                    display_name="Agent 1",
                    tools=[{"shell": {"enable_run_shell_command": True}}],
                ),
            },
        ),
    )

    changed = _get_changed_agents(old_config, new_config, agent_bots={"agent1": AsyncMock()})
    assert changed == {"agent1"}


def test_config_update_plan_restarts_running_entities_when_construction_prompts_change() -> None:
    """Construction-time root prompt overrides should restart running agents, teams, and router."""
    old_config = _runtime_bound_config(
        Config(
            agents={"general": AgentConfig(display_name="General Agent")},
            teams={
                "team1": TeamConfig(
                    display_name="Team 1",
                    role="Collaborate",
                    agents=["general"],
                ),
            },
            router=RouterConfig(model="default"),
        ),
    )
    new_config = _runtime_bound_config(
        Config(
            agents={"general": AgentConfig(display_name="General Agent")},
            teams={
                "team1": TeamConfig(
                    display_name="Team 1",
                    role="Collaborate",
                    agents=["general"],
                ),
            },
            router=RouterConfig(model="default"),
            prompts={"HIDDEN_TOOL_CALLS_PROMPT": "Custom hidden tool-call prompt."},
        ),
    )

    running_entities = {ROUTER_AGENT_NAME, "general", "team1"}
    plan = build_config_update_plan(
        current_config=old_config,
        new_config=new_config,
        configured_entities=running_entities,
        existing_entities=running_entities,
        agent_bots={entity: AsyncMock() for entity in running_entities},
    )

    assert plan.entities_to_restart == running_entities
    assert plan.only_support_service_changes is False


def test_config_update_plan_does_not_restart_for_request_time_prompt_change() -> None:
    """Request-time prompt overrides should not tear down running entities."""
    old_config = _runtime_bound_config(
        Config(
            agents={"general": AgentConfig(display_name="General Agent")},
            teams={
                "team1": TeamConfig(
                    display_name="Team 1",
                    role="Collaborate",
                    agents=["general"],
                ),
            },
            router=RouterConfig(model="default"),
        ),
    )
    new_config = _runtime_bound_config(
        Config(
            agents={"general": AgentConfig(display_name="General Agent")},
            teams={
                "team1": TeamConfig(
                    display_name="Team 1",
                    role="Collaborate",
                    agents=["general"],
                ),
            },
            router=RouterConfig(model="default"),
            prompts={"ROUTER_AGENT_SELECTION_PROMPT_TEMPLATE": "Pick one agent: {agents_info}\n{message}"},
        ),
    )

    running_entities = {ROUTER_AGENT_NAME, "general", "team1"}
    plan = build_config_update_plan(
        current_config=old_config,
        new_config=new_config,
        configured_entities=running_entities,
        existing_entities=running_entities,
        agent_bots={entity: AsyncMock() for entity in running_entities},
    )

    assert plan.entities_to_restart == set()
    assert plan.only_support_service_changes is True


def test_config_update_plan_reconciles_room_metadata_without_restarting_bots() -> None:
    """Room display-name edits should reconcile Matrix room state without bot restarts."""
    old_config = _runtime_bound_config(
        Config(
            agents={"general": AgentConfig(display_name="General Agent", rooms=["lobby"])},
            rooms={"lobby": RoomConfig(display_name="Lobby")},
            router=RouterConfig(model="default"),
        ),
    )
    new_config = _runtime_bound_config(
        Config(
            agents={"general": AgentConfig(display_name="General Agent", rooms=["lobby"])},
            rooms={"lobby": RoomConfig(display_name="Project Lobby")},
            router=RouterConfig(model="default"),
        ),
    )

    running_entities = {ROUTER_AGENT_NAME, "general"}
    plan = build_config_update_plan(
        current_config=old_config,
        new_config=new_config,
        configured_entities=running_entities,
        existing_entities=running_entities,
        agent_bots={entity: AsyncMock() for entity in running_entities},
    )

    assert plan.entities_to_restart == set()
    assert plan.room_metadata_changed is True
    assert plan.only_support_service_changes is False


def test_config_update_plan_restarts_agents_when_tool_output_threshold_changes() -> None:
    """The tool output auto-save threshold is captured when agent and team toolkits are built."""
    old_config = _runtime_bound_config(
        Config(
            agents={"general": AgentConfig(display_name="General Agent")},
            teams={
                "team1": TeamConfig(
                    display_name="Team 1",
                    role="Collaborate",
                    agents=["general"],
                ),
            },
            router=RouterConfig(model="default"),
        ),
    )
    new_config = _runtime_bound_config(
        Config(
            agents={"general": AgentConfig(display_name="General Agent")},
            teams={
                "team1": TeamConfig(
                    display_name="Team 1",
                    role="Collaborate",
                    agents=["general"],
                ),
            },
            router=RouterConfig(model="default"),
            defaults={"tool_output_auto_save_threshold_bytes": 64 * 1024},
        ),
    )

    running_entities = {ROUTER_AGENT_NAME, "general", "team1"}
    plan = build_config_update_plan(
        current_config=old_config,
        new_config=new_config,
        configured_entities=running_entities,
        existing_entities=running_entities,
        agent_bots={entity: AsyncMock() for entity in running_entities},
    )

    assert plan.entities_to_restart == {"general", "team1"}
    assert plan.only_support_service_changes is False


def test_config_update_plan_restarts_old_and_new_call_agents_when_calls_change() -> None:
    """A changed call model or agent selection rebuilds every affected call manager."""
    old_config = _runtime_bound_config(
        Config(
            agents={
                "general": AgentConfig(display_name="General Agent"),
                "writer": AgentConfig(display_name="Writer Agent"),
            },
            calls=_calls_for("general", voice="marin"),
            router=RouterConfig(model="default"),
        ),
    )
    new_config = _runtime_bound_config(
        Config(
            agents={
                "general": AgentConfig(display_name="General Agent"),
                "writer": AgentConfig(display_name="Writer Agent"),
            },
            calls=_calls_for("writer", voice="cedar"),
            router=RouterConfig(model="default"),
        ),
    )
    running_entities = {ROUTER_AGENT_NAME, "general", "writer"}

    plan = build_config_update_plan(
        current_config=old_config,
        new_config=new_config,
        configured_entities=running_entities,
        existing_entities=running_entities,
        agent_bots={entity: AsyncMock() for entity in running_entities},
    )

    assert plan.entities_to_restart == {"general", "writer"}


def test_config_update_plan_restarts_call_agents_when_profile_changes() -> None:
    """A call profile change rebuilds managers assigned to it."""
    old_config = _runtime_bound_config(
        Config(
            agents={"general": AgentConfig(display_name="General Agent")},
            calls=_calls_for("general", model="gpt-realtime-old"),
            router=RouterConfig(model="default"),
        ),
    )
    new_config = _runtime_bound_config(
        Config(
            agents={"general": AgentConfig(display_name="General Agent")},
            calls=_calls_for("general", model="gpt-realtime-new"),
            router=RouterConfig(model="default"),
        ),
    )
    running_entities = {ROUTER_AGENT_NAME, "general"}

    plan = build_config_update_plan(
        current_config=old_config,
        new_config=new_config,
        configured_entities=running_entities,
        existing_entities=running_entities,
        agent_bots={entity: AsyncMock() for entity in running_entities},
    )

    assert plan.entities_to_restart == {"general"}


def test_config_update_plan_restarts_call_agent_when_profile_voice_changes() -> None:
    """Changing a profile voice rebuilds its assigned call manager."""
    common = {
        "agents": {"general": AgentConfig(display_name="General Agent")},
        "router": RouterConfig(model="default"),
    }
    old_config = _runtime_bound_config(
        Config(
            **common,
            calls=_calls_for("general", voice="marin"),
        ),
    )
    new_config = _runtime_bound_config(
        Config(
            **common,
            calls=_calls_for("general", voice="cedar"),
        ),
    )

    plan = build_config_update_plan(
        current_config=old_config,
        new_config=new_config,
        configured_entities={ROUTER_AGENT_NAME, "general"},
        existing_entities={ROUTER_AGENT_NAME, "general"},
        agent_bots={ROUTER_AGENT_NAME: AsyncMock(), "general": AsyncMock()},
    )

    assert plan.entities_to_restart == {"general"}


def test_config_update_plan_restarts_call_agents_when_authorization_changes() -> None:
    """Call managers restart so admission uses the current authorization rules."""
    old_config = _runtime_bound_config(
        Config(
            agents={"general": AgentConfig(display_name="General Agent")},
            calls=_calls_for("general"),
            authorization={"global_users": ["@allowed:example.org"]},
            router=RouterConfig(model="default"),
        ),
    )
    new_config = _runtime_bound_config(
        Config(
            agents={"general": AgentConfig(display_name="General Agent")},
            calls=_calls_for("general"),
            authorization={"global_users": ["@replacement:example.org"]},
            router=RouterConfig(model="default"),
        ),
    )
    running_entities = {ROUTER_AGENT_NAME, "general"}

    plan = build_config_update_plan(
        current_config=old_config,
        new_config=new_config,
        configured_entities=running_entities,
        existing_entities=running_entities,
        agent_bots={entity: AsyncMock() for entity in running_entities},
    )

    assert plan.entities_to_restart == {"general"}


def test_config_update_plan_restarts_call_agents_for_captured_policy_changes() -> None:
    """Call tool closures are rebuilt when any captured config policy changes."""
    old_config = _runtime_bound_config(
        Config(
            agents={"general": AgentConfig(display_name="General Agent")},
            calls=_calls_for("general"),
            router=RouterConfig(model="default"),
        ),
    )
    new_config = _runtime_bound_config(
        Config(
            agents={"general": AgentConfig(display_name="General Agent")},
            calls=_calls_for("general"),
            tool_approval={"rules": [{"match": "read_file", "action": "require_approval"}]},
            router=RouterConfig(model="default"),
        ),
    )
    running_entities = {ROUTER_AGENT_NAME, "general"}

    plan = build_config_update_plan(
        current_config=old_config,
        new_config=new_config,
        configured_entities=running_entities,
        existing_entities=running_entities,
        agent_bots={entity: AsyncMock() for entity in running_entities},
    )

    assert plan.entities_to_restart == {"general"}


def test_config_update_plan_stops_call_agents_when_calls_are_disabled() -> None:
    """Disabling calls restarts former call agents so their managers shut down."""
    old_config = _runtime_bound_config(
        Config(
            agents={"general": AgentConfig(display_name="General Agent")},
            calls=_calls_for("general"),
            router=RouterConfig(model="default"),
        ),
    )
    new_config = _runtime_bound_config(
        Config(
            agents={"general": AgentConfig(display_name="General Agent")},
            calls=_calls_for("general", enabled=False),
            router=RouterConfig(model="default"),
        ),
    )
    running_entities = {ROUTER_AGENT_NAME, "general"}

    plan = build_config_update_plan(
        current_config=old_config,
        new_config=new_config,
        configured_entities=running_entities,
        existing_entities=running_entities,
        agent_bots={entity: AsyncMock() for entity in running_entities},
    )

    assert plan.entities_to_restart == {"general"}


def test_config_update_plan_tracks_added_entities_even_when_they_restart() -> None:
    """Added entities stay visible to account-preparation gates even when restart planning creates them."""
    old_config = _runtime_bound_config(
        Config(
            agents={"general": AgentConfig(display_name="General Agent")},
            router=RouterConfig(model="default"),
        ),
    )
    new_config = _runtime_bound_config(
        Config(
            agents={
                "general": AgentConfig(display_name="General Agent"),
                "writer": AgentConfig(display_name="Writer Agent"),
            },
            router=RouterConfig(model="default"),
        ),
    )

    plan = build_config_update_plan(
        current_config=old_config,
        new_config=new_config,
        configured_entities={ROUTER_AGENT_NAME, "general", "writer"},
        existing_entities={ROUTER_AGENT_NAME, "general"},
        agent_bots={ROUTER_AGENT_NAME: AsyncMock(), "general": AsyncMock()},
    )

    assert plan.added_entities == {"writer"}
    assert plan.new_entities == set()
    assert plan.entities_to_restart == {"writer"}


@pytest.fixture
def initial_config() -> Config:
    """Initial configuration with some agents and rooms."""
    return _runtime_bound_config(
        Config(
            agents={
                "agent1": AgentConfig(
                    display_name="Agent 1",
                    role="Test agent",
                    rooms=["room1", "room2"],
                ),
                "agent2": AgentConfig(
                    display_name="Agent 2",
                    role="Another test agent",
                    rooms=["room1"],
                ),
            },
            teams={
                "team1": TeamConfig(
                    display_name="Team 1",
                    role="Test team",
                    agents=["agent1", "agent2"],
                    rooms=["room3"],
                ),
            },
            models={
                "default": ModelConfig(
                    provider="ollama",
                    id="llama3.2",
                    host="http://localhost:11434",
                ),
            },
        ),
    )


@pytest.fixture
def updated_config() -> Config:
    """Updated configuration with changed room assignments."""
    return _runtime_bound_config(
        Config(
            agents={
                "agent1": AgentConfig(
                    display_name="Agent 1",
                    role="Test agent",
                    rooms=["room1", "room4"],  # Changed: removed room2, added room4
                ),
                "agent2": AgentConfig(
                    display_name="Agent 2",
                    role="Another test agent",
                    rooms=["room2", "room3"],  # Changed: removed room1, added room2 and room3
                ),
                "agent3": AgentConfig(  # New agent
                    display_name="Agent 3",
                    role="New agent",
                    rooms=["room5"],
                ),
            },
            teams={
                "team1": TeamConfig(
                    display_name="Team 1",
                    role="Test team",
                    agents=["agent1", "agent2", "agent3"],  # Added agent3
                    rooms=["room3", "room6"],  # Added room6
                ),
            },
            models={
                "default": ModelConfig(
                    provider="ollama",
                    id="llama3.2",
                    host="http://localhost:11434",
                ),
            },
        ),
    )


@pytest.fixture
def mock_agent_users() -> dict[str, AgentMatrixUser]:
    """Create mock agent users."""
    return {
        ROUTER_AGENT_NAME: AgentMatrixUser(
            agent_name=ROUTER_AGENT_NAME,
            user_id=f"@mindroom_{ROUTER_AGENT_NAME}:localhost",
            display_name="RouterAgent",
            password=TEST_PASSWORD,
        ),
        "agent1": AgentMatrixUser(
            agent_name="agent1",
            user_id="@mindroom_agent1:localhost",
            display_name="Agent 1",
            password=TEST_PASSWORD,
        ),
        "agent2": AgentMatrixUser(
            agent_name="agent2",
            user_id="@mindroom_agent2:localhost",
            display_name="Agent 2",
            password=TEST_PASSWORD,
        ),
        "agent3": AgentMatrixUser(
            agent_name="agent3",
            user_id="@mindroom_agent3:localhost",
            display_name="Agent 3",
            password=TEST_PASSWORD,
        ),
        "team1": AgentMatrixUser(
            agent_name="team1",
            user_id="@mindroom_team1:localhost",
            display_name="Team 1",
            password=TEST_PASSWORD,
        ),
    }


@pytest.mark.asyncio
async def test_agent_joins_new_rooms_on_config_reload(  # noqa: C901
    initial_config: Config,  # noqa: ARG001
    updated_config: Config,  # noqa: ARG001
    mock_agent_users: dict[str, AgentMatrixUser],
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Test that agents join new rooms when their configuration is updated."""
    # Track room operations
    joined_rooms: dict[str, list[str]] = {}
    left_rooms: dict[str, list[str]] = {}

    async def mock_join_room(client: AsyncMock, room_id: str) -> RoomJoinOutcome:
        user_id = client.user_id
        if user_id not in joined_rooms:
            joined_rooms[user_id] = []
        joined_rooms[user_id].append(room_id)
        return RoomJoinOutcome.JOINED

    async def mock_leave_room(client: AsyncMock, room_id: str) -> bool:
        user_id = client.user_id
        if user_id not in left_rooms:
            left_rooms[user_id] = []
        left_rooms[user_id].append(room_id)
        return True

    monkeypatch.setattr("mindroom.bot_room_lifecycle.join_room", mock_join_room)
    monkeypatch.setattr("mindroom.matrix.rooms.leave_room", mock_leave_room)

    # Mock restore_scheduled_tasks
    async def mock_restore_scheduled_tasks(
        _client: AsyncMock,
        _room_id: str,
        _config: Config,
        _runtime_paths: object,
        _event_cache: object,
        _conversation_cache: object,
    ) -> int:
        return 0

    monkeypatch.setattr("mindroom.bot.restore_scheduled_tasks", mock_restore_scheduled_tasks)

    # Mock resolve_room_aliases
    def mock_resolve_room_aliases(aliases: list[str], _runtime_paths: object | None = None) -> list[str]:
        return list(aliases)

    monkeypatch.setattr("mindroom.bot.resolve_room_aliases", mock_resolve_room_aliases)

    # Mock get_joined_rooms to simulate current room membership
    async def mock_get_joined_rooms(client: AsyncMock) -> list[str]:
        user_id = client.user_id
        if "agent1" in user_id:
            return ["room1", "room2"]  # agent1 is currently in room1 and room2
        if "agent2" in user_id:
            return ["room1"]  # agent2 is currently in room1
        if "team1" in user_id:
            return ["room3"]  # team1 is currently in room3
        if ROUTER_AGENT_NAME in user_id:
            return ["room1", "room2", "room3"]  # router is in all initial rooms
        return []

    monkeypatch.setattr("mindroom.bot_room_lifecycle.get_joined_rooms", mock_get_joined_rooms)

    # Create agent1 bot with initial config
    config = _runtime_bound_config(Config(router=RouterConfig(model="default")), tmp_path)
    agent1_bot = AgentBot(
        agent_user=mock_agent_users["agent1"],
        storage_path=tmp_path,
        config=config,
        runtime_paths=runtime_paths_for(config),
        rooms=["room1", "room2"],  # Initial rooms
    )
    mock_client = AsyncMock()
    mock_client.user_id = "@mindroom_agent1:localhost"
    setup_test_bot(agent1_bot, mock_client)

    # Update to new config rooms
    agent1_bot.rooms = ["room1", "room4"]  # New rooms: removed room2, added room4

    # Apply room updates
    await agent1_bot.join_configured_rooms()
    await agent1_bot.leave_unconfigured_rooms()

    # Verify agent1 joined room4 (new room)
    assert "room4" in joined_rooms.get("@mindroom_agent1:localhost", [])
    # Verify agent1 left room2 (no longer configured)
    assert "room2" in left_rooms.get("@mindroom_agent1:localhost", [])


@pytest.mark.asyncio
async def test_router_updates_rooms_on_config_reload(
    initial_config: Config,
    updated_config: Config,
    mock_agent_users: dict[str, AgentMatrixUser],
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Test that the router updates its room list when agents/teams change their rooms."""
    # Track room operations
    joined_rooms: list[str] = []
    left_rooms: list[str] = []

    async def mock_join_room(_client: AsyncMock, room_id: str) -> RoomJoinOutcome:
        joined_rooms.append(room_id)
        return RoomJoinOutcome.JOINED

    async def mock_leave_room(_client: AsyncMock, room_id: str) -> bool:
        left_rooms.append(room_id)
        return True

    monkeypatch.setattr("mindroom.bot_room_lifecycle.join_room", mock_join_room)
    monkeypatch.setattr("mindroom.matrix.rooms.leave_room", mock_leave_room)

    # Mock restore_scheduled_tasks
    async def mock_restore_scheduled_tasks(
        _client: AsyncMock,
        _room_id: str,
        _config: Config,
        _runtime_paths: object,
        _event_cache: object,
        _conversation_cache: object,
    ) -> int:
        return 0

    monkeypatch.setattr("mindroom.bot.restore_scheduled_tasks", mock_restore_scheduled_tasks)

    # Mock resolve_room_aliases
    def mock_resolve_room_aliases(aliases: list[str], _runtime_paths: object | None = None) -> list[str]:
        return list(aliases)

    monkeypatch.setattr("mindroom.bot.resolve_room_aliases", mock_resolve_room_aliases)

    # Mock get_joined_rooms to simulate current room membership
    async def mock_get_joined_rooms(_client: AsyncMock) -> list[str]:
        # Router is currently in initial config rooms
        return ["room1", "room2", "room3"]

    monkeypatch.setattr("mindroom.bot_room_lifecycle.get_joined_rooms", mock_get_joined_rooms)

    # Get initial router rooms
    initial_router_rooms = initial_config.get_all_configured_rooms()
    assert initial_router_rooms == {"room1", "room2", "room3"}

    # Get updated router rooms
    updated_router_rooms = updated_config.get_all_configured_rooms()
    assert updated_router_rooms == {"room1", "room2", "room3", "room4", "room5", "room6"}

    # Create router bot with updated config
    config = _runtime_bound_config(Config(router=RouterConfig(model="default")), tmp_path)
    router_bot = AgentBot(
        agent_user=mock_agent_users[ROUTER_AGENT_NAME],
        storage_path=tmp_path,
        config=config,
        runtime_paths=runtime_paths_for(config),
        rooms=list(updated_router_rooms),
    )
    mock_client = AsyncMock()
    mock_client.user_id = f"@mindroom_{ROUTER_AGENT_NAME}:localhost"
    setup_test_bot(router_bot, mock_client)

    # Apply room updates
    await router_bot.join_configured_rooms()
    await router_bot.leave_unconfigured_rooms()

    # Verify router joined new rooms
    for new_room in ["room4", "room5", "room6"]:
        assert new_room in joined_rooms

    # Router should not leave any rooms (all initial rooms still have agents)
    assert len(left_rooms) == 0


@pytest.mark.asyncio
async def test_new_agent_joins_rooms_on_config_reload(
    initial_config: Config,  # noqa: ARG001
    updated_config: Config,  # noqa: ARG001
    mock_agent_users: dict[str, AgentMatrixUser],
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Test that new agents are created and join their configured rooms."""
    # Track room operations
    joined_rooms: dict[str, list[str]] = {}

    async def mock_join_room(client: AsyncMock, room_id: str) -> RoomJoinOutcome:
        user_id = client.user_id
        if user_id not in joined_rooms:
            joined_rooms[user_id] = []
        joined_rooms[user_id].append(room_id)
        return RoomJoinOutcome.JOINED

    monkeypatch.setattr("mindroom.bot_room_lifecycle.join_room", mock_join_room)

    # Mock restore_scheduled_tasks
    async def mock_restore_scheduled_tasks(
        _client: AsyncMock,
        _room_id: str,
        _config: Config,
        _runtime_paths: object,
        _event_cache: object,
        _conversation_cache: object,
    ) -> int:
        return 0

    monkeypatch.setattr("mindroom.bot.restore_scheduled_tasks", mock_restore_scheduled_tasks)

    # Mock resolve_room_aliases
    def mock_resolve_room_aliases(aliases: list[str]) -> list[str]:
        return list(aliases)

    monkeypatch.setattr("mindroom.bot.resolve_room_aliases", mock_resolve_room_aliases)

    # Mock get_joined_rooms
    async def mock_get_joined_rooms(_client: AsyncMock) -> list[str]:
        return []  # New agent has no rooms initially

    monkeypatch.setattr("mindroom.bot_room_lifecycle.get_joined_rooms", mock_get_joined_rooms)

    # Create agent3 bot (new agent in updated config)
    config = _runtime_bound_config(Config(router=RouterConfig(model="default")), tmp_path)
    agent3_bot = AgentBot(
        agent_user=mock_agent_users["agent3"],
        storage_path=tmp_path,
        config=config,
        runtime_paths=runtime_paths_for(config),
        rooms=["room5"],
    )
    mock_client = AsyncMock()
    mock_client.user_id = "@mindroom_agent3:localhost"
    setup_test_bot(agent3_bot, mock_client)

    # Apply room updates for new agent
    await agent3_bot.join_configured_rooms()

    # Verify agent3 joined its configured room
    assert "room5" in joined_rooms.get("@mindroom_agent3:localhost", [])


@pytest.mark.asyncio
async def test_team_room_changes_on_config_reload(
    initial_config: Config,  # noqa: ARG001
    updated_config: Config,  # noqa: ARG001
    mock_agent_users: dict[str, AgentMatrixUser],
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Test that teams update their room memberships when configuration changes."""
    # Track room operations
    joined_rooms: dict[str, list[str]] = {}
    left_rooms: dict[str, list[str]] = {}

    async def mock_join_room(client: AsyncMock, room_id: str) -> RoomJoinOutcome:
        user_id = client.user_id
        if user_id not in joined_rooms:
            joined_rooms[user_id] = []
        joined_rooms[user_id].append(room_id)
        return RoomJoinOutcome.JOINED

    async def mock_leave_room(client: AsyncMock, room_id: str) -> bool:
        user_id = client.user_id
        if user_id not in left_rooms:
            left_rooms[user_id] = []
        left_rooms[user_id].append(room_id)
        return True

    monkeypatch.setattr("mindroom.bot_room_lifecycle.join_room", mock_join_room)
    monkeypatch.setattr("mindroom.matrix.rooms.leave_room", mock_leave_room)

    # Mock restore_scheduled_tasks
    async def mock_restore_scheduled_tasks(
        _client: AsyncMock,
        _room_id: str,
        _config: Config,
        _runtime_paths: object,
        _event_cache: object,
        _conversation_cache: object,
    ) -> int:
        return 0

    monkeypatch.setattr("mindroom.bot.restore_scheduled_tasks", mock_restore_scheduled_tasks)

    # Mock resolve_room_aliases
    def mock_resolve_room_aliases(aliases: list[str]) -> list[str]:
        return list(aliases)

    monkeypatch.setattr("mindroom.bot.resolve_room_aliases", mock_resolve_room_aliases)

    # Mock get_joined_rooms to simulate current room membership
    async def mock_get_joined_rooms(client: AsyncMock) -> list[str]:
        user_id = client.user_id
        if "team1" in user_id:
            return ["room3"]  # team1 is currently only in room3
        return []

    monkeypatch.setattr("mindroom.bot_room_lifecycle.get_joined_rooms", mock_get_joined_rooms)

    # Create team1 bot with updated config
    config = _runtime_bound_config(Config(router=RouterConfig(model="default")), tmp_path)
    team1_bot = AgentBot(
        agent_user=mock_agent_users["team1"],
        storage_path=tmp_path,
        config=config,
        runtime_paths=runtime_paths_for(config),
        rooms=["room3", "room6"],
    )
    mock_client = AsyncMock()
    mock_client.user_id = "@mindroom_team1:localhost"
    setup_test_bot(team1_bot, mock_client)

    # Apply room updates
    await team1_bot.join_configured_rooms()
    await team1_bot.leave_unconfigured_rooms()

    # Verify team1 joined room6 (new room)
    assert "room6" in joined_rooms.get("@mindroom_team1:localhost", [])
    # Team1 should not leave room3 (still configured)
    assert "room3" not in left_rooms.get("@mindroom_team1:localhost", [])


@pytest.mark.asyncio
@pytest.mark.requires_matrix  # This test requires a real Matrix server or extensive mocking
@pytest.mark.timeout(10)  # Add timeout to prevent hanging on real server connection
async def test_orchestrator_handles_config_reload(  # noqa: PLR0915
    initial_config: Config,
    updated_config: Config,
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Test that the orchestrator properly handles config reloads and updates all bots."""
    # Track config loads
    config_loads = [initial_config, updated_config]
    load_count = [0]

    def mock_load_config(
        _runtime_paths: object | None = None,
        **_kwargs: object,
    ) -> Config:
        result = config_loads[min(load_count[0], len(config_loads) - 1)]
        load_count[0] += 1
        return result

    monkeypatch.setattr("mindroom.orchestrator.load_config", mock_load_config)
    monkeypatch.setattr("mindroom.orchestration.config_lifecycle.load_config", mock_load_config)

    def mock_resolve_room_aliases(aliases: list[str], _runtime_paths: object | None = None) -> list[str]:
        return list(aliases)

    monkeypatch.setattr("mindroom.bot.resolve_room_aliases", mock_resolve_room_aliases)

    # Mock topic generation to avoid calling AI
    async def mock_generate_room_topic_ai(room_key: str, room_name: str, config: Config) -> str:  # noqa: ARG001
        return f"Test topic for {room_name}"

    monkeypatch.setattr("mindroom.topic_generator.generate_room_topic_ai", mock_generate_room_topic_ai)
    monkeypatch.setattr("mindroom.matrix.rooms.generate_room_topic_ai", mock_generate_room_topic_ai)

    # Create orchestrator
    # Mock start/sync at class level so newly created bots during update_config don't perform real login/sync
    # But we need to ensure client gets set when start() is called
    async def mock_start(self: AgentBot) -> None:
        """Mock start that sets a mock client."""
        self.client = AsyncMock()
        self.client.user_id = self.agent_user.user_id
        self.running = True

    monkeypatch.setattr("mindroom.bot.AgentBot.start", mock_start)
    monkeypatch.setattr("mindroom.bot.AgentBot.sync_forever", AsyncMock())
    monkeypatch.setattr("mindroom.bot.TeamBot.start", mock_start)
    monkeypatch.setattr("mindroom.bot.TeamBot.sync_forever", AsyncMock())
    monkeypatch.setattr("mindroom.orchestrator._MultiAgentOrchestrator._ensure_user_account", AsyncMock())
    monkeypatch.setattr("mindroom.orchestrator._MultiAgentOrchestrator._setup_rooms_and_memberships", AsyncMock())

    orchestrator = _MultiAgentOrchestrator(runtime_paths=orchestrator_runtime_paths(tmp_path))

    # Initialize with initial config
    await orchestrator.initialize()

    # Verify initial state
    assert "agent1" in orchestrator.agent_bots
    assert "agent2" in orchestrator.agent_bots
    assert "agent3" not in orchestrator.agent_bots  # Not in initial config
    assert "team1" in orchestrator.agent_bots
    assert ROUTER_AGENT_NAME in orchestrator.agent_bots

    # Check initial room assignments
    assert set(orchestrator.agent_bots["agent1"].rooms) == {"room1", "room2"}
    assert set(orchestrator.agent_bots["agent2"].rooms) == {"room1"}
    assert set(orchestrator.agent_bots["team1"].rooms) == {"room3"}
    assert set(orchestrator.agent_bots[ROUTER_AGENT_NAME].rooms) == {"room1", "room2", "room3"}

    # Create a mock start method that initializes client
    async def mock_start_with_thread_manager(self: AgentBot) -> None:
        """Mock start that initializes client."""
        if not hasattr(self, "client") or self.client is None:
            self.client = AsyncMock()
            self.client.user_id = self.agent_user.user_id

    # Patch AgentBot.start and TeamBot.start to use our mock
    monkeypatch.setattr("mindroom.bot.AgentBot.start", mock_start_with_thread_manager)
    monkeypatch.setattr("mindroom.bot.TeamBot.start", mock_start_with_thread_manager)

    # Mock bot operations for update
    for bot in orchestrator.agent_bots.values():
        monkeypatch.setattr(bot, "stop", AsyncMock())
        monkeypatch.setattr(bot, "start", mock_start_with_thread_manager)
        monkeypatch.setattr(bot, "ensure_user_account", AsyncMock())
        monkeypatch.setattr(bot, "sync_forever", AsyncMock(side_effect=asyncio.CancelledError()))

    # Update config
    updated = await orchestrator.config_reload._update_config()
    assert updated  # Should return True since config changed

    # Verify updated state
    assert "agent1" in orchestrator.agent_bots
    assert "agent2" in orchestrator.agent_bots
    assert "agent3" in orchestrator.agent_bots  # New agent added
    assert "team1" in orchestrator.agent_bots
    assert ROUTER_AGENT_NAME in orchestrator.agent_bots

    # Check updated room assignments
    assert set(orchestrator.agent_bots["agent1"].rooms) == {"room1", "room4"}
    assert set(orchestrator.agent_bots["agent2"].rooms) == {"room2", "room3"}
    assert set(orchestrator.agent_bots["agent3"].rooms) == {"room5"}
    assert set(orchestrator.agent_bots["team1"].rooms) == {"room3", "room6"}
    assert set(orchestrator.agent_bots[ROUTER_AGENT_NAME].rooms) == {
        "room1",
        "room2",
        "room3",
        "room4",
        "room5",
        "room6",
    }


@pytest.mark.asyncio
async def test_room_membership_state_after_config_update(  # noqa: C901, PLR0915
    initial_config: Config,  # noqa: ARG001
    updated_config: Config,  # noqa: ARG001
    mock_agent_users: dict[str, AgentMatrixUser],
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Test that room membership state is correct after config updates."""
    # Simulate room membership state
    room_memberships = {
        "room1": [
            "@mindroom_agent1:localhost",
            "@mindroom_agent2:localhost",
            f"@mindroom_{ROUTER_AGENT_NAME}:localhost",
        ],
        "room2": ["@mindroom_agent1:localhost", f"@mindroom_{ROUTER_AGENT_NAME}:localhost"],
        "room3": ["@mindroom_team1:localhost", f"@mindroom_{ROUTER_AGENT_NAME}:localhost"],
    }

    def update_room_membership(user_id: str, room_id: str, action: str) -> None:
        """Update simulated room membership."""
        if action == "join":
            if room_id not in room_memberships:
                room_memberships[room_id] = []
            if user_id not in room_memberships[room_id]:
                room_memberships[room_id].append(user_id)
        elif action == "leave":
            if room_id in room_memberships and user_id in room_memberships[room_id]:
                room_memberships[room_id].remove(user_id)

    async def mock_join_room(client: AsyncMock, room_id: str) -> RoomJoinOutcome:
        update_room_membership(client.user_id, room_id, "join")
        return RoomJoinOutcome.JOINED

    async def mock_leave_room(client: AsyncMock, room_id: str) -> bool:
        update_room_membership(client.user_id, room_id, "leave")
        return True

    monkeypatch.setattr("mindroom.bot_room_lifecycle.join_room", mock_join_room)
    monkeypatch.setattr("mindroom.matrix.rooms.leave_room", mock_leave_room)

    # Mock restore_scheduled_tasks
    async def mock_restore_scheduled_tasks(
        _client: AsyncMock,
        _room_id: str,
        _config: Config,
        _runtime_paths: object,
        _event_cache: object,
        _conversation_cache: object,
    ) -> int:
        return 0

    monkeypatch.setattr("mindroom.bot.restore_scheduled_tasks", mock_restore_scheduled_tasks)

    # Mock resolve_room_aliases
    def mock_resolve_room_aliases(aliases: list[str]) -> list[str]:
        return list(aliases)

    monkeypatch.setattr("mindroom.bot.resolve_room_aliases", mock_resolve_room_aliases)

    # Mock get_joined_rooms based on room_memberships
    async def mock_get_joined_rooms(client: AsyncMock) -> list[str]:
        user_id = client.user_id
        rooms = []
        for room_id, members in room_memberships.items():
            if user_id in members:
                rooms.append(room_id)
        return rooms

    monkeypatch.setattr("mindroom.bot_room_lifecycle.get_joined_rooms", mock_get_joined_rooms)

    # Apply config updates for each bot
    bots_config = {
        "@mindroom_agent1:localhost": {"old": ["room1", "room2"], "new": ["room1", "room4"]},
        "@mindroom_agent2:localhost": {"old": ["room1"], "new": ["room2", "room3"]},
        "@mindroom_agent3:localhost": {"old": [], "new": ["room5"]},
        "@mindroom_team1:localhost": {"old": ["room3"], "new": ["room3", "room6"]},
        f"@mindroom_{ROUTER_AGENT_NAME}:localhost": {
            "old": ["room1", "room2", "room3"],
            "new": ["room1", "room2", "room3", "room4", "room5", "room6"],
        },
    }

    # Simulate config update for each bot
    for user_id, bot_config in bots_config.items():
        mock_client = AsyncMock()
        mock_client.user_id = user_id

        # Determine which agent this is
        if "agent1" in user_id:
            agent_user = mock_agent_users["agent1"]
        elif "agent2" in user_id:
            agent_user = mock_agent_users["agent2"]
        elif "agent3" in user_id:
            agent_user = mock_agent_users["agent3"]
        elif "team1" in user_id:
            agent_user = mock_agent_users["team1"]
        else:
            agent_user = mock_agent_users[ROUTER_AGENT_NAME]

        config = _runtime_bound_config(Config(router=RouterConfig(model="default")), tmp_path)

        bot = AgentBot(
            agent_user=agent_user,
            storage_path=tmp_path,
            config=config,
            runtime_paths=runtime_paths_for(config),
            rooms=bot_config["new"],
        )
        setup_test_bot(bot, mock_client)

        await bot.join_configured_rooms()
        await bot.leave_unconfigured_rooms()

    # Verify final room membership state
    assert set(room_memberships.get("room1", [])) == {
        "@mindroom_agent1:localhost",
        f"@mindroom_{ROUTER_AGENT_NAME}:localhost",
    }
    assert set(room_memberships.get("room2", [])) == {
        "@mindroom_agent2:localhost",
        f"@mindroom_{ROUTER_AGENT_NAME}:localhost",
    }
    assert set(room_memberships.get("room3", [])) == {
        "@mindroom_agent2:localhost",
        "@mindroom_team1:localhost",
        f"@mindroom_{ROUTER_AGENT_NAME}:localhost",
    }
    assert set(room_memberships.get("room4", [])) == {
        "@mindroom_agent1:localhost",
        f"@mindroom_{ROUTER_AGENT_NAME}:localhost",
    }
    assert set(room_memberships.get("room5", [])) == {
        "@mindroom_agent3:localhost",
        f"@mindroom_{ROUTER_AGENT_NAME}:localhost",
    }
    assert set(room_memberships.get("room6", [])) == {
        "@mindroom_team1:localhost",
        f"@mindroom_{ROUTER_AGENT_NAME}:localhost",
    }


@pytest.mark.asyncio
async def test_in_flight_response_count_nonzero_during_send_response(
    tmp_path: Path,
    mock_agent_users: dict[str, AgentMatrixUser],
) -> None:
    """in_flight_response_count must be >0 even while the visible send is still awaiting."""
    config = _runtime_bound_config(
        Config(
            agents={"agent1": AgentConfig(display_name="Agent 1")},
            router=RouterConfig(model="default"),
        ),
        tmp_path,
    )
    bot = AgentBot(
        agent_user=mock_agent_users["agent1"],
        storage_path=tmp_path,
        config=config,
        runtime_paths=runtime_paths_for(config),
    )
    setup_test_bot(bot, AsyncMock())

    send_entered = asyncio.Event()
    release_send = asyncio.Event()

    async def slow_send(*_args: object, **_kwargs: object) -> str:
        send_entered.set()
        await release_send.wait()
        return "$msg"

    send_response = AsyncMock(side_effect=slow_send)
    install_send_response_mock(bot, send_response)

    async def response_function(message_id: str | None) -> None:
        pass

    runner = unwrap_extracted_collaborator(bot._response_runner)
    request = ResponseRequest(
        thread_history=(),
        prompt="Hello",
        response_envelope=request_envelope(
            room_id="!room:localhost",
            reply_to_event_id="$reply",
            agent_name=bot.agent_name,
        ),
    )

    async def run_response(
        target: MessageTarget,
        _early_placeholder: object,
    ) -> str | None:
        return await runner._run_cancellable_response(
            target=target,
            response_function=response_function,
            thinking_message="Thinking...",
        )

    task = asyncio.create_task(
        runner._run_locked_response_lifecycle(
            request,
            response_kind="ai",
            locked_operation=run_response,
        ),
    )

    try:
        await asyncio.wait_for(send_entered.wait(), timeout=1)
        assert bot.in_flight_response_count == 1
    finally:
        release_send.set()
        await asyncio.gather(task, return_exceptions=True)
        for cleanup_task in bot.stop_manager.cleanup_tasks:
            cleanup_task.cancel()
        await asyncio.gather(*bot.stop_manager.cleanup_tasks, return_exceptions=True)


@pytest.mark.asyncio
async def test_in_flight_response_count_stays_per_entity_across_bots(
    tmp_path: Path,
    mock_agent_users: dict[str, AgentMatrixUser],
) -> None:
    """One bot's active response must not make a different bot look busy.

    Every bot shares one admission gate, whose count is process-wide. Callers
    like the todo-poke idle check ask whether *this* entity is busy, so the
    per-bot count must stay distinct from the gate total.
    """
    config = _runtime_bound_config(
        Config(
            agents={"agent1": AgentConfig(display_name="Agent 1"), "agent2": AgentConfig(display_name="Agent 2")},
            router=RouterConfig(model="default"),
        ),
        tmp_path,
    )
    bots = {}
    for agent_name in ("agent1", "agent2"):
        bot = AgentBot(
            agent_user=mock_agent_users[agent_name],
            storage_path=tmp_path,
            config=config,
            runtime_paths=runtime_paths_for(config),
        )
        setup_test_bot(bot, AsyncMock())
        bots[agent_name] = bot

    shared_gate = ResponseAdmissionGate()
    for bot in bots.values():
        bot.admission_gate = shared_gate

    busy, idle = bots["agent1"], bots["agent2"]
    entered = asyncio.Event()
    release = asyncio.Event()

    async def blocked_response(_target: MessageTarget, _early_placeholder: object) -> str | None:
        entered.set()
        await release.wait()
        return None

    busy_runner = unwrap_extracted_collaborator(busy._response_runner)
    task = asyncio.create_task(
        busy_runner._run_locked_response_lifecycle(
            ResponseRequest(
                thread_history=(),
                prompt="Hello",
                response_envelope=request_envelope(
                    room_id="!room:localhost",
                    reply_to_event_id="$reply",
                    agent_name=busy.agent_name,
                ),
            ),
            response_kind="ai",
            locked_operation=blocked_response,
        ),
    )
    try:
        await asyncio.wait_for(entered.wait(), timeout=1)
        assert busy.in_flight_response_count == 1
        # The shared gate sees the response, so a reload correctly defers...
        assert shared_gate.in_flight_response_count == 1
        # ...but the unrelated bot is still idle.
        assert idle.in_flight_response_count == 0
    finally:
        release.set()
        await asyncio.gather(task, return_exceptions=True)

    assert busy.in_flight_response_count == 0
    assert shared_gate.in_flight_response_count == 0


@pytest.mark.asyncio
async def test_closed_admission_defers_response_until_gate_reopens(
    tmp_path: Path,
    mock_agent_users: dict[str, AgentMatrixUser],
) -> None:
    """A response arriving during config apply should run after the gate reopens."""
    config = _runtime_bound_config(
        Config(
            agents={"agent1": AgentConfig(display_name="Agent 1")},
            router=RouterConfig(model="default"),
        ),
        tmp_path,
    )
    bot = AgentBot(
        agent_user=mock_agent_users["agent1"],
        storage_path=tmp_path,
        config=config,
        runtime_paths=runtime_paths_for(config),
    )
    setup_test_bot(bot, AsyncMock())
    runner = unwrap_extracted_collaborator(bot._response_runner)
    admission_gate = ResponseAdmissionGate()
    bot.admission_gate = admission_gate
    response_started = asyncio.Event()

    async def run_response(_target: MessageTarget, _early_placeholder: object) -> str:
        response_started.set()
        return "$response"

    assert admission_gate.close_if_idle()
    task = asyncio.create_task(
        runner._run_locked_response_lifecycle(
            ResponseRequest(
                thread_history=(),
                prompt="Hello",
                response_envelope=request_envelope(
                    room_id="!room:localhost",
                    reply_to_event_id="$reply",
                    agent_name=bot.agent_name,
                ),
            ),
            response_kind="ai",
            locked_operation=run_response,
        ),
    )
    try:
        await asyncio.sleep(0)
        assert not task.done()
        assert not response_started.is_set()
        assert bot.in_flight_response_count == 0

        admission_gate.reopen()

        assert await asyncio.wait_for(task, timeout=1) == "$response"
        assert response_started.is_set()
        assert bot.in_flight_response_count == 0
    finally:
        task.cancel()
        await asyncio.gather(task, return_exceptions=True)


@pytest.mark.asyncio
async def test_replaced_runtime_refuses_deferred_response_without_matrix_io(
    tmp_path: Path,
    mock_agent_users: dict[str, AgentMatrixUser],
) -> None:
    """A replaced runtime should fail its waiter so checkpoint replay owns recovery."""
    config = _runtime_bound_config(
        Config(
            agents={"agent1": AgentConfig(display_name="Agent 1")},
            router=RouterConfig(model="default"),
        ),
        tmp_path,
    )
    bot = AgentBot(
        agent_user=mock_agent_users["agent1"],
        storage_path=tmp_path,
        config=config,
        runtime_paths=runtime_paths_for(config),
    )
    setup_test_bot(bot, AsyncMock())
    send_response = AsyncMock(return_value="$thinking")
    install_send_response_mock(bot, send_response)
    runner = unwrap_extracted_collaborator(bot._response_runner)
    admission_gate = ResponseAdmissionGate()
    bot.admission_gate = admission_gate
    refusal_logger = MagicMock()
    runner.deps = replace(runner.deps, logger=refusal_logger)

    # Stand where a config apply stands: admission closed, plan in progress.
    assert admission_gate.close_if_idle()
    assert bot.admission_gate is admission_gate
    task = bot._response_runner.track_inbox_response(
        bot._response_runner.generate_response(
            ResponseRequest(
                thread_history=(),
                prompt="Hello",
                response_envelope=request_envelope(
                    room_id="!room:localhost",
                    reply_to_event_id="$reply",
                    agent_name=bot.agent_name,
                ),
            ),
        ),
        name="test_reload_admission_race",
        recovery_proof_ready=lambda: False,
    )
    try:
        await asyncio.sleep(0)
        assert not task.done()
        assert bot.in_flight_response_count == 0
        send_response.assert_not_awaited()

        # Replacement shutdown wakes pre-admission waiters without cancelling
        # them, so the callback failure poisons the old sync checkpoint.
        await bot.prepare_for_sync_shutdown(shutdown_intent=SYNC_RESTART_SHUTDOWN)

        # A refusal is an admission failure, not a cancellation: handlers that
        # treat CancelledError as teardown must not mistake it for one.
        assert task.done()
        assert not task.cancelled()
        assert isinstance(task.exception(), ResponseAdmissionRefusedError)
        assert str(task.exception()) == "Runtime replacement is restarting this entity"
        assert bot.in_flight_response_count == 0
        send_response.assert_not_awaited()
        assert any(
            call.args and call.args[0] == "response_deferred_during_replacement"
            for call in refusal_logger.info.call_args_list
        )
        assert any(
            call.args and call.args[0] == "response_refused_after_runtime_replacement"
            for call in refusal_logger.warning.call_args_list
        )
    finally:
        task.cancel()
        await asyncio.gather(task, return_exceptions=True)


@pytest.mark.asyncio
async def test_run_cancellable_response_marks_thinking_placeholder_pending(
    tmp_path: Path,
    mock_agent_users: dict[str, AgentMatrixUser],
) -> None:
    """Initial thinking messages should carry pending stream metadata for restart-safe classification."""
    config = _runtime_bound_config(
        Config(
            agents={"agent1": AgentConfig(display_name="Agent 1")},
            router=RouterConfig(model="default"),
        ),
        tmp_path,
    )
    bot = AgentBot(
        agent_user=mock_agent_users["agent1"],
        storage_path=tmp_path,
        config=config,
        runtime_paths=runtime_paths_for(config),
    )
    setup_test_bot(bot, AsyncMock())

    captured_send: dict[str, object] = {}

    async def fake_send_response(
        *,
        target: object,
        response_text: str,
        skip_mentions: bool = False,
        tool_trace: list[object] | None = None,
        extra_content: dict[str, object] | None = None,
    ) -> str:
        captured_send["response_text"] = response_text
        captured_send["target"] = target
        captured_send["skip_mentions"] = skip_mentions
        captured_send["tool_trace"] = tool_trace
        captured_send["extra_content"] = extra_content
        return "$thinking"

    send_response = AsyncMock(side_effect=fake_send_response)
    install_send_response_mock(bot, send_response)

    async def response_function(message_id: str | None) -> None:
        assert message_id == "$thinking"

    await bot._response_runner._run_cancellable_response(
        target=MessageTarget.resolve("!room:localhost", None, "$reply"),
        response_function=response_function,
        thinking_message="Thinking...",
    )

    assert captured_send["response_text"] == "Thinking..."
    assert captured_send["extra_content"] == {STREAM_STATUS_KEY: STREAM_STATUS_PENDING}


@pytest.mark.asyncio
async def test_shutdown_during_active_drain_cancels_reload(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Calling stop() during an active drain must cancel the reload without applying it."""
    monkeypatch.setattr("mindroom.orchestration.config_lifecycle._CONFIG_RELOAD_DEBOUNCE_SECONDS", 0.01)
    monkeypatch.setattr("mindroom.orchestration.config_lifecycle._REPLACEMENT_DRAIN_IDLE_POLL_SECONDS", 0.01)

    orchestrator = _MultiAgentOrchestrator(runtime_paths=orchestrator_runtime_paths(tmp_path))
    orchestrator.running = True

    mock_bot = MagicMock(spec=AgentBot)
    mock_bot.stop = AsyncMock()
    orchestrator.agent_bots["agent1"] = mock_bot
    # An admitted response that never finishes, so the drain never goes idle.
    assert orchestrator.config_reload.response_admission_gate.admit()
    orchestrator.config_reload._update_config = AsyncMock(return_value=True)
    orchestrator.config_reload.request_reload()
    task = orchestrator.config_reload._reload_task
    assert task is not None

    # Let the drain loop start polling
    await asyncio.sleep(0.05)
    orchestrator.config_reload._update_config.assert_not_awaited()

    # Shutdown
    await orchestrator.stop()

    # The reload task should have been cancelled
    assert task.done()
    orchestrator.config_reload._update_config.assert_not_awaited()
