"""Direct unit tests for the debounced config-reload lifecycle."""

from __future__ import annotations

import asyncio
import threading
from typing import TYPE_CHECKING
from unittest.mock import AsyncMock, MagicMock

import pytest

import mindroom.orchestration.config_lifecycle as lifecycle_module
from mindroom.config.agent import AgentConfig
from mindroom.config.main import Config
from mindroom.orchestration.config_lifecycle import ConfigReloadLifecycle, _ReplacementDrainState
from mindroom.orchestration.config_updates import ConfigUpdatePlan
from mindroom.orchestration.runtime import create_logged_task
from mindroom.response_admission import ResponseAdmissionGate
from tests.conftest import bind_runtime_paths, test_runtime_paths

if TYPE_CHECKING:
    from collections.abc import Mapping
    from pathlib import Path

    from mindroom.bot import AgentBot, TeamBot


def _make_lifecycle(
    tmp_path: Path,
    *,
    running: bool = True,
    current_config: Config | None = None,
    agent_bots: Mapping[str, AgentBot | TeamBot] | None = None,
    response_admission_gate: ResponseAdmissionGate | None = None,
) -> ConfigReloadLifecycle:
    """Return a lifecycle wired to stub dependencies."""
    gate = response_admission_gate or ResponseAdmissionGate()
    return ConfigReloadLifecycle(
        runtime_paths=test_runtime_paths(tmp_path),
        is_running=lambda: running,
        current_config=lambda: current_config,
        agent_bots=lambda: agent_bots if agent_bots is not None else {},
        load_initial_config=AsyncMock(return_value=False),
        apply_update_plan=AsyncMock(return_value=True),
        response_admission_gate=gate,
    )


def test_replacement_drain_state_tracks_wait_warning_and_force() -> None:
    """Replacement-drain helpers should model wait, warning, and force transitions."""
    state = _ReplacementDrainState()

    assert state.waiting_for_idle is False

    state.begin_wait(now=10.0)

    assert state.waiting_for_idle is True
    assert (
        state.should_warn(
            now=10.5,
            warning_after_seconds=1.0,
            warning_interval_seconds=10.0,
        )
        is False
    )
    assert (
        state.should_warn(
            now=11.0,
            warning_after_seconds=1.0,
            warning_interval_seconds=10.0,
        )
        is True
    )

    state.mark_warning(11.0)

    assert (
        state.should_warn(
            now=15.0,
            warning_after_seconds=1.0,
            warning_interval_seconds=10.0,
        )
        is False
    )
    assert (
        state.should_warn(
            now=21.0,
            warning_after_seconds=1.0,
            warning_interval_seconds=10.0,
        )
        is True
    )
    assert state.should_force_apply(now=11.9, force_after_seconds=2.0) is False
    assert state.should_force_apply(now=12.0, force_after_seconds=2.0) is True


