"""Multi-agent orchestration runtime."""

from __future__ import annotations

import asyncio
import signal
import time
from collections.abc import Awaitable, Callable
from contextlib import suppress
from dataclasses import dataclass, field, replace
from functools import partial
from typing import TYPE_CHECKING, NoReturn, cast, overload
from uuid import uuid4

import uvicorn

from mindroom import constants
from mindroom.agents import ensure_default_agent_workspaces
from mindroom.approval_transport import ApprovalMatrixTransport
from mindroom.authorization import is_authorized_sender
from mindroom.background_tasks import create_background_task, wait_for_background_tasks
from mindroom.constants import ROUTER_AGENT_NAME
from mindroom.embedder_health import check_embedder_health, handle_embedder_config_reload
from mindroom.entity_resolution import (
    DuplicateManagedEntityIdentityError,
    MissingManagedEntityAccountError,
    configured_bot_user_ids_for_room,
    entity_identity_registry,
    is_configured_room,
)
from mindroom.entity_rooms import get_rooms_for_entity
from mindroom.event_loop_stall import EventLoopStallDetector, start_event_loop_stall_detector
from mindroom.hooks import (
    EVENT_CONFIG_RELOADED,
    ConfigReloadedContext,
    HookRegistry,
    build_hook_matrix_admin,
    build_hook_room_state_putter,
    build_hook_room_state_querier,
    emit,
)
from mindroom.knowledge import KnowledgeRefreshScheduler, reconcile_knowledge_mode_transition_states
from mindroom.knowledge.watch import KnowledgeSourceWatcher
from mindroom.matrix.client_room_admin import get_joined_rooms, get_room_members, invite_to_room
from mindroom.matrix.health import reset_matrix_sync_health
from mindroom.matrix.identity import managed_account_user_id
from mindroom.matrix.rooms import ensure_all_rooms_exist, ensure_root_space, ensure_user_in_rooms
from mindroom.matrix.stale_stream_cleanup import (
    StaleStreamCleanupActor,
    recover_stale_streaming_messages,
)
from mindroom.matrix.state import load_rooms, resolve_room_aliases
from mindroom.matrix.users import (
    INTERNAL_USER_ACCOUNT_KEY,
    INTERNAL_USER_AGENT_NAME,
    AgentMatrixUser,
    ManagedAccountProvisioningRequest,
    create_agent_user,
    preflight_managed_account_provisioning,
)
from mindroom.matrix_identifiers import extract_server_name_from_homeserver
from mindroom.mcp.manager import MCPServerManager
from mindroom.mcp.registry import mcp_tool_name
from mindroom.mcp.toolkit import bind_mcp_server_manager
from mindroom.memory import MemoryAutoFlushWorker, auto_flush_enabled
from mindroom.response_admission import ResponseAdmissionGate
from mindroom.runtime_shutdown import ORDERLY_SHUTDOWN
from mindroom.runtime_state import (
    clear_api_server_address,
    reset_runtime_state,
    set_api_server_address,
    set_runtime_failed,
    set_runtime_ready,
    set_runtime_starting,
)
from mindroom.scheduling_executor import set_scheduling_hook_registry
from mindroom.startup_errors import PermanentStartupError
from mindroom.startup_maintenance import StartupMaintenanceController
from mindroom.tool_approval import shutdown_approval_runtime
from mindroom.tool_system.plugins import (
    PluginReloadResult,
    apply_prepared_plugin_reload,
    deactivate_plugins,
    load_plugins,
    prepare_plugin_reload,
    reload_plugins,
)
from mindroom.tool_system.skills import clear_skill_cache, get_skill_snapshot
from mindroom.workers.runtime import clear_worker_validation_snapshot_cache, shutdown_primary_worker_manager

from . import file_watcher
from .bot import AgentBot, TeamBot, create_bot_for_entity
from .config.main import Config, load_config
from .credentials_sync import sync_env_to_credentials
from .logging_config import get_logger, setup_logging
from .orchestration.config_lifecycle import ConfigReloadLifecycle
from .orchestration.config_updates import configured_entity_names
from .orchestration.external_trigger_runtime import ExternalTriggerRuntimeCoordinator
from .orchestration.plugin_watch import PluginWatchState, watch_plugins_task
from .orchestration.rooms import get_authorized_user_ids_to_invite, get_root_space_user_ids_to_invite
from .orchestration.runtime import (
    STARTUP_RETRY_INITIAL_DELAY_SECONDS,
    STARTUP_RETRY_MAX_DELAY_SECONDS,
    EntityStartResults,
    cancel_sync_task,
    cancel_task,
    create_logged_task,
    is_permanent_startup_error,
    log_startup_phase_finished,
    log_startup_phase_started,
    retry_delay_seconds,
    run_with_retry,
    stop_entities,
    sync_forever_with_restart,
    wait_for_matrix_homeserver,
)
from .orchestration.todo_poke_runtime import TodoPokeRuntimeCoordinator
from .runtime_support import (
    OwnedRuntimeSupport,
    build_owned_runtime_support,
    close_owned_runtime_support,
    sync_owned_runtime_support,
)

if TYPE_CHECKING:
    import socket
    from collections.abc import Awaitable, Callable, Iterable
    from pathlib import Path
    from types import FrameType

    from mindroom.hooks import HookMatrixAdmin, HookMessageSender, HookRoomStatePutter, HookRoomStateQuerier
    from mindroom.matrix.cache import ConversationEventCache

    from .constants import RuntimePaths
    from .orchestration.config_updates import ConfigUpdatePlan
logger = get_logger(__name__)

_AUXILIARY_TASK_RESTART_INITIAL_DELAY_SECONDS = 1.0
_AUXILIARY_TASK_RESTART_MAX_DELAY_SECONDS = 30.0
_EMBEDDED_API_SHUTDOWN_GRACE_SECONDS = 5.0


@dataclass(frozen=True)
class _EmbeddedApiServerContext:
    """Shared identity fields for embedded API server lifecycle logs."""

    host: str
    port: int

    def log_context(self) -> dict[str, object]:
        """Return structured fields common to API server lifecycle logs."""
        return {"host": self.host, "port": self.port}


def _signal_name(sig: int) -> str:
    """Return a stable signal name for lifecycle logs."""
    for candidate in signal.Signals:
        if candidate.value == sig:
            return candidate.name
    return str(sig)


def _raise_embedded_api_server_exit(
    api_server: _EmbeddedApiServerContext,
    *,
    reason: str,
    cause: BaseException | None = None,
) -> NoReturn:
    """Raise the fatal lifecycle error for an unexpected API server exit."""
    logger.error(
        "fatal_embedded_api_server_exit",
        **api_server.log_context(),
        reason=reason,
        exc_info=(type(cause), cause, cause.__traceback__) if cause is not None else None,
    )
    msg = "Embedded API server exited unexpectedly"
    if cause is not None:
        raise RuntimeError(msg) from cause
    raise RuntimeError(msg)


def _raise_orchestrator_exit(*, reason: str) -> NoReturn:
    """Raise the fatal lifecycle error for an unexpected orchestrator exit."""
    logger.error("fatal_orchestrator_exit", reason=reason)
    msg = "MindRoom orchestrator exited unexpectedly"
    raise RuntimeError(msg)


class _SignalAwareUvicornServer(uvicorn.Server):
    """Uvicorn server that marks the shared shutdown event on signal exit."""

    def __init__(self, config: uvicorn.Config, shutdown_requested: asyncio.Event | None) -> None:
        super().__init__(config)
        self._shutdown_requested = shutdown_requested

    async def startup(self, sockets: list[socket.socket] | None = None) -> None:
        """Publish the API address only after Uvicorn successfully binds it."""
        await super().startup(sockets=sockets)
        if not self.started:
            return
        listeners = self.servers[0].sockets
        assert listeners is not None
        bound_address = cast(
            "tuple[str, int] | tuple[str, int, int, int]",
            listeners[0].getsockname(),
        )
        bound_host = bound_address[0]
        bound_port = bound_address[1]
        set_api_server_address(bound_host, bound_port)
        logger.info("embedded_api_server_started", host=bound_host, port=bound_port)

    def handle_exit(self, sig: int, frame: FrameType | None) -> None:
        """Mirror Uvicorn signal handling and surface shutdown to the orchestrator."""
        del frame
        signal_number = int(sig)
        logger.info(
            "embedded_api_server_signal_received",
            signal_number=signal_number,
            signal_name=_signal_name(signal_number),
        )
        if self._shutdown_requested is not None:
            self._shutdown_requested.set()
        if self.should_exit and signal_number == int(signal.SIGINT):
            self.force_exit = True
            return
        self.should_exit = True