@pytest.mark.asyncio
async def test_request_reload_is_ignored_until_running(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Reload requests before startup finishes should be dropped."""
    logger_mock = MagicMock()
    monkeypatch.setattr("mindroom.orchestration.config_lifecycle.logger", logger_mock)
    lifecycle = _make_lifecycle(tmp_path, running=False)

    lifecycle.request_reload()

    assert lifecycle._requested_at is None
    assert lifecycle._reload_task is None
    assert any(
        call.args and call.args[0] == "Ignoring config change while startup is still in progress"
        for call in logger_mock.info.call_args_list
    )


@pytest.mark.asyncio
async def test_rapid_requests_coalesce_into_one_reload(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Multiple quick reload requests should extend the debounce and apply once."""
    monkeypatch.setattr("mindroom.orchestration.config_lifecycle._CONFIG_RELOAD_DEBOUNCE_SECONDS", 0.05)
    monkeypatch.setattr("mindroom.orchestration.config_lifecycle._REPLACEMENT_DRAIN_IDLE_POLL_SECONDS", 0.01)
    lifecycle = _make_lifecycle(tmp_path)
    lifecycle._update_config = AsyncMock(return_value=True)

    lifecycle.request_reload()
    task = lifecycle._reload_task
    assert task is not None

    await asyncio.sleep(0.02)
    lifecycle.request_reload()

    assert lifecycle._reload_task is task
    await asyncio.wait_for(task, timeout=1)
    lifecycle._update_config.assert_awaited_once()


@pytest.mark.asyncio
async def test_reload_drains_active_responses_before_applying(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """A queued reload should wait until in-flight responses finish."""
    monkeypatch.setattr("mindroom.orchestration.config_lifecycle._CONFIG_RELOAD_DEBOUNCE_SECONDS", 0.01)
    monkeypatch.setattr("mindroom.orchestration.config_lifecycle._REPLACEMENT_DRAIN_IDLE_POLL_SECONDS", 0.01)
    gate = ResponseAdmissionGate()
    assert gate.admit()
    lifecycle = _make_lifecycle(tmp_path, response_admission_gate=gate)
    lifecycle._update_config = AsyncMock(return_value=True)

    lifecycle.request_reload()
    task = lifecycle._reload_task
    assert task is not None

    await asyncio.sleep(0.05)
    lifecycle._update_config.assert_not_awaited()

    gate.release()
    await asyncio.wait_for(task, timeout=1)
    lifecycle._update_config.assert_awaited_once()
    assert gate.closed is False


@pytest.mark.asyncio
async def test_replacement_admission_serializes_config_and_mcp_owners(tmp_path: Path) -> None:
    """Concurrent replacement flows must never share or prematurely reopen gate ownership."""
    gate = ResponseAdmissionGate()
    lifecycle = _make_lifecycle(tmp_path, response_admission_gate=gate)
    mcp_started = asyncio.Event()
    config_started = asyncio.Event()

    async def apply_mcp_restart() -> None:
        mcp_started.set()
        await asyncio.Future()

    async def apply_config_reload() -> None:
        config_started.set()
        await asyncio.Future()

    mcp_task = asyncio.create_task(
        lifecycle.apply_with_response_admission(
            apply_mcp_restart,
            operation_name="MCP catalog restart",
            request_is_current=lambda: True,
        ),
    )
    await mcp_started.wait()
    config_task = asyncio.create_task(
        lifecycle.apply_with_response_admission(
            apply_config_reload,
            operation_name="configuration reload",
            request_is_current=lambda: True,
        ),
    )
    await asyncio.sleep(0)
    assert gate.closed
    assert not config_started.is_set()
    mcp_task.cancel()
    await asyncio.gather(mcp_task, return_exceptions=True)
    await asyncio.wait_for(config_started.wait(), timeout=1)
    assert gate.closed
    config_task.cancel()
    await asyncio.gather(config_task, return_exceptions=True)
    assert gate.closed is False


@pytest.mark.asyncio
async def test_stuck_drain_warns_then_stops_deferring(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """A wedged drain should warn, keep waiting, and only stop deferring at the bound."""
    warning_after_seconds = 0.5
    force_after_seconds = 1_000.0
    wait_started_at = 10.0
    monkeypatch.setattr("mindroom.orchestration.config_lifecycle._REPLACEMENT_DRAIN_IDLE_POLL_SECONDS", 0)
    monkeypatch.setattr(
        "mindroom.orchestration.config_lifecycle._REPLACEMENT_DRAIN_WARNING_AFTER_SECONDS",
        warning_after_seconds,
    )
    monkeypatch.setattr(
        "mindroom.orchestration.config_lifecycle._REPLACEMENT_DRAIN_WARNING_INTERVAL_SECONDS",
        1.0,
    )
    monkeypatch.setattr(
        "mindroom.orchestration.config_lifecycle._REPLACEMENT_DRAIN_FORCE_AFTER_SECONDS",
        force_after_seconds,
    )
    logger_mock = MagicMock()
    monkeypatch.setattr("mindroom.orchestration.config_lifecycle.logger", logger_mock)
    lifecycle = _make_lifecycle(tmp_path)
    drain_state = _ReplacementDrainState()
    loop = MagicMock(spec=asyncio.AbstractEventLoop)
    loop.time.side_effect = [wait_started_at, wait_started_at + 1.0, wait_started_at + force_after_seconds]

    should_defer = await lifecycle._should_defer_replacement_for_active_responses(
        drain_state=drain_state,
        active_response_count=1,
        loop=loop,
        operation_name="configuration reload",
    )
    assert should_defer is True
    logger_mock.info.assert_any_call(
        "Deferring replacement until active responses finish",
        operation="configuration reload",
        active_response_count=1,
    )

    # Past the warning threshold but still inside the bound: warn and keep waiting.
    should_defer = await lifecycle._should_defer_replacement_for_active_responses(
        drain_state=drain_state,
        active_response_count=1,
        loop=loop,
        operation_name="configuration reload",
    )
    assert should_defer is True
    assert any(
        call.args
        and call.args[0] == "Replacement still waiting for active responses to finish"
        and call.kwargs["operation"] == "configuration reload"
        for call in logger_mock.warning.call_args_list
    )
    logger_mock.error.assert_not_called()

    # At the bound: stop deferring so the change cannot be starved forever.
    should_defer = await lifecycle._should_defer_replacement_for_active_responses(
        drain_state=drain_state,
        active_response_count=1,
        loop=loop,
        operation_name="configuration reload",
    )
    assert should_defer is False
    assert any(
        call.args
        and call.args[0] == "Applying replacement while responses are still active"
        and call.kwargs["operation"] == "configuration reload"
        for call in logger_mock.error.call_args_list
    )


@pytest.mark.asyncio
async def test_new_request_during_drain_keeps_waiting_for_idle(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """A newer config change should not make an active response reload early."""
    monkeypatch.setattr("mindroom.orchestration.config_lifecycle._CONFIG_RELOAD_DEBOUNCE_SECONDS", 0.01)
    monkeypatch.setattr("mindroom.orchestration.config_lifecycle._REPLACEMENT_DRAIN_IDLE_POLL_SECONDS", 0.005)
    gate = ResponseAdmissionGate()
    assert gate.admit()
    lifecycle = _make_lifecycle(tmp_path, response_admission_gate=gate)

    lifecycle._update_config = AsyncMock(return_value=True)

    lifecycle.request_reload()
    await asyncio.sleep(0.06)
    lifecycle.request_reload()
    await asyncio.sleep(0.06)

    task = lifecycle._reload_task
    assert task is not None
    lifecycle._update_config.assert_not_awaited()

    gate.release()
    await asyncio.wait_for(task, timeout=1)

    lifecycle._update_config.assert_awaited_once()


@pytest.mark.asyncio
async def test_failed_update_does_not_strand_queued_reload(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """A failed update must not prevent a subsequently queued reload from running."""
    monkeypatch.setattr("mindroom.orchestration.config_lifecycle._CONFIG_RELOAD_DEBOUNCE_SECONDS", 0.01)
    monkeypatch.setattr("mindroom.orchestration.config_lifecycle._REPLACEMENT_DRAIN_IDLE_POLL_SECONDS", 0.01)
    lifecycle = _make_lifecycle(tmp_path)

    call_count = 0

    async def failing_then_succeeding_update() -> bool:
        nonlocal call_count
        call_count += 1
        if call_count == 1:
            # First call fails; queue a new reload during the failure
            lifecycle.request_reload()
            msg = "Simulated config update failure"
            raise RuntimeError(msg)
        return True

    lifecycle._update_config = AsyncMock(side_effect=failing_then_succeeding_update)
    lifecycle.request_reload()
    task = lifecycle._reload_task
    assert task is not None

    await asyncio.wait_for(task, timeout=2)

    assert lifecycle._update_config.await_count == 2


@pytest.mark.asyncio
async def test_config_change_during_update_triggers_second_reload(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """A config change arriving while an update runs should cause a second reload."""
    monkeypatch.setattr("mindroom.orchestration.config_lifecycle._CONFIG_RELOAD_DEBOUNCE_SECONDS", 0.01)
    monkeypatch.setattr("mindroom.orchestration.config_lifecycle._REPLACEMENT_DRAIN_IDLE_POLL_SECONDS", 0.01)
    lifecycle = _make_lifecycle(tmp_path)

    call_count = 0

    async def update_config_with_second_change() -> bool:
        nonlocal call_count
        call_count += 1
        if call_count == 1:
            lifecycle.request_reload()
        return True

    lifecycle._update_config = AsyncMock(side_effect=update_config_with_second_change)
    lifecycle.request_reload()
    task = lifecycle._reload_task
    assert task is not None

    await asyncio.wait_for(task, timeout=2)

    assert lifecycle._update_config.await_count == 2


@pytest.mark.asyncio
async def test_cancel_logs_exception_instead_of_suppressing_silently(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Reload-task cancellation should log unexpected failures and keep shutdown moving."""
    logger_mock = MagicMock()
    monkeypatch.setattr("mindroom.orchestration.runtime.logger", logger_mock)
    lifecycle = _make_lifecycle(tmp_path)
    started = asyncio.Event()

    async def fail_during_cancel() -> None:
        started.set()
        try:
            await asyncio.Future()
        except asyncio.CancelledError as err:
            msg = "boom"
            raise RuntimeError(msg) from err

    lifecycle._reload_task = create_logged_task(
        fail_during_cancel(),
        name="config_reload",
        failure_message="config_reload failed",
    )
    await asyncio.wait_for(started.wait(), timeout=1)

    await lifecycle.cancel()

    assert lifecycle._reload_task is None
    assert any(
        call.args
        and call.args[0] == "Detached task failed while being cancelled"
        and call.kwargs.get("task_name") == "config_reload"
        for call in logger_mock.debug.call_args_list
    )
    logger_mock.exception.assert_not_called()


@pytest.mark.asyncio
async def test_cancel_clears_queued_reload(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Cancelling should stop the queued reload before it applies."""
    monkeypatch.setattr("mindroom.orchestration.config_lifecycle._CONFIG_RELOAD_DEBOUNCE_SECONDS", 0.01)
    monkeypatch.setattr("mindroom.orchestration.config_lifecycle._REPLACEMENT_DRAIN_IDLE_POLL_SECONDS", 0.01)
    busy_gate = ResponseAdmissionGate()
    assert busy_gate.admit()
    lifecycle = _make_lifecycle(tmp_path, response_admission_gate=busy_gate)
    lifecycle._update_config = AsyncMock(return_value=True)

    lifecycle.request_reload()
    task = lifecycle._reload_task
    assert task is not None
    await asyncio.sleep(0.05)

    await lifecycle.cancel()

    assert lifecycle._reload_task is None
    assert lifecycle._requested_at is None
    assert task.done()
    lifecycle._update_config.assert_not_awaited()


@pytest.mark.asyncio
async def test_response_start_during_config_load_waits_until_apply_finishes(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """A response racing blocked config loading must be refused until apply finishes."""
    monkeypatch.setattr("mindroom.orchestration.config_lifecycle._CONFIG_RELOAD_DEBOUNCE_SECONDS", 0)
    load_started = threading.Event()
    release_load = threading.Event()
    observed_apply_counts: list[int] = []
    gate = ResponseAdmissionGate()
    current_config = Config()
    new_config = Config()

    def blocked_load(*_args: object, **_kwargs: object) -> Config:
        load_started.set()
        assert release_load.wait(timeout=2)
        return new_config

    monkeypatch.setattr("mindroom.orchestration.config_lifecycle.load_config", blocked_load)
    lifecycle = _make_lifecycle(
        tmp_path,
        current_config=current_config,
        response_admission_gate=gate,
    )

    async def apply_plan(*_args: object) -> bool:
        observed_apply_counts.append(gate.in_flight_response_count)
        return True

    lifecycle.apply_update_plan = AsyncMock(side_effect=apply_plan)

    lifecycle.request_reload()
    reload_task = lifecycle._reload_task
    assert reload_task is not None
    assert await asyncio.to_thread(load_started.wait, 1)

    try:
        # Admission is already closed while the apply is in progress, and asking
        # never blocks on the applier, so the response is refused immediately.
        assert gate.admit() is False

        release_load.set()
        await asyncio.wait_for(reload_task, timeout=1)

        lifecycle.apply_update_plan.assert_awaited_once()
        assert observed_apply_counts == [0]
        # The gate reopens once the apply completes.
        assert gate.admit() is True
        gate.release()
    finally:
        release_load.set()
        await asyncio.gather(reload_task, return_exceptions=True)


@pytest.mark.asyncio
async def test_apply_does_not_block_response_drain_started_by_the_apply(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Applying a plan must not stall the response drain that stopping bots performs.

    Stopping a bot drains its detached responses. If the applier held the
    admission gate, a response parked at admission could not finish, so the
    drain would burn its whole timeout and then cancel live work.
    """
    monkeypatch.setattr("mindroom.orchestration.config_lifecycle._CONFIG_RELOAD_DEBOUNCE_SECONDS", 0)
    gate = ResponseAdmissionGate()
    lifecycle = _make_lifecycle(tmp_path, current_config=Config(), response_admission_gate=gate)
    monkeypatch.setattr("mindroom.orchestration.config_lifecycle.load_config", lambda *_a, **_k: Config())

    async def response_lifecycle() -> str:
        if not gate.admit():
            return "refused"
        gate.release()
        return "admitted"

    async def apply_plan(*_args: object) -> bool:
        # Stand in for stop_entities -> prepare_for_sync_shutdown -> drain_inbox_responses.
        task = asyncio.create_task(response_lifecycle())
        _done, pending = await asyncio.wait([task], timeout=1)
        assert not pending, "drain stalled: response could not settle during apply"
        assert task.result() == "refused"
        return True

    lifecycle.apply_update_plan = AsyncMock(side_effect=apply_plan)

    lifecycle.request_reload()
    task = lifecycle._reload_task
    assert task is not None
    await asyncio.wait_for(task, timeout=2)

    lifecycle.apply_update_plan.assert_awaited_once()
    assert gate.closed is False


@pytest.mark.asyncio
async def test_drain_applies_reload_after_force_timeout(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """A never-idle install must still get its config change applied eventually."""
    monkeypatch.setattr("mindroom.orchestration.config_lifecycle._CONFIG_RELOAD_DEBOUNCE_SECONDS", 0)
    monkeypatch.setattr("mindroom.orchestration.config_lifecycle._REPLACEMENT_DRAIN_IDLE_POLL_SECONDS", 0)
    monkeypatch.setattr("mindroom.orchestration.config_lifecycle._REPLACEMENT_DRAIN_FORCE_AFTER_SECONDS", 0.05)
    gate = ResponseAdmissionGate()
    # A response that never finishes, so the gate is never idle.
    assert gate.admit()
    lifecycle = _make_lifecycle(tmp_path, response_admission_gate=gate)
    lifecycle._update_config = AsyncMock(return_value=True)

    lifecycle.request_reload()
    task = lifecycle._reload_task
    assert task is not None
    await asyncio.wait_for(task, timeout=2)

    lifecycle._update_config.assert_awaited_once()
    # Admission reopens even though the forced apply ran over a live response.
    assert gate.closed is False


@pytest.mark.asyncio
async def test_update_config_delegates_initial_load(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Without an active config, update_config should hand off to the initial loader."""
    new_config = Config()
    monkeypatch.setattr(
        "mindroom.orchestration.config_lifecycle.load_config",
        lambda *_args, **_kwargs: new_config,
    )
    lifecycle = _make_lifecycle(tmp_path, current_config=None)

    assert await lifecycle._update_config() is False

    lifecycle.load_initial_config.assert_awaited_once_with(new_config)
    lifecycle.apply_update_plan.assert_not_awaited()


@pytest.mark.asyncio
async def test_update_config_builds_plan_and_dispatches(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """With an active config, update_config should diff configs and dispatch the plan."""
    runtime_paths = test_runtime_paths(tmp_path)
    current_config = bind_runtime_paths(Config(agents={"agent1": AgentConfig(display_name="Agent 1")}), runtime_paths)
    new_config = bind_runtime_paths(
        Config(agents={"agent1": AgentConfig(display_name="Agent 1", role="changed role")}),
        runtime_paths,
    )
    monkeypatch.setattr(
        "mindroom.orchestration.config_lifecycle.load_config",
        lambda *_args, **_kwargs: new_config,
    )
    lifecycle = _make_lifecycle(
        tmp_path,
        current_config=current_config,
        agent_bots={"router": MagicMock(), "agent1": MagicMock()},
    )

    assert await lifecycle._update_config() is True

    lifecycle.load_initial_config.assert_not_awaited()
    lifecycle.apply_update_plan.assert_awaited_once()
    dispatched_config, plan, plugin_changes = lifecycle.apply_update_plan.await_args.args
    assert dispatched_config is current_config
    assert plan.new_config is new_config
    assert "agent1" in plan.entities_to_restart
    assert plugin_changes == ()


@pytest.mark.asyncio
async def test_update_config_plugin_changes_restart_all_bots(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Plugin entry changes should expand the plan to restart every managed bot."""
    current_config = Config()
    new_config = Config()
    plan = ConfigUpdatePlan(
        new_config=new_config,
        changed_mcp_servers=set(),
        configured_entities={"router"},
        entities_to_restart={"agent1"},
        new_entities=set(),
        removed_entities=set(),
        mindroom_user_changed=False,
        matrix_room_access_changed=False,
        matrix_space_changed=False,
        authorization_changed=False,
    )
    monkeypatch.setattr(
        "mindroom.orchestration.config_lifecycle.load_config",
        lambda *_args, **_kwargs: new_config,
    )
    monkeypatch.setattr(
        "mindroom.orchestration.config_lifecycle.plugin_change_paths",
        lambda *_args: ("plugins/demo",),
    )
    monkeypatch.setattr(
        "mindroom.orchestration.config_lifecycle.build_config_update_plan",
        lambda **_kwargs: plan,
    )
    lifecycle = _make_lifecycle(
        tmp_path,
        current_config=current_config,
        agent_bots={"router": MagicMock(), "agent1": MagicMock(), "agent2": MagicMock()},
    )

    assert await lifecycle._update_config() is True

    _, dispatched_plan, plugin_changes = lifecycle.apply_update_plan.await_args.args
    assert dispatched_plan.entities_to_restart == {"router", "agent1", "agent2"}
    assert plugin_changes == ("plugins/demo",)


@pytest.mark.asyncio
async def test_update_config_loads_config_off_event_loop(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """update_config must keep slow config loading off the loop so heartbeats keep ticking."""
    gate = threading.Event()
    load_started = threading.Event()
    loaded_config = Config(agents={"general": AgentConfig(display_name="General")})

    def slow_load_config(*_args: object, **_kwargs: object) -> Config:
        load_started.set()
        gate.wait()
        return loaded_config

    monkeypatch.setattr(lifecycle_module, "load_config", slow_load_config)
    lifecycle = _make_lifecycle(tmp_path)
    lifecycle.load_initial_config = AsyncMock(return_value=True)

    update_task = asyncio.get_running_loop().create_task(lifecycle._update_config())
    await asyncio.to_thread(load_started.wait, 5.0)

    # The loader thread is parked on the gate; the loop must stay live.
    heartbeats = 0
    while heartbeats < 50:
        await asyncio.sleep(0)
        heartbeats += 1
    assert not update_task.done()

    gate.set()
    assert await update_task is True
    lifecycle.load_initial_config.assert_awaited_once_with(loaded_config)


@pytest.mark.asyncio
async def test_update_config_waits_for_lock_before_loading(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Config loading must not start until the shared update lock is available."""
    lock = asyncio.Lock()
    config = Config()
    to_thread_started = asyncio.Event()
    real_to_thread = asyncio.to_thread

    async def observed_to_thread(*args: object, **kwargs: object) -> object:
        to_thread_started.set()
        return await real_to_thread(*args, **kwargs)

    monkeypatch.setattr(lifecycle_module.asyncio, "to_thread", observed_to_thread)
    load_config_mock = MagicMock(return_value=config)
    monkeypatch.setattr(lifecycle_module, "load_config", load_config_mock)
    lifecycle = _make_lifecycle(tmp_path, current_config=config)
    lifecycle.config_update_lock = lock

    await lock.acquire()
    update_task = asyncio.create_task(lifecycle._update_config())
    await asyncio.sleep(0)
    assert not to_thread_started.is_set()

    lock.release()
    await asyncio.wait_for(to_thread_started.wait(), timeout=1.0)
    assert await update_task is True
    load_config_mock.assert_called_once()
    lifecycle.apply_update_plan.assert_awaited_once()