@dataclass
class _MultiAgentOrchestrator:
    """Orchestrates multiple agent bots."""

    runtime_paths: RuntimePaths
    api_enabled: bool = True
    storage_path: Path = field(init=False)
    config_path: Path = field(init=False)
    agent_bots: dict[str, AgentBot | TeamBot] = field(default_factory=dict, init=False)
    running: bool = field(default=False, init=False)
    config: Config | None = field(default=None, init=False)
    _sync_tasks: dict[str, asyncio.Task] = field(default_factory=dict, init=False)
    _bot_start_tasks: dict[str, asyncio.Task] = field(default_factory=dict, init=False)
    _memory_auto_flush_worker: MemoryAutoFlushWorker | None = field(default=None, init=False)
    _memory_auto_flush_task: asyncio.Task | None = field(default=None, init=False)
    _todo_poke_runtime: TodoPokeRuntimeCoordinator = field(init=False, repr=False)
    config_reload: ConfigReloadLifecycle = field(init=False)
    _mcp_manager: MCPServerManager | None = field(default=None, init=False)
    _config_update_lock: asyncio.Lock = field(default_factory=asyncio.Lock, init=False)
    _dispatch_recovery_lock: asyncio.Lock = field(default_factory=asyncio.Lock, init=False)
    _dispatch_recovery_task_owner: object = field(default_factory=object, init=False, repr=False)
    _dispatch_recovery_task: asyncio.Task[None] | None = field(default=None, init=False, repr=False)
    _dispatch_recovery_requested: bool = field(default=False, init=False, repr=False)
    _response_admission_gate: ResponseAdmissionGate = field(default_factory=ResponseAdmissionGate, init=False)
    _mcp_catalog_change_task_owner: object = field(default_factory=object, init=False, repr=False)
    _pending_replacement_recovery_room_ids: dict[str, set[str]] = field(default_factory=dict, init=False)
    _runtime_support: OwnedRuntimeSupport = field(init=False)
    _event_cache_write_task_owner: object = field(default_factory=object, init=False)
    plugin_watch: PluginWatchState = field(init=False)
    _knowledge_refresh_scheduler: KnowledgeRefreshScheduler = field(init=False)
    _knowledge_source_watcher: KnowledgeSourceWatcher = field(init=False)
    hook_registry: HookRegistry = field(default_factory=HookRegistry.empty, init=False)
    _runtime_shutdown_event: asyncio.Event | None = field(default=None, init=False, repr=False)
    _external_trigger_runtime: ExternalTriggerRuntimeCoordinator = field(init=False, repr=False)
    _approval_transport: ApprovalMatrixTransport = field(init=False, repr=False)
    _router_principal_id: str | None = field(default=None, init=False, repr=False)
    _startup_maintenance: StartupMaintenanceController = field(init=False, repr=False)

    def __post_init__(self) -> None:
        """Store canonical derived paths from the explicit runtime context."""
        self.storage_path = self.runtime_paths.storage_root
        self.config_path = self.runtime_paths.config_path
        self._runtime_support = build_owned_runtime_support(
            db_path=self.storage_path / "event_cache.db",
            logger=logger,
            background_task_owner=self._event_cache_write_task_owner,
        )
        self._knowledge_refresh_scheduler = KnowledgeRefreshScheduler()
        self._knowledge_source_watcher = KnowledgeSourceWatcher(self._knowledge_refresh_scheduler)
        self.plugin_watch = PluginWatchState(runtime_paths=self.runtime_paths)
        self._external_trigger_runtime = ExternalTriggerRuntimeCoordinator(
            runtime_paths=self.runtime_paths,
            api_enabled=self.api_enabled,
        )
        self._todo_poke_runtime = TodoPokeRuntimeCoordinator(
            runtime_paths=self.runtime_paths,
            config_provider=lambda: self.config,
            bot_provider=lambda entity_name: self.agent_bots.get(entity_name),
        )
        self.config_reload = ConfigReloadLifecycle(
            runtime_paths=self.runtime_paths,
            is_running=lambda: self.running,
            current_config=lambda: self.config,
            agent_bots=lambda: self.agent_bots,
            load_initial_config=self._load_initial_config,
            apply_update_plan=self._apply_config_update_plan,
            response_admission_gate=self._response_admission_gate,
            config_update_lock=self._config_update_lock,
        )
        self._approval_transport = ApprovalMatrixTransport(
            runtime_paths=self.runtime_paths,
            bot_provider=lambda agent_name: self.agent_bots.get(agent_name),
            event_cache_provider=self._approval_event_cache,
        )
        self._startup_maintenance = StartupMaintenanceController(
            recover_stale_streams=lambda bots, config, startup_cutoff_ms, scanned_room_ids: (
                self._recover_stale_streams_after_restart(
                    bots,
                    config,
                    startup_cutoff_ms,
                    scanned_room_ids,
                )
            ),
            setup_rooms_and_memberships=self._setup_startup_rooms_and_memberships,
            sync_runtime_support=lambda config: self._sync_runtime_support_services(config, start_watcher=True),
            mark_runtime_support_ready=lambda: self._approval_transport.mark_startup_runtime_support_ready(),
        )

    @property
    def knowledge_refresh_scheduler(self) -> KnowledgeRefreshScheduler:
        """Return the orchestrator-owned background knowledge refresh scheduler."""
        return self._knowledge_refresh_scheduler

    def entity_first_sync_complete(self, entity_name: str) -> bool | None:
        """Return first-sync readiness for the current entity generation."""
        bot = self.agent_bots.get(entity_name)
        if bot is None:
            return None
        return bot.running and bot.first_sync_complete

    async def _stop_memory_auto_flush_worker(self) -> None:
        """Stop the background memory auto-flush worker if running."""
        worker = self._memory_auto_flush_worker
        task = self._memory_auto_flush_task
        self._memory_auto_flush_worker = None
        self._memory_auto_flush_task = None

        if worker is not None:
            worker.stop()
        if task is not None:
            await asyncio.gather(task, return_exceptions=True)

    async def _sync_memory_auto_flush_worker(self) -> None:
        """Start or stop background memory auto-flush worker based on current config."""
        config = self.config
        if config is None:
            await self._stop_memory_auto_flush_worker()
            return

        if not auto_flush_enabled(config):
            await self._stop_memory_auto_flush_worker()
            return

        task = self._memory_auto_flush_task
        if task is not None and not task.done():
            return

        worker = MemoryAutoFlushWorker(
            storage_path=self.storage_path,
            runtime_paths=self.runtime_paths,
            config_provider=lambda: self.config,
        )
        self._memory_auto_flush_worker = worker
        self._memory_auto_flush_task = asyncio.create_task(worker.run(), name="memory_auto_flush_worker")

    def _reset_runtime_shutdown_event(self) -> asyncio.Event:
        """Create the shutdown event for the current orchestrator run."""
        shutdown_event = asyncio.Event()
        self._runtime_shutdown_event = shutdown_event
        return shutdown_event

    def _capture_runtime_loop(self) -> None:
        """Remember the runtime loop that owns Matrix client I/O."""
        self._approval_transport.capture_runtime_loop()

    async def send_approval_notice(
        self,
        *,
        room_id: str,
        approval_event_id: str,
        thread_id: str | None,
        reason: str,
    ) -> bool:
        """Send one approval notice through the public runtime protocol."""
        return await self._approval_transport.send_notice(
            room_id=room_id,
            approval_event_id=approval_event_id,
            thread_id=thread_id,
            reason=reason,
        )

    def _bind_runtime_support_services(self, bot: AgentBot | TeamBot) -> None:
        """Bind the current runtime support services to one managed bot."""
        bot.admission_gate = self._response_admission_gate
        bot.event_cache = self._runtime_support.event_cache.for_principal(bot.matrix_id.full_id)
        bot.event_cache_write_coordinator = self._runtime_support.event_cache_write_coordinator
        bot.startup_thread_prewarm_registry = self._runtime_support.startup_thread_prewarm_registry

    def _approval_event_cache(self) -> ConversationEventCache:
        """Return the router principal's isolated cache before or after bot construction."""
        router_bot = self.agent_bots.get(ROUTER_AGENT_NAME)
        if router_bot is not None:
            return router_bot.event_cache
        if self._router_principal_id is None:
            msg = "Router Matrix principal is unavailable for approval cache binding"
            raise RuntimeError(msg)
        return self._runtime_support.event_cache.for_principal(self._router_principal_id)

    def _rebind_runtime_support_services(self) -> None:
        """Rebind the current runtime support services to every managed bot."""
        for bot in self.agent_bots.values():
            self._bind_runtime_support_services(bot)

    def _bind_started_runtime_support_services(self, bots: list[AgentBot | TeamBot]) -> None:
        """Bind current runtime support objects needed by live callbacks."""
        for bot in bots:
            self._bind_runtime_support_services(bot)
        self._configure_approval_store_transport()

    async def _setup_startup_rooms_and_memberships(self, bots: list[AgentBot | TeamBot]) -> None:
        """Run startup room setup, then publish trigger delivery runtime."""
        await run_with_retry(
            "Setting up Matrix rooms and memberships",
            lambda: self._setup_rooms_and_memberships(bots),
            permanent_error_check=is_permanent_startup_error,
            update_runtime_state=False,
        )
        self._external_trigger_runtime.bind_if_ready(self.config, self.agent_bots)

    async def _sync_event_cache_service(self, config: Config) -> None:
        """Ensure the runtime has one initialized shared event-cache service."""
        self._runtime_support = await sync_owned_runtime_support(
            self._runtime_support,
            cache_config=config.cache,
            runtime_paths=self.runtime_paths,
            logger=logger,
            background_task_owner=self._event_cache_write_task_owner,
            init_failure_reason_prefix="shared_runtime_init_failed",
            log_db_path_change=True,
        )
        self._rebind_runtime_support_services()

    def _configure_approval_store_transport(self) -> None:
        """Bind approval transport hooks to the current shared runtime services."""
        self._approval_transport.bind_approval_runtime()

    async def _close_runtime_support_services(self) -> None:
        """Close the shared runtime-owned cache services."""
        await close_owned_runtime_support(self._runtime_support, logger=logger)

    async def _ensure_user_account(self, config: Config) -> None:
        """Ensure a user account exists, creating one if necessary.

        This reuses the same `create_agent_user` flow that agents use,
        treating the user as a special internal "agent" account.
        Skipped when `mindroom_user` is not configured, such as hosted/public profiles.
        """
        if config.mindroom_user is None:
            logger.debug("mindroom_user not configured, skipping user account creation")
            return
        # The user account is managed through the same Matrix account lifecycle as bots.
        user_account = await create_agent_user(
            constants.runtime_matrix_homeserver(runtime_paths=self.runtime_paths),
            INTERNAL_USER_AGENT_NAME,
            config.mindroom_user.display_name,
            username=config.mindroom_user.username,
            runtime_paths=self.runtime_paths,
        )
        logger.info("user_account_ready", user_id=user_account.user_id)

    def _require_config(self) -> Config:
        """Return the active config or fail fast if it has not been loaded."""
        config = self.config
        if config is None:
            msg = "Configuration not loaded"
            raise RuntimeError(msg)
        return config

    async def _prepare_user_account(
        self,
        config: Config,
        *,
        update_runtime_state: bool,
    ) -> None:
        """Ensure the internal user account exists, retrying only transient failures."""
        await run_with_retry(
            "Preparing MindRoom user account",
            lambda: self._ensure_user_account(config),
            permanent_error_check=is_permanent_startup_error,
            update_runtime_state=update_runtime_state,
        )

    async def _cancel_bot_start_task(self, entity_name: str) -> None:
        """Cancel any background start task for one bot."""
        task = self._bot_start_tasks.pop(entity_name, None)
        await cancel_task(task)

    async def _cancel_bot_start_tasks(self) -> None:
        """Cancel all background bot start tasks."""
        for entity_name in tuple(self._bot_start_tasks):
            await self._cancel_bot_start_task(entity_name)

    def _running_startup_maintenance_bots(self) -> list[AgentBot | TeamBot]:
        """Return currently running bots in startup-maintenance order."""
        router_bot = self.agent_bots.get(ROUTER_AGENT_NAME)
        bots: list[AgentBot | TeamBot] = []
        if router_bot is not None and router_bot.running:
            bots.append(router_bot)
        bots.extend(
            bot for entity_name, bot in self.agent_bots.items() if entity_name != ROUTER_AGENT_NAME and bot.running
        )
        return bots

    def _start_sync_task(self, entity_name: str, bot: AgentBot | TeamBot) -> None:
        """Ensure one sync task exists for a running bot."""
        existing_task = self._sync_tasks.get(entity_name)
        if existing_task is not None and not existing_task.done():
            return
        self._sync_tasks[entity_name] = asyncio.create_task(
            sync_forever_with_restart(bot),
            name=f"sync_{entity_name}",
        )

    def _bots_to_setup_after_background_start(self, entity_name: str) -> list[AgentBot | TeamBot]:
        """Return the bots whose room memberships should be reconciled after a background start."""
        if entity_name == ROUTER_AGENT_NAME:
            return self._running_bots_for_entities(self.agent_bots)
        return self._running_bots_for_entities((entity_name,))

    def _running_bots_for_entities(self, entity_names: Iterable[str]) -> list[AgentBot | TeamBot]:
        """Return running bots for the given entity names."""
        running_bots: list[AgentBot | TeamBot] = []
        for entity_name in entity_names:
            bot = self.agent_bots.get(entity_name)
            if bot is not None and bot.running:
                running_bots.append(bot)
        return running_bots

    async def _recover_ready_turn_dispatch_obligations(self) -> None:
        """Recover fleet-dependent turns whose required runtimes are ready."""
        async with self._dispatch_recovery_lock:
            config = self.config
            if config is None:
                return
            first_error: Exception | None = None
            for entity_name in configured_entity_names(config):
                bot = self.agent_bots.get(entity_name)
                required_entities = (
                    [entity_name]
                    if entity_name == ROUTER_AGENT_NAME
                    else (
                        [entity_name, *config.teams[entity_name].agents]
                        if entity_name in config.teams
                        else [entity_name]
                    )
                )
                if bot is None or any(
                    self.entity_first_sync_complete(required_entity) is not True
                    for required_entity in required_entities
                ):
                    continue
                try:
                    await bot.recover_pending_turn_dispatch_obligations()
                except asyncio.CancelledError:
                    raise
                except Exception as error:
                    first_error = first_error or error
                    logger.exception(
                        "turn_dispatch_recovery_failed",
                        agent_name=entity_name,
                    )
            if first_error is not None:
                raise first_error

    def _schedule_ready_turn_dispatch_recovery(self) -> None:
        """Coalesce bot-ready signals into one orchestrator-owned recovery task."""
        self._dispatch_recovery_requested = True
        task = self._dispatch_recovery_task
        if task is not None and not task.done():
            return
        self._dispatch_recovery_task = create_background_task(
            self._run_scheduled_turn_dispatch_recovery(),
            name="recover_ready_turn_dispatch_obligations",
            owner=self._dispatch_recovery_task_owner,
        )

    async def _run_scheduled_turn_dispatch_recovery(self) -> None:
        """Drain readiness signals without blocking a Matrix sync callback."""
        current_task = asyncio.current_task()
        try:
            while self._dispatch_recovery_requested:
                self._dispatch_recovery_requested = False
                await run_with_retry(
                    "Recovering ready turn dispatch obligations",
                    self._recover_ready_turn_dispatch_obligations,
                    update_runtime_state=False,
                )
        finally:
            if self._dispatch_recovery_task is current_task:
                self._dispatch_recovery_task = None

    async def _try_start_bot_once(self, entity_name: str, bot: AgentBot | TeamBot) -> bool | None:
        """Run one bot start attempt and classify the result."""
        try:
            started = bool(await bot.try_start())
        except PermanentStartupError:
            logger.error(  # noqa: TRY400
                "Bot startup failed permanently; leaving bot disabled until configuration changes",
                agent_name=entity_name,
            )
            return None
        else:
            return started

    async def _run_bot_start_retry(self, entity_name: str) -> None:
        """Keep retrying one bot start until it succeeds or the task is cancelled."""
        current_task = asyncio.current_task()
        attempt = 0
        try:
            while True:
                bot = self.agent_bots.get(entity_name)
                if bot is None:
                    return

                config = self.config
                if config is not None and entity_name in self._entities_blocked_by_failed_mcp_servers(
                    {entity_name},
                    config,
                ):
                    start_status = False
                else:
                    start_status = await self._try_start_bot_once(entity_name, bot)
                if start_status is None:
                    return
                if start_status:
                    logger.info("Bot recovered after startup failure", agent_name=entity_name)
                    bots_to_setup = self._bots_to_setup_after_background_start(entity_name)
                    self._bind_started_runtime_support_services([bot])
                    self._schedule_ready_turn_dispatch_recovery()
                    config = self.config
                    if config is not None:
                        self._resolve_bot_room_aliases(bots_to_setup, config)
                    self._start_sync_task(entity_name, bot)
                    if bots_to_setup:
                        await run_with_retry(
                            f"Updating Matrix room memberships for {entity_name}",
                            partial(self._setup_rooms_and_memberships, bots_to_setup),
                            permanent_error_check=is_permanent_startup_error,
                            update_runtime_state=False,
                        )
                    if config is not None:
                        await self._recover_pending_replacement_rooms(config)
                    self._external_trigger_runtime.bind_if_ready(self.config, self.agent_bots)
                    return

                attempt += 1
                retry_in_seconds = retry_delay_seconds(
                    attempt,
                    initial_delay_seconds=STARTUP_RETRY_INITIAL_DELAY_SECONDS,
                    max_delay_seconds=STARTUP_RETRY_MAX_DELAY_SECONDS,
                )
                logger.warning(
                    "Bot startup failed; retrying in background",
                    agent_name=entity_name,
                    attempt=attempt,
                    retry_in_seconds=retry_in_seconds,
                )
                await asyncio.sleep(retry_in_seconds)
        finally:
            if self._bot_start_tasks.get(entity_name) is current_task:
                del self._bot_start_tasks[entity_name]

    async def _schedule_bot_start_retry(self, entity_name: str) -> None:
        """Schedule background retries for one failed bot startup."""
        await self._cancel_bot_start_task(entity_name)
        self._bot_start_tasks[entity_name] = create_logged_task(
            self._run_bot_start_retry(entity_name),
            name=f"retry_start_{entity_name}",
            failure_message="Background bot start task failed",
        )

    async def _sync_runtime_support_services(
        self,
        config: Config,
        *,
        start_watcher: bool,
        previous_config: Config | None = None,
    ) -> None:
        """Refresh runtime support services that depend on the active config."""
        if previous_config is not None:
            await asyncio.to_thread(
                reconcile_knowledge_mode_transition_states,
                previous_config,
                config,
                self.runtime_paths,
            )
        await self._knowledge_source_watcher.sync(
            config=config if start_watcher else None,
            runtime_paths=self.runtime_paths,
        )
        ensure_default_agent_workspaces(config, self.storage_path)
        await self._sync_event_cache_service(config)
        self._configure_approval_store_transport()
        await self._sync_memory_auto_flush_worker()
        await self._todo_poke_runtime.sync()

    async def _stop_mcp_manager(self) -> None:
        """Stop the MCP manager and clear the active runtime binding."""
        manager = self._mcp_manager
        self._mcp_manager = None
        bind_mcp_server_manager(None)
        if manager is not None:
            await manager.shutdown()

    async def _sync_mcp_manager(self, config: Config) -> set[str]:
        """Create or refresh the orchestrator-owned MCP manager."""
        manager = self._mcp_manager
        if manager is None:
            manager = MCPServerManager(
                self.runtime_paths,
                on_catalog_change=self._notify_mcp_catalog_change,
            )
            self._mcp_manager = manager
        bind_mcp_server_manager(manager)
        return await manager.sync_servers(config)

    def _entities_blocked_by_failed_mcp_servers(self, entity_names: set[str], config: Config) -> set[str]:
        """Return entities blocked because a required MCP server is currently unavailable."""
        manager = self._mcp_manager
        if manager is None:
            return set()
        failed_server_ids = manager.failed_required_server_ids()
        if not failed_server_ids:
            return set()
        blocked_entities = config.get_entities_referencing_tools(
            {mcp_tool_name(server_id) for server_id in failed_server_ids},
        )
        return blocked_entities & entity_names

    def _log_mcp_degraded_entities(self, config: Config) -> None:
        """Warn once per unavailable optional MCP server about entities running without its tools."""
        manager = self._mcp_manager
        if manager is None:
            return
        running_entities = {entity_name for entity_name, bot in self.agent_bots.items() if bot.running}
        for server_id in sorted(manager.failed_server_ids() - manager.failed_required_server_ids()):
            degraded_entities = config.get_entities_referencing_tools({mcp_tool_name(server_id)}) & running_entities
            if not degraded_entities:
                continue
            logger.warning(
                "Entities running without tools from unavailable MCP server",
                server_id=server_id,
                degraded_entities=sorted(degraded_entities),
            )

    @staticmethod
    def _entity_display_name(config: Config, entity_name: str) -> str:
        """Return the Matrix display name for one configured managed entity."""
        if entity_name == ROUTER_AGENT_NAME:
            return "RouterAgent"
        if entity_name in config.agents:
            return config.agents[entity_name].display_name
        if entity_name in config.teams:
            return config.teams[entity_name].display_name
        return entity_name

    async def _prepare_entity_accounts(
        self,
        config: Config,
        entity_names: Iterable[str],
    ) -> dict[str, AgentMatrixUser]:
        """Ensure managed Matrix accounts exist before runtime bot construction."""
        homeserver = constants.runtime_matrix_homeserver(runtime_paths=self.runtime_paths)
        users: dict[str, AgentMatrixUser] = {}
        entity_names = tuple(entity_names)
        self._preflight_account_provisioning(config, entity_names=entity_names, include_internal_user=False)

        async def _prepare_accounts() -> None:
            nonlocal users
            prepared_users: dict[str, AgentMatrixUser] = {}
            for entity_name in entity_names:
                prepared_users[entity_name] = await create_agent_user(
                    homeserver,
                    entity_name,
                    self._entity_display_name(config, entity_name),
                    runtime_paths=self.runtime_paths,
                )
            self._validate_entity_accounts(config)
            users = prepared_users

        await run_with_retry(
            "Preparing managed Matrix accounts",
            _prepare_accounts,
            permanent_error_check=is_permanent_startup_error,
            update_runtime_state=not self.running,
        )
        return users

    def _validate_entity_accounts(self, config: Config) -> None:
        """Validate persisted Matrix identities for all configured runtime entities."""
        try:
            entity_identity_registry(config, self.runtime_paths)
        except (DuplicateManagedEntityIdentityError, MissingManagedEntityAccountError) as exc:
            raise PermanentStartupError(str(exc)) from exc

    def _preflight_account_provisioning(
        self,
        config: Config,
        *,
        entity_names: Iterable[str],
        include_internal_user: bool,
    ) -> None:
        """Reject account localpart collisions before any missing account is created."""
        requests: list[ManagedAccountProvisioningRequest] = []
        if include_internal_user and config.mindroom_user is not None:
            requests.append(
                ManagedAccountProvisioningRequest(
                    INTERNAL_USER_AGENT_NAME,
                    username=config.mindroom_user.username,
                ),
            )
        requests.extend(ManagedAccountProvisioningRequest(entity_name) for entity_name in entity_names)
        preflight_managed_account_provisioning(requests, self.runtime_paths)

    def validate_managed_entity_identities(self) -> None:
        """Validate persisted managed Matrix identities for the live config."""
        self._validate_entity_accounts(self._require_config())

    def _create_managed_bot(
        self,
        entity_name: str,
        config: Config,
        agent_user: AgentMatrixUser,
    ) -> AgentBot | TeamBot:
        """Create and register one runtime-managed bot."""
        bot = cast(
            "AgentBot | TeamBot",
            create_bot_for_entity(
                entity_name,
                agent_user,
                config,
                self.runtime_paths,
                self.storage_path,
                config_path=self.config_path,
            ),
        )
        bot.orchestrator = self
        bot.hook_registry = self.hook_registry
        self._bind_runtime_support_services(bot)
        self.agent_bots[entity_name] = bot
        if entity_name == ROUTER_AGENT_NAME:
            self._router_principal_id = agent_user.user_id
        return bot

    def _build_hook_registry(self, config: Config) -> HookRegistry:
        """Load plugins and rebuild the immutable hook-registry snapshot."""
        plugins = load_plugins(config, self.runtime_paths)
        return HookRegistry.from_plugins(plugins)

    def _activate_hook_registry(self, hook_registry: HookRegistry) -> None:
        """Commit one hook-registry snapshot to the live runtime."""
        set_scheduling_hook_registry(hook_registry)
        self.hook_registry = hook_registry
        for bot in self.agent_bots.values():
            bot.hook_registry = hook_registry

    @overload
    async def reload_plugins_now(
        self,
        *,
        source: str,
        changed_paths: tuple[Path, ...] = (),
    ) -> PluginReloadResult: ...

    @overload
    async def reload_plugins_now(
        self,
        *,
        source: str,
        changed_paths: tuple[Path, ...] = (),
        expected_revision: int,
    ) -> PluginReloadResult | None: ...

    async def reload_plugins_now(
        self,
        *,
        source: str,
        changed_paths: tuple[Path, ...] = (),
        expected_revision: int | None = None,
    ) -> PluginReloadResult | None:
        """Rebuild and atomically swap the live plugin registry snapshot."""
        if not self.running:
            msg = "Plugin reload unavailable until startup finishes."
            raise RuntimeError(msg)
        async with self._config_update_lock:
            if expected_revision is not None and self.plugin_watch.revision != expected_revision:
                logger.info(
                    "Skipping stale watcher plugin reload",
                    expected_revision=expected_revision,
                    current_revision=self.plugin_watch.revision,
                )
                return None
            config = self._require_config()
            logger.info(
                "Reloading plugins",
                source=source,
                changed_paths=[str(path) for path in changed_paths],
            )
            watch_roots, watch_root_snapshots = self.plugin_watch.capture(config)
            try:
                result = reload_plugins(config, self.runtime_paths)
            except Exception:
                recovery_result, warning_message, warning_kwargs = _recover_failed_plugin_reload(
                    config,
                    self.runtime_paths,
                )
                self._activate_hook_registry(recovery_result.hook_registry)
                clear_worker_validation_snapshot_cache()
                self.plugin_watch.replace_snapshots(watch_roots, watch_root_snapshots)
                logger.warning(warning_message, source=source, **warning_kwargs)
                raise
            self._activate_hook_registry(result.hook_registry)
            clear_worker_validation_snapshot_cache()
            self.plugin_watch.replace_snapshots(watch_roots, watch_root_snapshots)
            logger.info(
                "Plugin reload complete",
                source=source,
                active_plugins=list(result.active_plugin_names),
                cancelled_task_count=result.cancelled_task_count,
            )
            return result

    async def _apply_plugin_changes_for_config_update(
        self,
        *,
        current_config: Config,
        new_config: Config,
        changed_server_ids: set[str],
    ) -> set[str]:
        """Stage and commit plugin changes without interleaving live reloads."""
        prepared_plugin_roots, prepared_plugin_root_snapshots = self.plugin_watch.capture(new_config)
        prepared_plugin_reload = prepare_plugin_reload(
            new_config,
            self.runtime_paths,
            skip_broken_plugins=True,
        )
        pre_stopped_mcp_entities = await self._stop_entities_before_mcp_sync(
            current_config,
            new_config,
            changed_server_ids,
        )
        self.config = new_config
        new_hook_registry = apply_prepared_plugin_reload(
            prepared_plugin_reload,
            cancel_existing_tasks=True,
        ).hook_registry
        self.plugin_watch.replace_snapshots(
            prepared_plugin_roots,
            prepared_plugin_root_snapshots,
        )
        self._activate_hook_registry(new_hook_registry)
        clear_worker_validation_snapshot_cache()
        return pre_stopped_mcp_entities

    async def _start_entities_once(
        self,
        entity_names: Iterable[str],
        *,
        start_sync_tasks: bool,
    ) -> EntityStartResults:
        """Try to start each named entity once and classify the results."""
        entity_bots: list[tuple[str, AgentBot | TeamBot]] = []
        for entity_name in entity_names:
            bot = self.agent_bots.get(entity_name)
            if bot is not None:
                entity_bots.append((entity_name, bot))

        results = EntityStartResults()
        if not entity_bots:
            return results

        config = self.config
        blocked_entities = (
            self._entities_blocked_by_failed_mcp_servers({entity_name for entity_name, _ in entity_bots}, config)
            if config is not None
            else set()
        )
        if blocked_entities:
            results.retryable_entities.extend(sorted(blocked_entities))
            entity_bots = [
                (entity_name, bot) for entity_name, bot in entity_bots if entity_name not in blocked_entities
            ]
        if not entity_bots:
            return results

        start_statuses = await asyncio.gather(
            *[self._try_start_bot_once(entity_name, bot) for entity_name, bot in entity_bots],
        )
        for (entity_name, bot), start_status in zip(entity_bots, start_statuses):
            if start_status:
                results.started_bots.append(bot)
                if start_sync_tasks:
                    self._start_sync_task(entity_name, bot)
                continue
            if start_status is None:
                results.permanently_failed_entities.append(entity_name)
                continue
            results.retryable_entities.append(entity_name)
        return results

    async def _create_and_start_entities(
        self,
        entity_names: set[str],
        config: Config,
        *,
        start_sync_tasks: bool,
    ) -> EntityStartResults:
        """Create configured entities and try to start them once."""
        if not entity_names:
            return EntityStartResults()
        entity_users = await self._prepare_entity_accounts(config, sorted(entity_names))
        for entity_name in sorted(entity_names):
            self._create_managed_bot(entity_name, config, entity_users[entity_name])
        return await self._start_entities_once(entity_names, start_sync_tasks=start_sync_tasks)

    async def initialize(self) -> None:
        """Initialize all managed bots from configuration."""
        self._capture_runtime_loop()
        set_runtime_starting("Loading config and preparing agents")
        logger.info("Initializing multi-agent system...")

        config = await asyncio.to_thread(load_config, self.runtime_paths, tolerate_plugin_load_errors=True)
        hook_registry = await asyncio.to_thread(self._build_hook_registry, config)
        entity_names = configured_entity_names(config)
        self._preflight_account_provisioning(config, entity_names=entity_names, include_internal_user=True)
        await self._prepare_user_account(config, update_runtime_state=True)
        entity_users = await self._prepare_entity_accounts(config, entity_names)
        self._router_principal_id = entity_users[ROUTER_AGENT_NAME].user_id
        self.config = config
        self._activate_hook_registry(hook_registry)
        await self._sync_mcp_manager(config)
        await self._sync_event_cache_service(config)
        self._configure_approval_store_transport()
        for entity_name in entity_names:
            self._create_managed_bot(entity_name, config, entity_users[entity_name])

        logger.info("Initialized agent bots", count=len(self.agent_bots))

    async def start(self) -> None:
        """Start all agent bots and publish readiness state."""
        try:
            await self._start_runtime()
        except asyncio.CancelledError:
            raise
        except Exception as exc:
            set_runtime_failed(str(exc))
            raise

    async def _start_router_bot(self) -> AgentBot:
        """Start the router bot, retrying until it succeeds."""
        config = self._require_config()
        if ROUTER_AGENT_NAME in self._entities_blocked_by_failed_mcp_servers({ROUTER_AGENT_NAME}, config):
            msg = "Router bot depends on unavailable MCP servers"
            raise RuntimeError(msg)
        router_bot = self.agent_bots.get(ROUTER_AGENT_NAME)
        if router_bot is None:
            msg = "Router bot is required for startup"
            raise RuntimeError(msg)

        async def _start_router() -> None:
            if await router_bot.try_start():
                return
            msg = "Router bot failed to start"
            raise RuntimeError(msg)

        set_runtime_starting("Starting router Matrix account")
        await run_with_retry(
            "Starting router Matrix account",
            _start_router,
            permanent_error_check=is_permanent_startup_error,
        )
        return router_bot

    def hook_message_sender(self) -> HookMessageSender | None:
        """Return a router-backed sender for hook contexts when available."""
        router_bot = self.agent_bots.get(ROUTER_AGENT_NAME)
        if router_bot is None:
            return None
        return router_bot._hook_send_message

    def hook_room_state_querier(self) -> HookRoomStateQuerier | None:
        """Return a router-backed room-state querier for hook contexts when available."""
        router_bot = self.agent_bots.get(ROUTER_AGENT_NAME)
        if router_bot is None or router_bot.client is None:
            return None
        return build_hook_room_state_querier(router_bot.client)

    def hook_room_state_putter(self) -> HookRoomStatePutter | None:
        """Return a router-backed room-state putter for hook contexts when available."""
        router_bot = self.agent_bots.get(ROUTER_AGENT_NAME)
        if router_bot is None or router_bot.client is None:
            return None
        return build_hook_room_state_putter(router_bot.client)

    def hook_matrix_admin(self) -> HookMatrixAdmin | None:
        """Return a router-backed Matrix admin helper for hook contexts when available."""
        router_bot = self.agent_bots.get(ROUTER_AGENT_NAME)
        if router_bot is None or router_bot.client is None:
            return None
        return build_hook_matrix_admin(router_bot.client, self.runtime_paths, config=self.config)

    def _log_degraded_startup(self, failed_agents: list[str]) -> None:
        """Log degraded startup status for failed non-router bots."""
        if failed_agents:
            logger.warning(
                "System starting in degraded mode",
                failed_agents=failed_agents,
                failed_agent_count=len(failed_agents),
                operational_agent_count=len(self.agent_bots) - len(failed_agents),
                total_agent_count=len(self.agent_bots),
            )
            return
        logger.info("All agent bots started successfully")

    async def _recover_stale_streams_after_restart(
        self,
        bots: list[AgentBot | TeamBot],
        config: Config,
        startup_cutoff_ms: int | None,
        scanned_room_ids: set[str],
        *,
        target_room_ids: set[str] | None = None,
    ) -> None:
        """Recover interrupted responses from one concurrent room scan."""
        actors: dict[str, StaleStreamCleanupActor] = {}
        for bot in bots:
            if bot.client is None or not bot.agent_user.user_id:
                continue
            actors[bot.agent_user.user_id] = StaleStreamCleanupActor(
                client=bot.client,
                conversation_cache=bot._conversation_cache,
            )
        if not actors:
            return
        router_bot = self._router_bot()

        result = await recover_stale_streaming_messages(
            actors,
            resume_client=router_bot.client if router_bot is not None else None,
            resume_conversation_cache=router_bot._conversation_cache if router_bot is not None else None,
            config=config,
            runtime_paths=self.runtime_paths,
            startup_cutoff_ms=startup_cutoff_ms,
            scanned_room_ids=scanned_room_ids,
            target_room_ids=target_room_ids,
        )
        logger.info(
            "Completed stale stream recovery",
            room_count=result.room_count,
            cleaned_count=result.cleaned_count,
            resumed_count=result.resumed_count,
        )

    def _capture_replacement_recovery_rooms(
        self,
        replaced_bots: dict[str, AgentBot | TeamBot],
    ) -> None:
        """Retain interrupted rooms after their old bot generation stops."""
        for entity_name, bot in replaced_bots.items():
            room_ids = set(bot.pending_sync_restart_retry_room_ids)
            if room_ids:
                self._pending_replacement_recovery_room_ids.setdefault(entity_name, set()).update(room_ids)

    def _replacement_bots(self, entity_names: set[str]) -> dict[str, AgentBot | TeamBot]:
        """Retain bot references across replacement shutdown."""
        return {
            entity_name: self.agent_bots[entity_name] for entity_name in entity_names if entity_name in self.agent_bots
        }

    def _restore_pending_replacement_rooms(
        self,
        claimed_room_ids: dict[str, frozenset[str]],
        scanned_room_ids: set[str],
    ) -> None:
        """Requeue claimed handoffs that were not successfully scanned."""
        for entity_name, room_ids in claimed_room_ids.items():
            unscanned_room_ids = room_ids - scanned_room_ids
            if unscanned_room_ids:
                self._pending_replacement_recovery_room_ids.setdefault(entity_name, set()).update(unscanned_room_ids)

    async def _recover_pending_replacement_rooms(self, config: Config) -> None:
        """Recover captured interruption markers through currently running replacements."""
        if not self._pending_replacement_recovery_room_ids:
            return
        if not config.defaults.auto_resume_after_restart:
            self._pending_replacement_recovery_room_ids.clear()
            return

        router_bot = self._router_bot()
        if router_bot is None or not router_bot.running:
            return
        recovery_bots = [
            bot
            for bot in self._running_bots_for_entities(self._pending_replacement_recovery_room_ids)
            if bot.client is not None and bot.agent_user.user_id
        ]
        if not recovery_bots:
            return
        claimed_room_ids = {
            bot.agent_name: frozenset(self._pending_replacement_recovery_room_ids[bot.agent_name])
            for bot in recovery_bots
        }
        for entity_name, room_ids in claimed_room_ids.items():
            pending_room_ids = self._pending_replacement_recovery_room_ids.get(entity_name)
            if pending_room_ids is None:
                continue
            pending_room_ids.difference_update(room_ids)
            if not pending_room_ids:
                del self._pending_replacement_recovery_room_ids[entity_name]
        scanned_room_ids: set[str] = set()
        try:
            await self._recover_stale_streams_after_restart(
                recovery_bots,
                config,
                None,
                scanned_room_ids,
                target_room_ids=set().union(*claimed_room_ids.values()),
            )
        except BaseException:
            self._restore_pending_replacement_rooms(claimed_room_ids, set())
            raise
        self._restore_pending_replacement_rooms(claimed_room_ids, scanned_room_ids)

    def _resolve_bot_room_aliases(self, bots: list[AgentBot | TeamBot], config: Config) -> None:
        """Resolve currently known room aliases into each bot's configured room IDs."""
        for bot in bots:
            room_aliases = get_rooms_for_entity(bot.agent_name, config)
            bot.rooms = resolve_room_aliases(room_aliases, runtime_paths=self.runtime_paths)

    async def handle_bot_ready(self, bot: AgentBot | TeamBot) -> None:
        """Handle bot-ready notifications through the public runtime protocol."""
        await self._approval_transport.handle_bot_ready(bot)
        self._schedule_ready_turn_dispatch_recovery()

    async def _start_runtime(self) -> None:
        """Run the startup sequence before handing off to the sync loops."""
        runtime_shutdown_event = self._reset_runtime_shutdown_event()
        self._approval_transport.reset_startup_cleanup_gate()
        phase_started = log_startup_phase_started("wait_for_matrix_homeserver")
        await wait_for_matrix_homeserver(runtime_paths=self.runtime_paths)
        log_startup_phase_finished("wait_for_matrix_homeserver", phase_started)

        if not self.agent_bots:
            phase_started = log_startup_phase_started("initialize_runtime")
            await self.initialize()
            log_startup_phase_finished("initialize_runtime", phase_started)

        phase_started = log_startup_phase_started("start_router_bot")
        router_bot = await self._start_router_bot()
        log_startup_phase_finished("start_router_bot", phase_started)

        set_runtime_starting("Starting remaining Matrix bot accounts")
        phase_started = log_startup_phase_started("start_remaining_bots")
        start_results = await self._start_entities_once(
            [entity_name for entity_name in self.agent_bots if entity_name != ROUTER_AGENT_NAME],
            start_sync_tasks=False,
        )
        log_startup_phase_finished("start_remaining_bots", phase_started)

        started_bots = [router_bot, *start_results.started_bots]
        self._log_degraded_startup(
            [*start_results.retryable_entities, *start_results.permanently_failed_entities],
        )

        config = self._require_config()
        self._log_mcp_degraded_entities(config)
        self._resolve_bot_room_aliases(started_bots, config)
        phase_started = log_startup_phase_started("bind_runtime_support")
        self._bind_started_runtime_support_services(started_bots)
        log_startup_phase_finished("bind_runtime_support", phase_started)

        self._schedule_ready_turn_dispatch_recovery()
        self.running = True

        # Create sync tasks for each bot with automatic restart on failure.
        set_runtime_starting("Starting Matrix sync loops")
        startup_cutoff_ms = int(time.time() * 1000)
        phase_started = log_startup_phase_started("start_matrix_sync_loops")
        for entity_name, bot in self.agent_bots.items():
            if bot.running:
                self._start_sync_task(entity_name, bot)
        log_startup_phase_finished("start_matrix_sync_loops", phase_started)

        self._startup_maintenance.start(started_bots, config, startup_cutoff_ms=startup_cutoff_ms)

        for entity_name in start_results.retryable_entities:
            await self._schedule_bot_start_retry(entity_name)

        create_background_task(
            check_embedder_health(config, self.runtime_paths, reason="startup"),
            name="embedder_startup_health_check",
        )
        set_runtime_ready()
        # Stay alive until explicit shutdown. Hot reload replaces sync tasks in
        # self._sync_tasks, so awaiting the initial task generation would let a
        # config-triggered restart look like normal orchestrator completion.
        await runtime_shutdown_event.wait()

    async def _load_initial_config(self, new_config: Config) -> bool:
        """Handle config loading before the runtime has an active config."""
        hook_registry = self._build_hook_registry(new_config)
        entity_names = configured_entity_names(new_config)
        self._preflight_account_provisioning(
            new_config,
            entity_names=entity_names,
            include_internal_user=True,
        )
        await self._prepare_user_account(new_config, update_runtime_state=not self.running)
        await self._prepare_entity_accounts(new_config, entity_names)
        self.config = new_config
        self._activate_hook_registry(hook_registry)
        await self._sync_mcp_manager(new_config)
        await self._sync_runtime_support_services(new_config, start_watcher=self.running)
        clear_worker_validation_snapshot_cache()
        return False

    async def _update_unchanged_bots(self, plan: ConfigUpdatePlan) -> None:
        """Apply the new config to bots that do not require restart."""
        unchanged_bots = [
            (entity_name, bot)
            for entity_name, bot in self.agent_bots.items()
            if entity_name not in plan.entities_to_restart
        ]
        for _, bot in unchanged_bots:
            bot.config = plan.new_config
            bot.enable_streaming = plan.new_config.defaults.enable_streaming
            bot.hook_registry = self.hook_registry

        for entity_name, bot in unchanged_bots:
            try:
                await bot._set_presence_with_model_info()
            except Exception:
                logger.exception("bot_presence_update_failed", agent=entity_name)
            logger.debug("bot_config_updated", agent=entity_name)

    async def _emit_config_reloaded(
        self,
        *,
        new_config: Config,
        changed_entities: set[str],
        added_entities: set[str],
        removed_entities: set[str],
        plugin_changes: tuple[str, ...],
    ) -> None:
        """Emit the config:reloaded observer event after applying a new snapshot."""
        if not self.hook_registry.has_hooks(EVENT_CONFIG_RELOADED):
            return

        router_bot = self.agent_bots.get(ROUTER_AGENT_NAME)
        context = ConfigReloadedContext(
            event_name=EVENT_CONFIG_RELOADED,
            plugin_name="",
            settings={},
            config=new_config,
            runtime_paths=self.runtime_paths,
            logger=logger.bind(event_name=EVENT_CONFIG_RELOADED),
            correlation_id=f"config-reload:{uuid4().hex}",
            runtime_started_at=(router_bot.runtime_started_at if isinstance(router_bot, AgentBot) else time.time()),
            message_sender=self.hook_message_sender(),
            agent_message_snapshot_reader=(
                router_bot._hook_agent_message_snapshot if isinstance(router_bot, AgentBot) else None
            ),
            matrix_admin=self.hook_matrix_admin(),
            room_state_querier=self.hook_room_state_querier(),
            room_state_putter=self.hook_room_state_putter(),
            changed_entities=tuple(sorted(changed_entities)),
            added_entities=tuple(sorted(added_entities)),
            removed_entities=tuple(sorted(removed_entities)),
            plugin_changes=plugin_changes,
        )
        await emit(self.hook_registry, EVENT_CONFIG_RELOADED, context)

    async def _remove_deleted_entities(self, removed_entities: set[str]) -> None:
        """Cancel, clean up, and unregister entities removed from config."""
        self._external_trigger_runtime.unbind_for_entity_changes(removed_entities)
        for entity_name in removed_entities:
            self._pending_replacement_recovery_room_ids.pop(entity_name, None)
            await self._cancel_bot_start_task(entity_name)
            await cancel_sync_task(entity_name, self._sync_tasks)

            bot = self.agent_bots.pop(entity_name, None)
            if bot is not None:
                await bot.cleanup()

    async def _stop_entities_before_mcp_sync(
        self,
        current_config: Config,
        new_config: Config,
        changed_server_ids: set[str],
    ) -> set[str]:
        """Stop MCP-dependent entities before removing or reconfiguring their servers."""
        if not changed_server_ids:
            return set()

        affected_entities = current_config.get_entities_referencing_tools(
            {mcp_tool_name(server_id) for server_id in changed_server_ids},
        ) | new_config.get_entities_referencing_tools({mcp_tool_name(server_id) for server_id in changed_server_ids})
        if not affected_entities:
            return set()

        self._external_trigger_runtime.unbind_for_entity_changes(affected_entities)
        replaced_bots = self._replacement_bots(affected_entities)
        for entity_name in affected_entities:
            await self._cancel_bot_start_task(entity_name)
        await stop_entities(
            affected_entities,
            self.agent_bots,
            self._sync_tasks,
            restart_entities=affected_entities & set(configured_entity_names(new_config)),
        )
        self._capture_replacement_recovery_rooms(replaced_bots)
        return affected_entities

    async def _restart_changed_entities(
        self,
        plan: ConfigUpdatePlan,
        *,
        already_stopped_entities: set[str] | None = None,
    ) -> tuple[set[str], list[str], list[str]]:
        """Restart or create entities affected by the config change."""
        entities_to_stop = plan.entities_to_restart - (already_stopped_entities or set())
        replaced_bots = self._replacement_bots(plan.entities_to_restart)
        if entities_to_stop:
            self._external_trigger_runtime.unbind_for_entity_changes(entities_to_stop)
            for entity_name in entities_to_stop:
                await self._cancel_bot_start_task(entity_name)
            await stop_entities(
                entities_to_stop,
                self.agent_bots,
                self._sync_tasks,
                restart_entities=entities_to_stop & plan.configured_entities,
            )

        self._capture_replacement_recovery_rooms(replaced_bots)
        entities_to_recreate = plan.entities_to_restart & plan.configured_entities
        changed_entities = entities_to_recreate | plan.new_entities
        start_results = await self._create_and_start_entities(
            changed_entities,
            plan.new_config,
            start_sync_tasks=True,
        )

        removed_restarted_entities = plan.entities_to_restart - plan.configured_entities
        for entity_name in removed_restarted_entities:
            self._pending_replacement_recovery_room_ids.pop(entity_name, None)
            self.agent_bots.pop(entity_name, None)

        await self._remove_deleted_entities(plan.removed_entities)
        self._schedule_ready_turn_dispatch_recovery()
        return changed_entities, start_results.retryable_entities, start_results.permanently_failed_entities

    async def _handle_mcp_catalog_change(self, server_id: str) -> None:
        """Restart entities that reference one changed MCP catalog."""
        config = self.config
        if not self.running or config is None:
            return
        if not config.get_entities_referencing_tools({mcp_tool_name(server_id)}):
            clear_worker_validation_snapshot_cache()
            return
        await self.config_reload.apply_with_response_admission(
            partial(self._apply_mcp_catalog_change, server_id),
            operation_name="MCP catalog restart",
            request_is_current=lambda: self.running and self.config is not None,
        )

    async def _notify_mcp_catalog_change(self, server_id: str) -> None:
        """Schedule a catalog restart so an admitted MCP call can release first."""
        if not self.running:
            return
        create_background_task(
            self._handle_mcp_catalog_change(server_id),
            name=f"mcp_catalog_change:{server_id}",
            owner=self._mcp_catalog_change_task_owner,
        )

    async def _apply_mcp_catalog_change(self, server_id: str) -> None:
        """Apply one MCP catalog-triggered entity replacement."""
        async with self._config_update_lock:
            if not self.running or self.config is None:
                return
            clear_worker_validation_snapshot_cache()
            changed_entities = self.config.get_entities_referencing_tools({mcp_tool_name(server_id)})
            if not changed_entities:
                return
            logger.info(
                "Restarting entities after MCP catalog change",
                server_id=server_id,
                entities=sorted(changed_entities),
            )
            self._external_trigger_runtime.unbind_for_entity_changes(changed_entities)
            replaced_bots = self._replacement_bots(changed_entities)
            for entity_name in changed_entities:
                await self._cancel_bot_start_task(entity_name)
            await stop_entities(
                changed_entities,
                self.agent_bots,
                self._sync_tasks,
                restart_entities=changed_entities,
            )
            self._capture_replacement_recovery_rooms(replaced_bots)
            start_results = await self._create_and_start_entities(
                changed_entities,
                self.config,
                start_sync_tasks=True,
            )
            self._schedule_ready_turn_dispatch_recovery()
            if start_results.started_bots:
                await self._setup_rooms_and_memberships(start_results.started_bots)
            await self._recover_pending_replacement_rooms(self.config)
            self._external_trigger_runtime.bind_if_ready(self.config, self.agent_bots)
            for entity_name in start_results.retryable_entities:
                await self._schedule_bot_start_retry(entity_name)
            if start_results.permanently_failed_entities:
                logger.warning(
                    "MCP catalog restart left some bots disabled",
                    server_id=server_id,
                    entities=start_results.permanently_failed_entities,
                )

    async def _reconcile_post_update_rooms(
        self,
        plan: ConfigUpdatePlan,
        changed_entities: set[str],
    ) -> None:
        """Reconcile rooms and memberships after entity/config updates."""
        bots_to_setup = self._running_bots_for_entities(changed_entities)
        if bots_to_setup or plan.mindroom_user_changed or plan.matrix_room_access_changed or plan.authorization_changed:
            await self._setup_rooms_and_memberships(bots_to_setup)
        if plan.matrix_space_changed or plan.room_metadata_changed:
            room_ids = await self._ensure_rooms_exist()
            await self._ensure_root_space(room_ids)

    async def _prepare_accounts_for_config_update(self, new_config: Config, plan: ConfigUpdatePlan) -> None:
        """Prepare or validate managed Matrix accounts before publishing a reloaded config."""
        entities_requiring_account_check = plan.added_entities | (plan.entities_to_restart & plan.configured_entities)
        self._preflight_account_provisioning(
            new_config,
            entity_names=entities_requiring_account_check,
            include_internal_user=plan.mindroom_user_changed,
        )
        if plan.mindroom_user_changed:
            await self._prepare_user_account(new_config, update_runtime_state=not self.running)
        if entities_requiring_account_check:
            await self._prepare_entity_accounts(new_config, entities_requiring_account_check)
        elif plan.mindroom_user_changed:
            self._validate_entity_accounts(new_config)

    async def _finalize_config_reload(
        self,
        *,
        new_config: Config,
        current_config: Config,
        changed_entities: set[str],
        added_entities: set[str],
        removed_entities: set[str],
        plugin_changes: tuple[str, ...],
    ) -> None:
        """Publish post-reload runtime services and external trigger delivery binding."""
        handle_embedder_config_reload(current_config, new_config, self.runtime_paths)
        await self._sync_runtime_support_services(
            new_config,
            start_watcher=self.running,
            previous_config=current_config,
        )
        await self._approval_transport.mark_startup_runtime_support_ready()
        self._external_trigger_runtime.bind_if_ready(new_config, self.agent_bots)
        await self._emit_config_reloaded(
            new_config=new_config,
            changed_entities=changed_entities,
            added_entities=added_entities,
            removed_entities=removed_entities,
            plugin_changes=plugin_changes,
        )

    async def _apply_config_update_plan(
        self,
        current_config: Config,
        plan: ConfigUpdatePlan,
        plugin_changes: tuple[str, ...],
    ) -> bool:
        """Apply one computed config update plan: restart entities and reconcile state."""
        new_config = plan.new_config
        await self._prepare_accounts_for_config_update(new_config, plan)
        replay_startup_maintenance = await self._startup_maintenance.cancel()

        try:
            if plugin_changes:
                pre_stopped_mcp_entities = await self._apply_plugin_changes_for_config_update(
                    current_config=current_config,
                    new_config=new_config,
                    changed_server_ids=plan.changed_mcp_servers,
                )
            else:
                pre_stopped_mcp_entities = await self._stop_entities_before_mcp_sync(
                    current_config,
                    new_config,
                    plan.changed_mcp_servers,
                )
                # Only apply the new config after validation and account checks succeed.
                self.config = new_config
                self.plugin_watch.sync_roots(new_config)
                self._activate_hook_registry(self.hook_registry)
                clear_worker_validation_snapshot_cache()
            changed_runtime_mcp_servers = await self._sync_mcp_manager(new_config)
            await self._sync_event_cache_service(new_config)
            logger.info(
                "updating_config_authorization",
                authorized_user_ids=new_config.authorization.global_users,
            )
            await self._external_trigger_runtime.sync_api_config_snapshot(new_config)
            if changed_runtime_mcp_servers:
                plan = replace(
                    plan,
                    entities_to_restart=plan.entities_to_restart
                    | new_config.get_entities_referencing_tools(
                        {mcp_tool_name(server_id) for server_id in changed_runtime_mcp_servers},
                    ),
                )
            await self._update_unchanged_bots(plan)

            if plan.only_support_service_changes:
                await self._finalize_config_reload(
                    new_config=new_config,
                    current_config=current_config,
                    changed_entities=set(),
                    added_entities=plan.added_entities,
                    removed_entities=plan.removed_entities,
                    plugin_changes=plugin_changes,
                )
                return False

            changed_entities, retryable_entities, permanently_failed_entities = await self._restart_changed_entities(
                plan,
                already_stopped_entities=pre_stopped_mcp_entities,
            )
            await self._reconcile_post_update_rooms(plan, changed_entities)
            await self._recover_pending_replacement_rooms(new_config)

            for entity_name in retryable_entities:
                await self._schedule_bot_start_retry(entity_name)

            if permanently_failed_entities:
                logger.warning(
                    "Configuration update left some bots disabled due to permanent startup errors",
                    agent_names=permanently_failed_entities,
                )

            await self._finalize_config_reload(
                new_config=new_config,
                current_config=current_config,
                changed_entities=changed_entities,
                added_entities=plan.added_entities,
                removed_entities=plan.removed_entities,
                plugin_changes=plugin_changes,
            )

            logger.info(
                "configuration_update_complete",
                affected_bot_count=len(plan.entities_to_restart) + len(plan.new_entities),
            )
            return True
        finally:
            if replay_startup_maintenance and self.running and self.config is not None:
                self._startup_maintenance.restart_after_config_reload(
                    config=self.config,
                    running_bots=self._running_startup_maintenance_bots,
                )

    def _router_bot(self) -> AgentBot | TeamBot | None:
        """Return the router bot when it exists and has an active client."""
        router_bot = self.agent_bots.get(ROUTER_AGENT_NAME)
        if router_bot is None:
            logger.warning("Router not available")
            return None
        if router_bot.client is None:
            logger.warning("Router client not available")
            return None
        return router_bot

    async def _setup_rooms_and_memberships(self, bots: list[AgentBot | TeamBot]) -> None:
        """Setup rooms and ensure all bots have correct memberships.

        This shared flow is used during both initial startup and config updates.
        """
        # Ensure all configured rooms exist before reconciling memberships.
        room_ids = await self._ensure_rooms_exist()
        await self._ensure_root_space(room_ids)

        # Resolve room aliases now that any missing rooms have been created.
        config = self._require_config()
        self._resolve_bot_room_aliases(bots, config)

        async def _ensure_internal_user_memberships() -> None:
            all_rooms = load_rooms(runtime_paths=self.runtime_paths)
            all_room_ids = {room_key: room.room_id for room_key, room in all_rooms.items()}
            if all_room_ids and config.mindroom_user is not None:
                await ensure_user_in_rooms(
                    constants.runtime_matrix_homeserver(runtime_paths=self.runtime_paths),
                    all_room_ids,
                    self.runtime_paths,
                )

        # First invitation and join pass for rooms the router already manages.
        await self._ensure_room_invitations()
        await _ensure_internal_user_memberships()
        await asyncio.gather(*(bot.ensure_rooms() for bot in bots))

        # Existing invite-only rooms may only become manageable after the router joins.
        # Rerun room reconciliation so topic and access policy updates apply in that case.
        if any(bot.agent_name == ROUTER_AGENT_NAME for bot in bots):
            room_ids = await self._ensure_rooms_exist()
            await self._ensure_root_space(room_ids)

        # Retry invitations once the router has completed its first join pass.
        await self._ensure_room_invitations()
        await _ensure_internal_user_memberships()

        follow_up_bots = [bot for bot in bots if bot.agent_name != ROUTER_AGENT_NAME]
        if follow_up_bots:
            await asyncio.gather(*(bot.ensure_rooms() for bot in follow_up_bots))

        logger.info("All agents have joined their configured rooms")

    async def _ensure_rooms_exist(self) -> dict[str, str]:
        """Ensure all configured rooms exist, creating them if necessary.

        The router bot performs room creation because it holds the required permissions.
        """
        router_bot = self._router_bot()
        if router_bot is None:
            return {}
        assert router_bot.client is not None

        config = self._require_config()
        room_ids = await ensure_all_rooms_exist(router_bot.client, config, self.runtime_paths)
        logger.info("ensured_room_existence", room_count=len(room_ids))
        return room_ids

    async def _ensure_root_space(self, room_ids: dict[str, str] | None = None) -> None:
        """Ensure the optional root Matrix Space exists and link the current managed rooms."""
        router_bot = self._router_bot()
        if router_bot is None:
            return
        assert router_bot.client is not None

        config = self._require_config()
        if not config.matrix_space.enabled:
            return

        normalized_room_ids = room_ids if isinstance(room_ids, dict) else {}
        root_space_user_ids = get_root_space_user_ids_to_invite(config, self.runtime_paths)
        root_space_id = await ensure_root_space(
            router_bot.client,
            config,
            self.runtime_paths,
            normalized_room_ids,
            admin_user_ids=root_space_user_ids,
        )
        if root_space_id is None:
            return

        invite_user_ids = root_space_user_ids
        if not invite_user_ids:
            return

        current_members = await get_room_members(router_bot.client, root_space_id)
        if current_members is None:
            logger.warning("room_invitations_skipped_members_unavailable", room_id=root_space_id)
            return
        for user_id in sorted(invite_user_ids):
            log_context = {"user_id": user_id, "room_id": root_space_id}
            await self._invite_user_if_missing(
                root_space_id,
                user_id,
                current_members,
                success_message="invited_user_to_root_space",
                failure_message="invite_user_to_root_space_failed",
                log_context=log_context,
            )

    async def _invite_user_if_missing(
        self,
        room_id: str,
        user_id: str,
        current_members: set[str],
        *,
        success_message: str,
        failure_message: str,
        log_context: dict[str, str] | None = None,
    ) -> None:
        """Invite one user if they are not already a member."""
        router_bot = self._router_bot()
        if router_bot is None:
            return
        assert router_bot.client is not None
        if user_id in current_members:
            return
        success = await invite_to_room(router_bot.client, room_id, user_id)
        if success:
            logger.info(success_message, **(log_context or {}))
            current_members.add(user_id)
        else:
            logger.warning(failure_message, **(log_context or {}))

    async def _invite_internal_user_to_rooms(
        self,
        config: Config,
        joined_rooms: list[str],
        authorized_user_ids: set[str],
    ) -> set[str]:
        """Invite the configured internal user to all joined rooms when needed."""
        router_bot = self._router_bot()
        if router_bot is None:
            return authorized_user_ids
        assert router_bot.client is not None

        server_name = extract_server_name_from_homeserver(
            constants.runtime_matrix_homeserver(runtime_paths=self.runtime_paths),
            runtime_paths=self.runtime_paths,
        )
        user_id = managed_account_user_id(INTERNAL_USER_ACCOUNT_KEY, server_name, self.runtime_paths)
        if config.mindroom_user is None or user_id is None:
            return authorized_user_ids

        authorized_user_ids.discard(user_id)
        for room_id in joined_rooms:
            room_members = await get_room_members(router_bot.client, room_id)
            if room_members is None:
                logger.warning("room_invitations_skipped_members_unavailable", room_id=room_id)
                continue
            await self._invite_user_if_missing(
                room_id,
                user_id,
                room_members,
                success_message=f"Invited user {user_id} to room {room_id}",
                failure_message=f"Failed to invite user {user_id} to room {room_id}",
            )
        return authorized_user_ids

    async def _invite_authorized_users_to_room(
        self,
        room_id: str,
        current_members: set[str],
        authorized_user_ids: set[str],
        config: Config,
    ) -> None:
        """Invite authorized human users who can access a given room."""
        for authorized_user_id in authorized_user_ids:
            if not is_authorized_sender(authorized_user_id, config, room_id, self.runtime_paths):
                continue
            await self._invite_user_if_missing(
                room_id,
                authorized_user_id,
                current_members,
                success_message=f"Invited authorized user {authorized_user_id} to room {room_id}",
                failure_message=f"Failed to invite authorized user {authorized_user_id} to room {room_id}",
            )

    async def _invite_configured_bots_to_room(
        self,
        room_id: str,
        current_members: set[str],
        configured_bot_ids: Iterable[str],
    ) -> None:
        """Invite all configured bots for a room."""
        for bot_user_id in configured_bot_ids:
            await self._invite_user_if_missing(
                room_id,
                bot_user_id,
                current_members,
                success_message=f"Invited {bot_user_id} to room {room_id}",
                failure_message=f"Failed to invite {bot_user_id} to room {room_id}",
            )

    async def _ensure_room_invitations(self) -> None:
        """Ensure all agents and the internal user are invited to their configured rooms.

        The router client performs these invitations because it has admin privileges
        across the managed rooms.
        """
        router_bot = self._router_bot()
        if router_bot is None:
            return
        assert router_bot.client is not None

        config = self.config
        if not config:
            logger.warning("No configuration available, cannot ensure room invitations")
            return

        joined_rooms = await get_joined_rooms(router_bot.client)
        if not joined_rooms:
            return

        authorized_user_ids = get_authorized_user_ids_to_invite(config)
        authorized_user_ids = await self._invite_internal_user_to_rooms(
            config,
            joined_rooms,
            authorized_user_ids,
        )

        for room_id in joined_rooms:
            configured_bots = configured_bot_user_ids_for_room(config, room_id, self.runtime_paths)
            if not configured_bots and not is_configured_room(config, room_id, self.runtime_paths):
                continue

            current_members = await get_room_members(router_bot.client, room_id)
            if current_members is None:
                logger.warning("room_invitations_skipped_members_unavailable", room_id=room_id)
                continue
            await self._invite_authorized_users_to_room(room_id, current_members, authorized_user_ids, config)
            if configured_bots:
                await self._invite_configured_bots_to_room(room_id, current_members, configured_bots)

        logger.info("Ensured room invitations for all configured responders and authorized users")

    async def stop(self) -> None:
        """Stop all agent bots."""
        self.running = False
        if self._runtime_shutdown_event is not None:
            self._runtime_shutdown_event.set()
        self._external_trigger_runtime.unbind()
        await shutdown_approval_runtime()
        await self.config_reload.cancel()
        owner = self._mcp_catalog_change_task_owner
        await wait_for_background_tasks(5.0, owner=owner, shutdown_intent=ORDERLY_SHUTDOWN)
        await wait_for_background_tasks(
            5.0,
            owner=self._dispatch_recovery_task_owner,
            shutdown_intent=ORDERLY_SHUTDOWN,
        )
        await self._startup_maintenance.cancel()
        await self._todo_poke_runtime.stop()
        await self._stop_memory_auto_flush_worker()
        await self._knowledge_source_watcher.shutdown()
        await self._knowledge_refresh_scheduler.shutdown()
        await self._cancel_bot_start_tasks()
        await self._stop_mcp_manager()

        # Cancel sync tasks first so shutdown does not race with active sync loops.
        for entity_name in list(self._sync_tasks.keys()):
            await cancel_sync_task(entity_name, self._sync_tasks)

        for bot in self.agent_bots.values():
            bot.running = False

        stop_tasks = [bot.stop(shutdown_intent=ORDERLY_SHUTDOWN) for bot in self.agent_bots.values()]
        await asyncio.gather(*stop_tasks)
        await self._close_runtime_support_services()
        logger.info("All agent bots stopped")


def _recover_failed_plugin_reload(
    config: Config,
    runtime_paths: RuntimePaths,
) -> tuple[PluginReloadResult, str, dict[str, object]]:
    """Return the fail-broken recovery result after one strict reload failure."""
    try:
        recovery_result = reload_plugins(config, runtime_paths, skip_broken_plugins=True)
    except Exception as degraded_error:
        return (
            deactivate_plugins(),
            "Plugin reload failed; all plugins deactivated",
            {"degraded_error": str(degraded_error)},
        )
    return (
        recovery_result,
        "Plugin reload failed; active plugin set degraded",
        {"active_plugins": list(recovery_result.active_plugin_names)},
    )


async def _handle_config_change(orchestrator: _MultiAgentOrchestrator) -> None:
    """Handle configuration file changes."""
    logger.info("Configuration file changed; queueing hot reload")
    orchestrator.config_reload.request_reload()


async def _watch_config_task(config_path: Path, orchestrator: _MultiAgentOrchestrator) -> None:
    """Watch the config file plus every file it pulls in via !include."""

    def config_source_paths() -> Iterable[Path]:
        config = orchestrator.config
        watched: set[Path] = {config_path}
        if config is not None and config.source_files:
            watched.update(config.source_files)
        # A failed reload's own source set covers include files the last good
        # config never referenced, so fixing them still triggers a retry.
        failed_source_files = orchestrator.config_reload.failed_reload_source_files
        if failed_source_files:
            watched.update(failed_source_files)
        return watched

    async def on_config_change() -> None:
        await _handle_config_change(orchestrator)

    await file_watcher.watch_paths(config_source_paths, on_config_change)


async def _watch_skills_task(orchestrator: _MultiAgentOrchestrator) -> None:
    """Watch skill roots for changes and clear cached skills."""
    while not orchestrator.running:  # noqa: ASYNC110
        await asyncio.sleep(0.1)
    last_snapshot = await asyncio.to_thread(get_skill_snapshot)
    while orchestrator.running:
        await asyncio.sleep(1.0)
        snapshot = await asyncio.to_thread(get_skill_snapshot)
        if snapshot != last_snapshot:
            last_snapshot = snapshot
            clear_skill_cache()
            logger.info("Skills changed; cache cleared")


async def _run_api_server(
    host: str,
    port: int,
    log_level: str,
    runtime_paths: RuntimePaths,
    knowledge_refresh_scheduler: KnowledgeRefreshScheduler | None = None,
    shutdown_requested: asyncio.Event | None = None,
) -> None:
    """Run the bundled dashboard/API server as an asyncio task."""
    from mindroom.api import main as api_main  # noqa: PLC0415

    api_server = _EmbeddedApiServerContext(host=host, port=port)
    api_main.initialize_api_app(api_main.app, runtime_paths)
    if knowledge_refresh_scheduler is not None:
        api_main.bind_orchestrator_knowledge_refresh_scheduler(api_main.app, knowledge_refresh_scheduler)
    config = uvicorn.Config(
        api_main.app,
        host=host,
        port=port,
        log_level=log_level.lower(),
        ws="websockets-sansio",
    )
    server = _SignalAwareUvicornServer(config, shutdown_requested)
    logger.info("embedded_api_server_starting", **api_server.log_context())
    try:
        try:
            await server.serve()
        except SystemExit as exc:
            _raise_embedded_api_server_exit(api_server, reason="server.serve() raised SystemExit", cause=exc)
    finally:
        clear_api_server_address()
    shutdown_expected = shutdown_requested.is_set() if shutdown_requested is not None else False
    logger.info(
        "embedded_api_server_serve_returned",
        **api_server.log_context(),
        shutdown_expected=shutdown_expected,
        server_should_exit=server.should_exit,
        server_force_exit=server.force_exit,
    )
    if not shutdown_expected:
        _raise_embedded_api_server_exit(
            api_server,
            reason="server.serve() returned while application shutdown was not requested",
        )


async def _run_auxiliary_task_forever(
    task_name: str,
    operation: Callable[[], Awaitable[None]],
    *,
    should_restart: Callable[[], bool] | None = None,
) -> None:
    """Restart a non-critical background task whenever it exits or crashes."""
    restart_allowed = (lambda: True) if should_restart is None else should_restart
    restart_count = 0
    while restart_allowed():
        started_at = time.monotonic()
        try:
            await operation()
            if not restart_allowed():
                return
            logger.warning("Auxiliary task exited; restarting", task_name=task_name)
        except asyncio.CancelledError:
            raise
        except Exception:
            if not restart_allowed():
                return
            logger.exception(
                "Auxiliary task crashed; restarting",
                task_name=task_name,
            )
        if not restart_allowed():
            return
        if time.monotonic() - started_at >= _AUXILIARY_TASK_RESTART_MAX_DELAY_SECONDS:
            restart_count = 0
        restart_count += 1
        await asyncio.sleep(
            retry_delay_seconds(
                restart_count,
                initial_delay_seconds=_AUXILIARY_TASK_RESTART_INITIAL_DELAY_SECONDS,
                max_delay_seconds=_AUXILIARY_TASK_RESTART_MAX_DELAY_SECONDS,
            ),
        )


async def _wait_for_runtime_completion(
    *,
    orchestrator_task: asyncio.Task[None],
    shutdown_wait_task: asyncio.Task[bool],
    api_task: asyncio.Task[None] | None,
    shutdown_requested: asyncio.Event,
    api_server: _EmbeddedApiServerContext,
) -> None:
    """Wait until the orchestrator, API server, or shutdown signal ends the run."""
    monitored_tasks: set[asyncio.Task] = {orchestrator_task, shutdown_wait_task}
    if api_task is not None:
        monitored_tasks.add(api_task)
    done, _pending = await asyncio.wait(monitored_tasks, return_when=asyncio.FIRST_COMPLETED)
    logger.info(
        "runtime_completion_detected",
        completed_tasks=sorted(task.get_name() for task in done),
        orchestrator_done=orchestrator_task in done,
        api_done=api_task in done if api_task is not None else False,
        shutdown_requested_done=shutdown_wait_task in done,
        shutdown_requested=shutdown_requested.is_set(),
    )
    await _consume_completed_runtime_tasks(
        done,
        orchestrator_task=orchestrator_task,
        shutdown_wait_task=shutdown_wait_task,
        api_task=api_task,
        shutdown_requested=shutdown_requested,
        api_server=api_server,
    )

    if shutdown_wait_task in done:
        logger.info("application_shutdown_requested")
        if api_task is not None and api_task not in done:
            await _await_api_task_graceful_shutdown(
                api_task,
                orchestrator_task=orchestrator_task,
                shutdown_wait_task=shutdown_wait_task,
                shutdown_requested=shutdown_requested,
                api_server=api_server,
            )


async def _consume_completed_runtime_tasks(
    done: set[asyncio.Task],
    *,
    orchestrator_task: asyncio.Task[None],
    shutdown_wait_task: asyncio.Task[bool],
    api_task: asyncio.Task[None] | None,
    shutdown_requested: asyncio.Event,
    api_server: _EmbeddedApiServerContext,
) -> None:
    """Consume completed runtime tasks and raise the first non-cancellation failure."""
    failures: list[Exception] = []
    for task in (orchestrator_task, api_task, shutdown_wait_task):
        if task is None or task not in done:
            continue
        try:
            task_name = task.get_name()
            if api_task is not None and task is api_task:
                await _await_api_task_completion(
                    api_task,
                    shutdown_requested=shutdown_requested,
                    api_server=api_server,
                )
                logger.info(
                    "runtime_task_completed",
                    task_name=task_name,
                    shutdown_requested=shutdown_requested.is_set(),
                )
                continue
            await task
            if task is orchestrator_task:
                if shutdown_requested.is_set():
                    logger.info(
                        "orchestrator_task_completed_after_shutdown_request",
                        task_name=task_name,
                    )
                    continue
                _raise_orchestrator_exit(
                    reason="Orchestrator task finished while application shutdown was not requested",
                )
            logger.info(
                "runtime_task_completed",
                task_name=task_name,
                shutdown_requested=shutdown_requested.is_set(),
            )
            continue
        except asyncio.CancelledError:
            logger.info("runtime_task_cancelled", task_name=task.get_name())
            continue
        except Exception as exc:
            logger.exception("runtime_task_failed", task_name=task.get_name())
            failures.append(exc)
    if failures:
        raise failures[0]


async def _await_api_task_completion(
    api_task: asyncio.Task[None],
    *,
    shutdown_requested: asyncio.Event,
    api_server: _EmbeddedApiServerContext,
) -> None:
    """Await the API task and classify normal shutdown versus unexpected exit."""
    await api_task
    if shutdown_requested.is_set():
        logger.info("embedded_api_server_requested_application_shutdown", **api_server.log_context())
        return
    _raise_embedded_api_server_exit(
        api_server,
        reason="API task finished while application shutdown was not requested",
    )


async def _await_api_task_graceful_shutdown(
    api_task: asyncio.Task[None],
    *,
    orchestrator_task: asyncio.Task[None],
    shutdown_wait_task: asyncio.Task[bool],
    shutdown_requested: asyncio.Event,
    api_server: _EmbeddedApiServerContext,
) -> None:
    """Give Uvicorn a bounded window to run FastAPI lifespan shutdown."""
    grace_deadline = time.monotonic() + _EMBEDDED_API_SHUTDOWN_GRACE_SECONDS
    monitored_tasks: set[asyncio.Task] = {api_task, orchestrator_task}
    while api_task in monitored_tasks:
        timeout_seconds = grace_deadline - time.monotonic()
        if timeout_seconds <= 0:
            break
        done, _pending = await asyncio.wait(
            monitored_tasks,
            timeout=timeout_seconds,
            return_when=asyncio.FIRST_COMPLETED,
        )
        if not done:
            break
        await _consume_completed_runtime_tasks(
            done,
            orchestrator_task=orchestrator_task,
            shutdown_wait_task=shutdown_wait_task,
            api_task=api_task,
            shutdown_requested=shutdown_requested,
            api_server=api_server,
        )
        if api_task in done:
            return
        monitored_tasks.difference_update(done)

    logger.warning(
        "embedded_api_server_shutdown_timeout",
        **api_server.log_context(),
        timeout_seconds=_EMBEDDED_API_SHUTDOWN_GRACE_SECONDS,
    )


async def _cancel_task_if_pending(task: asyncio.Task | None) -> None:
    """Cancel and await one task only when it is still pending."""
    if task is None or task.done():
        return
    task.cancel()
    with suppress(asyncio.CancelledError):
        await task


async def _finish_runtime_shutdown(
    *,
    shutdown_wait_task: asyncio.Task[bool] | None,
    api_task: asyncio.Task[None] | None,
    orchestrator_task: asyncio.Task[None] | None,
    auxiliary_tasks: list[asyncio.Task],
    orchestrator: _MultiAgentOrchestrator | None,
    stall_detector: EventLoopStallDetector | None,
) -> None:
    """Finish one runtime cleanup sequence without duplicating partial teardown."""
    await _cancel_task_if_pending(shutdown_wait_task)
    await _cancel_task_if_pending(api_task)
    await _cancel_task_if_pending(orchestrator_task)
    for task in auxiliary_tasks:
        task.cancel()
    for task in auxiliary_tasks:
        with suppress(asyncio.CancelledError):
            await task
    try:
        if orchestrator is not None:
            await orchestrator.stop()
    finally:
        if stall_detector is not None:
            stall_detector.stop()
        reset_matrix_sync_health()
        reset_runtime_state()
        shutdown_primary_worker_manager()


async def _wait_for_runtime_shutdown_cleanup(
    cleanup_task: asyncio.Task[None],
    *,
    shutdown_was_requested: bool,
) -> None:
    """Wait for cleanup while preserving caller cancellation semantics."""
    while True:
        try:
            await asyncio.shield(cleanup_task)
        except asyncio.CancelledError:
            if cleanup_task.done():
                await cleanup_task
                return
            if not shutdown_was_requested:
                cleanup_task.cancel()
                try:
                    await cleanup_task
                except asyncio.CancelledError:
                    pass
                except Exception:
                    logger.exception("Runtime cleanup failed while propagating caller cancellation")
                raise
            # Once orderly shutdown starts, later cancellation is duplicate
            # shutdown pressure; cleanup must finish before main returns.
            logger.info("Ignoring repeated signal cancellation during runtime cleanup")
        else:
            return


async def main(
    log_level: str,
    runtime_paths: RuntimePaths,
    *,
    api: bool = True,
    api_port: int = 8765,
    api_host: str = "0.0.0.0",  # noqa: S104
) -> None:
    """Main entry point for the multi-agent bot system."""
    storage_path = runtime_paths.storage_root
    orchestrator: _MultiAgentOrchestrator | None = None
    auxiliary_tasks: list[asyncio.Task] = []
    shutdown_requested = asyncio.Event()
    api_server = _EmbeddedApiServerContext(host=api_host, port=api_port)
    orchestrator_task: asyncio.Task[None] | None = None
    shutdown_wait_task: asyncio.Task[bool] | None = None
    api_task: asyncio.Task[None] | None = None
    stall_detector: EventLoopStallDetector | None = None

    try:
        # Drop any stale worker manager before startup work builds the active runtime.
        shutdown_primary_worker_manager(timeout_seconds=0.0)

        # Configure logging before any background tasks or account setup begin.
        setup_logging(level=log_level, runtime_paths=runtime_paths)

        stall_detector = start_event_loop_stall_detector(runtime_paths)

        logger.info("Syncing API keys from environment to CredentialsManager...")
        sync_env_to_credentials(runtime_paths=runtime_paths)

        # Ensure storage exists before any runtime components try to write into it.
        storage_path.mkdir(parents=True, exist_ok=True)

        logger.info("Starting orchestrator...")
        orchestrator = _MultiAgentOrchestrator(runtime_paths=runtime_paths, api_enabled=api)
        set_runtime_starting()
        auxiliary_specs = [
            (
                "config watcher",
                lambda: _watch_config_task(orchestrator.config_path, orchestrator),
                "config_watcher_supervisor",
            ),
            ("plugins watcher", lambda: watch_plugins_task(orchestrator), "plugins_watcher_supervisor"),
            ("skills watcher", lambda: _watch_skills_task(orchestrator), "skills_watcher_supervisor"),
        ]

        for task_name, operation, supervisor_name in auxiliary_specs:
            auxiliary_tasks.append(
                asyncio.create_task(
                    _run_auxiliary_task_forever(
                        task_name,
                        operation,
                        should_restart=lambda: not shutdown_requested.is_set(),
                    ),
                    name=supervisor_name,
                ),
            )

        if api:
            api_task = asyncio.create_task(
                _run_api_server(
                    api_host,
                    api_port,
                    log_level,
                    runtime_paths,
                    orchestrator.knowledge_refresh_scheduler,
                    shutdown_requested,
                ),
                name="api_server",
            )

        orchestrator_task = asyncio.create_task(orchestrator.start(), name="orchestrator")
        shutdown_wait_task = asyncio.create_task(shutdown_requested.wait(), name="application_shutdown_wait")
        await _wait_for_runtime_completion(
            orchestrator_task=orchestrator_task,
            shutdown_wait_task=shutdown_wait_task,
            api_task=api_task,
            shutdown_requested=shutdown_requested,
            api_server=api_server,
        )

    except asyncio.CancelledError:
        if not shutdown_requested.is_set():
            raise
        # After an explicit shutdown request, cancellation is shutdown pressure
        # rather than caller cancellation and must not bypass orderly cleanup.
        logger.info("Ignoring duplicate signal cancellation during requested shutdown")
    except KeyboardInterrupt:
        shutdown_requested.set()
        logger.info("Multi-agent bot system stopped by user")
    except PermanentStartupError as exc:
        shutdown_requested.set()
        logger.error("MindRoom startup failed", error=str(exc))  # noqa: TRY400
        raise
    except Exception:
        shutdown_requested.set()
        logger.exception("Error in MindRoom runtime")
        raise
    finally:
        shutdown_was_requested = shutdown_requested.is_set()
        shutdown_requested.set()
        cleanup_task = asyncio.create_task(
            _finish_runtime_shutdown(
                shutdown_wait_task=shutdown_wait_task,
                api_task=api_task,
                orchestrator_task=orchestrator_task,
                auxiliary_tasks=auxiliary_tasks,
                orchestrator=orchestrator,
                stall_detector=stall_detector,
            ),
            name="runtime_shutdown",
        )
        await _wait_for_runtime_shutdown_cleanup(
            cleanup_task,
            shutdown_was_requested=shutdown_was_requested,
        )
