"""Regression tests for queued-message mid-turn notifications."""

from __future__ import annotations

import asyncio
import threading
from contextlib import contextmanager
from dataclasses import replace
from types import SimpleNamespace
from typing import TYPE_CHECKING, Literal, Protocol, Self, cast
from unittest.mock import AsyncMock, MagicMock, patch

import nio
import pytest
from agno.agent import Agent as AgnoAgent
from agno.db.base import BaseDb, SessionType
from agno.media import Image
from agno.models.message import Message
from agno.run.agent import RunCompletedEvent, RunContentEvent, RunOutput
from agno.run.base import RunStatus
from agno.run.team import TeamRunOutput
from agno.session.agent import AgentSession
from agno.session.team import TeamSession
from structlog.testing import capture_logs

from mindroom import ai_runtime, turn_controller
from mindroom.agent_storage import create_state_storage, get_agent_session
from mindroom.ai import _PreparedAgentRun, ai_response, stream_agent_response
from mindroom.ai_runtime import (
    install_queued_message_notice_hook,
    queued_message_signal_context,
    register_queued_notice_storage,
)
from mindroom.bot import AgentBot
from mindroom.bot_runtime_view import BotRuntimeState
from mindroom.coalescing_batch import (
    CoalescingKey,
    PendingEvent,
    active_follow_up_coalescing_key,
    build_coalesced_batch,
)
from mindroom.config.agent import AgentConfig
from mindroom.config.auth import AuthorizationConfig
from mindroom.config.main import Config
from mindroom.config.models import ModelConfig
from mindroom.constants import prompt_roles_for_history_storage
from mindroom.conversation_resolver import MessageContext
from mindroom.dispatch_handoff import PendingDispatchMetadata, PreparedTextEvent
from mindroom.dispatch_source import (
    ACTIVE_THREAD_FOLLOW_UP_SOURCE_KIND,
    HOOK_DISPATCH_SOURCE_KIND,
    HOOK_SOURCE_KIND,
    MESSAGE_SOURCE_KIND,
    SCHEDULED_SOURCE_KIND,
    TRUSTED_INTERNAL_RELAY_SOURCE_KIND,
    VOICE_SOURCE_KIND,
)
from mindroom.entity_resolution import current_internal_sender_ids
from mindroom.final_delivery import FinalDeliveryOutcome
from mindroom.history.runtime import open_bound_scope_session_context
from mindroom.history.types import HistoryScope
from mindroom.hooks import MessageEnvelope
from mindroom.interactive import InteractiveMetadata
from mindroom.matrix.cache import ThreadHistoryResult
from mindroom.matrix.client import ResolvedVisibleMessage
from mindroom.matrix.users import AgentMatrixUser
from mindroom.message_target import MessageTarget
from mindroom.post_response_effects import (
    PostResponseEffectsDeps,
    PostResponseEffectsSupport,
    ResponseOutcome,
    apply_post_response_effects,
)
from mindroom.prompts import QUEUED_MESSAGE_NOTICE_TEXT
from mindroom.response_lifecycle import ResponseLifecycleCoordinator, _QueuedMessageState
from mindroom.response_payload_preparation import DispatchPayloadInputs, ResponsePayloadPreparation
from mindroom.response_runner import (
    PostLockRequestPreparationError,
    ResponseRequest,
    ResponseRunner,
    _ResponseGenerationOutcome,
)
from mindroom.teams import TeamMode, _create_team_instance
from mindroom.turn_controller import _PrecheckedEvent
from mindroom.turn_policy import PreparedDispatch, ResponseAction, _DispatchPlan
from tests.conftest import (
    TEST_PASSWORD,
    bind_runtime_paths,
    install_runtime_cache_support,
    make_event_cache_mock,
    make_event_cache_write_coordinator_mock,
    make_turn_context,
    message_origin,
    prepared_dispatch_result,
    request_envelope,
    runtime_paths_for,
    test_runtime_paths,
    unwrap_extracted_collaborator,
    wrap_extracted_collaborators,
)
from tests.history_helpers import RecordingModel

if TYPE_CHECKING:
    from collections.abc import AsyncIterator, Awaitable, Callable, Coroutine
    from pathlib import Path

    from mindroom.delivery_gateway import FinalDeliveryRequest
    from mindroom.turn_origin import TurnOrigin


class _ReservationLike(Protocol):
    def cancel(self) -> None:
        """Release the reserved queued-human notice."""


class _NoopResponseLifecycle:
    def __init__(self) -> None:
        self.session_thread_ids: list[str | None] = []

    def setup_session_watch(self, **kwargs: object) -> object:
        thread_id = kwargs["thread_id"]
        assert isinstance(thread_id, str) or thread_id is None
        self.session_thread_ids.append(thread_id)
        return object()

    async def emit_session_started(self, _watch: object) -> None:
        return

    async def finalize(
        self,
        final_delivery_outcome: FinalDeliveryOutcome,
        **_kwargs: object,
    ) -> FinalDeliveryOutcome:
        return final_delivery_outcome


def _config(tmp_path: Path) -> Config:
    return bind_runtime_paths(
        Config(
            agents={"general": AgentConfig(display_name="General", rooms=["!room:localhost"])},
            teams={},
            models={"default": ModelConfig(provider="openai", id="test-model")},
            authorization=AuthorizationConfig(default_room_access=True),
        ),
        test_runtime_paths(tmp_path),
    )


def _bot(tmp_path: Path) -> AgentBot:
    config = _config(tmp_path)
    agent_user = AgentMatrixUser(
        agent_name="general",
        password=TEST_PASSWORD,
        display_name="General",
        user_id="@mindroom_general:localhost",
    )
    bot = AgentBot(agent_user, tmp_path, config, runtime_paths_for(config), rooms=["!room:localhost"])
    bot.client = AsyncMock(spec=nio.AsyncClient)
    bot.client.rooms = {}
    install_runtime_cache_support(bot)
    wrap_extracted_collaborators(bot)
    return bot


def _envelope(
    *,
    source_kind: str = MESSAGE_SOURCE_KIND,
    dispatch_policy_source_kind: str | None = None,
    source_event_id: str = "$event",
    target: MessageTarget | None = None,
    requester_id: str = "@user:localhost",
    sender_id: str = "@user:localhost",
    origin: TurnOrigin | None = None,
) -> MessageEnvelope:
    target = target or MessageTarget.resolve(
        room_id="!room:localhost",
        thread_id=None,
        reply_to_event_id="$event",
    )
    return MessageEnvelope(
        source_event_id=source_event_id,
        target=target,
        body="hello",
        attachment_ids=(),
        mentioned_agents=(),
        agent_name="general",
        dispatch_policy_source_kind=dispatch_policy_source_kind,
        origin=origin or message_origin(sender_id=sender_id, requester_id=requester_id, source_kind=source_kind),
    )


def _prepared_text_event(*, event_id: str = "$event") -> PreparedTextEvent:
    return PreparedTextEvent(
        sender="@user:localhost",
        event_id=event_id,
        body="hello",
        source={"content": {"body": "hello"}},
        server_timestamp=1234,
    )


def _prepared_run(agent: object, *, prompt: str = "prompt") -> _PreparedAgentRun:
    return _PreparedAgentRun(
        agent=agent,
        messages=(Message(role="user", content=prompt),),
        unseen_event_ids=[],
        prepared_history=MagicMock(),
        runtime_model_name="default",
    )


def _message_context() -> MessageContext:
    return MessageContext(
        am_i_mentioned=False,
        is_thread=False,
        thread_id=None,
        thread_history=[],
        mentioned_agents=[],
        has_non_agent_mentions=False,
    )


def _payload_preparation(target: MessageTarget) -> ResponsePayloadPreparation:
    """Build a minimal under-lock preparation carrier bound to ``target``."""
    return ResponsePayloadPreparation(
        dispatch=PreparedDispatch(
            requester_user_id="@user:localhost",
            context=_message_context(),
            target=target,
            correlation_id="$event",
            envelope=_envelope(source_event_id="$event", target=target),
        ),
        prompt="hello",
        action_kind="individual",
        payload_inputs=DispatchPayloadInputs((), (), ()),
        target_member_names=None,
        dispatch_started_at=0.0,
        context_ready_monotonic=0.0,
    )


def _reserved_follow_up_case(
    bot: AgentBot,
    room: nio.MatrixRoom,
    *,
    event_id: str,
    body: str = "hello",
) -> SimpleNamespace:
    target = MessageTarget.resolve(room.room_id, "$thread", event_id)
    envelope = _envelope(
        dispatch_policy_source_kind=ACTIVE_THREAD_FOLLOW_UP_SOURCE_KIND,
        source_event_id=event_id,
        target=target,
    )
    event = PreparedTextEvent(
        sender="@user:localhost",
        event_id=event_id,
        body=body,
        source={"content": {"body": body}},
        server_timestamp=1234,
    )
    dispatch = PreparedDispatch(
        requester_user_id="@user:localhost",
        context=MessageContext(
            am_i_mentioned=False,
            is_thread=True,
            thread_id="$thread",
            thread_history=[],
            mentioned_agents=[],
            has_non_agent_mentions=False,
        ),
        target=target,
        correlation_id=event_id,
        envelope=envelope,
    )
    coordinator = unwrap_extracted_collaborator(bot._response_runner)
    lifecycle = coordinator._lifecycle_coordinator
    queued_signal = lifecycle._get_or_create_queued_signal(target)
    queued_signal.begin_response_turn()
    reservation = lifecycle.reserve_waiting_human_message(target=target, response_envelope=envelope)
    assert reservation is not None
    return SimpleNamespace(
        dispatch=dispatch,
        event=event,
        queued_signal=queued_signal,
        reservation=reservation,
    )


def _queued_notice_metadata(reservation: _ReservationLike) -> tuple[PendingDispatchMetadata, ...]:
    return (
        PendingDispatchMetadata(
            kind="queued_notice_reservation",
            payload=reservation,
            close=reservation.cancel,
            requires_solo_batch=True,
        ),
    )


def _targeted_queued_notice_metadata(
    reservation: _ReservationLike,
    target: MessageTarget,
) -> tuple[PendingDispatchMetadata, ...]:
    return (
        PendingDispatchMetadata(
            kind="queued_notice_reservation",
            payload=reservation,
            close=reservation.cancel,
            target_key=(target.room_id, target.resolved_thread_id),
        ),
    )


def test_response_lifecycle_rejects_mismatched_reservation_target(tmp_path: Path) -> None:
    """Queued notices should use the envelope's canonical lifecycle target."""
    bot = _bot(tmp_path)
    coordinator = unwrap_extracted_collaborator(bot._response_runner)
    lifecycle = coordinator._lifecycle_coordinator
    envelope = _envelope()
    mismatched_target = MessageTarget.resolve("!other:localhost", None, "$event")

    with pytest.raises(ValueError, match=r"MessageEnvelope\.target"):
        lifecycle.reserve_waiting_human_message(
            target=mismatched_target,
            response_envelope=envelope,
        )


@pytest.mark.asyncio
async def test_response_lifecycle_rejects_mismatched_locked_response_target(tmp_path: Path) -> None:
    """Response locking should use the envelope's canonical lifecycle target."""
    bot = _bot(tmp_path)
    coordinator = unwrap_extracted_collaborator(bot._response_runner)
    lifecycle = coordinator._lifecycle_coordinator
    envelope = _envelope()
    mismatched_target = MessageTarget.resolve("!other:localhost", None, "$event")

    async def locked_operation(_target: MessageTarget) -> str:
        return "$response"

    with pytest.raises(ValueError, match=r"MessageEnvelope\.target"):
        await lifecycle.run_locked_response(
            target=mismatched_target,
            response_envelope=envelope,
            queued_notice_reservation=None,
            pipeline_timing=None,
            locked_operation=locked_operation,
        )


_QUEUED_NOTICE_MARKER_KEY = "mindroom_queued_message_notice"
_QUEUED_NOTICE_RESPONSE_TURN_ID_KEY = "mindroom_queued_message_notice_response_turn_id"


def _notice_messages(
    messages: list[Message],
    *,
    marker: bool | str | None = None,
    response_turn_id: str | None = None,
) -> list[Message]:
    matches: list[Message] = []
    for message in messages:
        provider_data = message.provider_data
        if not isinstance(provider_data, dict):
            continue
        notice_marker = provider_data.get(_QUEUED_NOTICE_MARKER_KEY)
        if notice_marker not in (True, "persisted"):
            continue
        if marker is not None and notice_marker != marker:
            continue
        if response_turn_id is not None and provider_data.get(_QUEUED_NOTICE_RESPONSE_TURN_ID_KEY) != response_turn_id:
            continue
        matches.append(message)
    return matches


def _notice_count(
    messages: list[Message],
    *,
    marker: bool | str | None = None,
    response_turn_id: str | None = None,
) -> int:
    return len(
        _notice_messages(
            messages,
            marker=marker,
            response_turn_id=response_turn_id,
        ),
    )


def _queued_notice_message(
    response_turn_id: str,
    *,
    marker: bool | str = True,
    content: str = QUEUED_MESSAGE_NOTICE_TEXT,
    from_history: bool = False,
) -> Message:
    return Message(
        role="user",
        content=content,
        provider_data={
            _QUEUED_NOTICE_MARKER_KEY: marker,
            _QUEUED_NOTICE_RESPONSE_TURN_ID_KEY: response_turn_id,
        },
        from_history=from_history,
    )


def _queued_notice_storage(tmp_path: Path) -> BaseDb:
    return create_state_storage(
        "queued_notice",
        tmp_path,
        subdir="sessions",
        session_table="queued_notice_sessions",
        prompt_roles=prompt_roles_for_history_storage(),
    )


class _FakeStorage:
    def __init__(self) -> None:
        self.session: AgentSession | TeamSession | None = None
        self.upserted = False
        self.upsert_count = 0
        self.closed = False

    def get_session(self, session_id: str, _session_type: object) -> AgentSession | TeamSession | None:
        if self.session is None or self.session.session_id != session_id:
            return None
        return self.session

    def upsert_session(self, session: AgentSession | TeamSession) -> AgentSession | TeamSession:
        self.session = session
        self.upserted = True
        self.upsert_count += 1
        return session

    def close(self) -> None:
        self.closed = True


class _FakeModel:
    def format_function_call_results(
        self,
        messages: list[Message],
        function_call_results: list[Message],
        _compress_tool_results: bool = False,
        **_kwargs: object,
    ) -> None:
        messages.extend(function_call_results)

    def _handle_function_call_media(
        self,
        messages: list[Message],
        function_call_results: list[Message],
        send_media_to_model: bool = True,
    ) -> None:
        if not send_media_to_model:
            return
        if any(message.images or message.videos or message.audio or message.files for message in function_call_results):
            messages.append(Message(role="user", content="Take note of the following content"))


class _FakeModelWithoutFunctionCallMedia:
    def format_function_call_results(
        self,
        messages: list[Message],
        function_call_results: list[Message],
        _compress_tool_results: bool = False,
        **_kwargs: object,
    ) -> None:
        messages.extend(function_call_results)


class _StaticQueuedState:
    def __init__(self, *, pending: bool) -> None:
        self.pending = pending

    def has_pending_human_messages(self) -> bool:
        return self.pending


def test_queued_message_state_tracks_source_event_ids_idempotently() -> None:
    """Queued-message state should track distinct source events, not anonymous increments."""
    state = _QueuedMessageState()

    assert state.add_waiting_human_message("$event")
    assert not state.add_waiting_human_message("$event")

    assert state.has_pending_human_messages()
    assert state.pending_human_messages == 1
    assert state.is_set()

    state.consume_waiting_human_message("$unknown")
    assert state.pending_human_messages == 1

    state.consume_waiting_human_message("$event")
    state.consume_waiting_human_message("$event")

    assert not state.has_pending_human_messages()
    assert state.pending_human_messages == 0
    assert not state.is_set()


def test_active_follow_up_batch_prompt_uses_queued_receive_order() -> None:
    """Target-scoped active follow-up batches should preserve timeline order and senders."""
    room = MagicMock(spec=nio.MatrixRoom)
    room.room_id = "!room:localhost"
    pending_events = [
        PendingEvent(
            event=PreparedTextEvent(
                sender="@alice:localhost",
                event_id="$a1",
                body="A first",
                source={"content": {"body": "A first"}},
                server_timestamp=1,
            ),
            room=room,
            requester_user_id="@alice:localhost",
            source_kind=MESSAGE_SOURCE_KIND,
            dispatch_policy_source_kind=ACTIVE_THREAD_FOLLOW_UP_SOURCE_KIND,
        ),
        PendingEvent(
            event=PreparedTextEvent(
                sender="@bob:localhost",
                event_id="$b1",
                body="B <context> & more",
                source={"content": {"body": "B <context> & more"}},
                server_timestamp=2,
            ),
            room=room,
            requester_user_id="@bob:localhost",
            source_kind=MESSAGE_SOURCE_KIND,
            dispatch_policy_source_kind=ACTIVE_THREAD_FOLLOW_UP_SOURCE_KIND,
        ),
        PendingEvent(
            event=PreparedTextEvent(
                sender="@alice:localhost",
                event_id="$a2",
                body="A follow-up",
                source={"content": {"body": "A follow-up"}},
                server_timestamp=3,
            ),
            room=room,
            requester_user_id="@alice:localhost",
            source_kind=MESSAGE_SOURCE_KIND,
            dispatch_policy_source_kind=ACTIVE_THREAD_FOLLOW_UP_SOURCE_KIND,
        ),
    ]

    batch = build_coalesced_batch(
        active_follow_up_coalescing_key(room.room_id, "$thread"),
        pending_events,
    )

    assert batch.source_event_ids == ["$a1", "$b1", "$a2"]
    assert batch.requester_user_id == "@alice:localhost"
    assert batch.source_event_prompts == {
        "$a1": "A first",
        "$b1": "B <context> & more",
        "$a2": "A follow-up",
    }
    assert batch.prompt == (
        "Messages arrived while the previous response was still running. "
        "They are in chat timeline order. Respond once to the combined context:\n\n"
        "<queued_messages>\n"
        '<msg event_id="$a1" from="@alice:localhost"><![CDATA[A first]]></msg>\n'
        '<msg event_id="$b1" from="@bob:localhost"><![CDATA[B <context> & more]]></msg>\n'
        '<msg event_id="$a2" from="@alice:localhost"><![CDATA[A follow-up]]></msg>\n'
        "</queued_messages>"
    )


def test_same_target_batch_reservation_consumes_all_pending_messages(tmp_path: Path) -> None:
    """A coalesced target batch should consume every matching queued-human notice."""
    bot = _bot(tmp_path)
    target = MessageTarget.resolve("!room:localhost", "$thread", "$a1")
    coordinator = unwrap_extracted_collaborator(bot._response_runner)
    lifecycle = coordinator._lifecycle_coordinator
    queued_signal = lifecycle._get_or_create_queued_signal(target)
    queued_signal.begin_response_turn()
    try:
        first_reservation = lifecycle.reserve_waiting_human_message(
            target=target,
            response_envelope=_envelope(source_event_id="$a1", target=target),
        )
        second_reservation = lifecycle.reserve_waiting_human_message(
            target=target,
            response_envelope=_envelope(source_event_id="$b1", target=target),
        )
        assert first_reservation is not None
        assert second_reservation is not None

        turn_controller._consume_queued_notice_reservations_from_metadata(
            (
                PendingDispatchMetadata(
                    kind="queued_notice_reservation",
                    payload=first_reservation,
                    close=first_reservation.cancel,
                    target_key=(target.room_id, target.resolved_thread_id),
                ),
                PendingDispatchMetadata(
                    kind="queued_notice_reservation",
                    payload=second_reservation,
                    close=second_reservation.cancel,
                    target_key=(target.room_id, target.resolved_thread_id),
                ),
            ),
            target_key=(target.room_id, target.resolved_thread_id),
        )

        assert queued_signal.pending_human_message_event_ids == set()
    finally:
        queued_signal.finish_response_turn()


@contextmanager
def _open_scope(storage: _FakeStorage) -> object:
    yield SimpleNamespace(
        storage=storage,
        storage_factory=lambda: storage,
        session=storage.session,
        scope=HistoryScope("agent", "general"),
    )


class _PrelockBarrierLock:
    def __init__(self) -> None:
        self._locked = False
        self.first_waiting = asyncio.Event()
        self._allow_first_entry = asyncio.Event()
        self._first_entered = asyncio.Event()
        self._released = asyncio.Event()
        self._released.set()

    def locked(self) -> bool:
        return self._locked

    async def acquire(self) -> None:
        if not self.first_waiting.is_set():
            self.first_waiting.set()
            await self._allow_first_entry.wait()
        else:
            await self._first_entered.wait()
        await self._released.wait()
        self._locked = True
        self._released.clear()
        self._first_entered.set()

    def release(self) -> None:
        self._locked = False
        self._released.set()

    async def __aenter__(self) -> Self:
        await self.acquire()
        return self

    async def __aexit__(self, *_args: object) -> None:
        self.release()


@pytest.mark.asyncio
async def test_post_response_effects_skip_thread_summary_for_suppressed_delivery() -> None:
    """Suppressed deliveries must not enqueue a thread summary."""
    queue_thread_summary = MagicMock()

    await apply_post_response_effects(
        FinalDeliveryOutcome(
            terminal_status="cancelled",
            event_id=None,
            suppressed=True,
        ),
        ResponseOutcome(
            interactive_target=MessageTarget.resolve(
                room_id="!room:localhost",
                thread_id="$thread",
                reply_to_event_id="$event",
            ),
            thread_summary_room_id="!room:localhost",
            thread_summary_thread_id="$thread",
            thread_summary_message_count_hint=3,
        ),
        PostResponseEffectsDeps(
            logger=MagicMock(),
            queue_thread_summary=queue_thread_summary,
        ),
    )

    queue_thread_summary.assert_not_called()


@pytest.mark.parametrize(
    ("terminal_status", "run_succeeded"),
    [
        ("error", False),
        ("cancelled", False),
        ("completed", False),
    ],
)
@pytest.mark.asyncio
async def test_post_response_effects_skip_thread_summary_for_unsuccessful_turn(
    terminal_status: Literal["error", "cancelled", "completed"],
    run_succeeded: bool,
) -> None:
    """Visible failed or cancelled turns must not enqueue thread enrichment."""
    should_queue_thread_summary = MagicMock(return_value=True)
    queue_thread_summary = MagicMock()

    await apply_post_response_effects(
        FinalDeliveryOutcome(
            terminal_status=terminal_status,
            event_id="$response",
            is_visible_response=True,
            final_visible_body="Turn did not complete",
        ),
        ResponseOutcome(
            run_succeeded=run_succeeded,
            thread_summary_room_id="!room:localhost",
            thread_summary_thread_id="$thread",
            thread_summary_message_count_hint=2,
        ),
        PostResponseEffectsDeps(
            logger=MagicMock(),
            should_queue_thread_summary=should_queue_thread_summary,
            queue_thread_summary=queue_thread_summary,
        ),
    )

    should_queue_thread_summary.assert_not_called()
    queue_thread_summary.assert_not_called()


@pytest.mark.asyncio
async def test_post_response_effects_skip_memory_persistence_for_failed_run() -> None:
    """Failed runs should not enqueue memory persistence for incomplete content."""
    queue_memory_persistence = MagicMock()

    await apply_post_response_effects(
        FinalDeliveryOutcome(
            terminal_status="error",
            event_id="$response",
            is_visible_response=True,
            final_visible_body="Provider failed",
        ),
        ResponseOutcome(run_succeeded=False),
        PostResponseEffectsDeps(
            logger=MagicMock(),
            queue_memory_persistence=queue_memory_persistence,
        ),
    )

    queue_memory_persistence.assert_not_called()


@pytest.mark.asyncio
async def test_post_response_effects_register_interactive_follow_up_for_preserved_stream_failure() -> None:
    """Preserved visible streamed replies should still register interactive follow-up."""
    register_interactive = AsyncMock()
    target = MessageTarget.resolve(
        room_id="!room:localhost",
        thread_id="$thread",
        reply_to_event_id="$event",
    )
    interactive_metadata = InteractiveMetadata._from_parts(
        {"1": "yes"},
        ({"emoji": "1", "label": "Yes", "value": "yes"},),
    )
    assert interactive_metadata is not None

    await apply_post_response_effects(
        FinalDeliveryOutcome(
            terminal_status="completed",
            event_id="$stream",
            is_visible_response=True,
            final_visible_body="Choose",
            delivery_kind="sent",
            interactive_metadata=interactive_metadata,
        ),
        ResponseOutcome(
            interactive_target=target,
        ),
        PostResponseEffectsDeps(
            logger=MagicMock(),
            register_interactive=register_interactive,
        ),
    )

    register_interactive.assert_awaited_once_with(
        "$stream",
        target,
        interactive_metadata,
    )


@pytest.mark.asyncio
async def test_post_response_effects_skip_interactive_follow_up_for_preserved_stream_error() -> None:
    """Failed preserved stream outcomes must not register interactive follow-up on a failed reply."""
    register_interactive = AsyncMock()
    target = MessageTarget.resolve(
        room_id="!room:localhost",
        thread_id="$thread",
        reply_to_event_id="$event",
    )
    interactive_metadata = InteractiveMetadata._from_parts(
        {"1": "yes"},
        ({"emoji": "1", "label": "Yes", "value": "yes"},),
    )
    assert interactive_metadata is not None

    await apply_post_response_effects(
        FinalDeliveryOutcome(
            terminal_status="error",
            event_id="$stream",
            is_visible_response=True,
            final_visible_body="Choose",
            interactive_metadata=interactive_metadata,
        ),
        ResponseOutcome(
            interactive_target=target,
        ),
        PostResponseEffectsDeps(
            logger=MagicMock(),
            register_interactive=register_interactive,
        ),
    )

    register_interactive.assert_not_awaited()


@pytest.mark.asyncio
async def test_post_response_effects_queues_summary_with_stale_hint_inside_margin(tmp_path: Path) -> None:
    """A stale hint just below threshold should still reach the live summary check."""
    config = _config(tmp_path)
    runtime_paths = runtime_paths_for(config)
    client = AsyncMock(spec=nio.AsyncClient)
    runtime = BotRuntimeState(
        client=client,
        config=config,
        runtime_paths=runtime_paths,
        enable_streaming=False,
        orchestrator=None,
        event_cache=make_event_cache_mock(),
        event_cache_write_coordinator=make_event_cache_write_coordinator_mock(),
    )
    conversation_cache = MagicMock()
    support = PostResponseEffectsSupport(
        runtime=runtime,
        logger=MagicMock(),
        runtime_paths=runtime_paths,
        delivery_gateway=MagicMock(),
        conversation_cache=conversation_cache,
    )
    deps = support.build_deps(
        room_id="!room:localhost",
        interactive_agent_name="general",
    )
    thread_history = [
        ResolvedVisibleMessage.synthetic(
            sender=f"@user{i}:localhost",
            body=f"Message {i}",
            timestamp=i,
            event_id=f"$message{i}",
        )
        for i in range(5)
    ]
    conversation_cache.get_strict_thread_history = AsyncMock(return_value=thread_history)
    scheduled_tasks: list[asyncio.Task[None]] = []

    def schedule_background_task(
        coro: Coroutine[object, object, None],
        *,
        name: str,
        error_handler: object | None = None,  # noqa: ARG001
        owner: object | None = None,  # noqa: ARG001
    ) -> asyncio.Task[None]:
        task = asyncio.create_task(coro, name=name)
        scheduled_tasks.append(task)
        return task

    with (
        patch("mindroom.post_response_effects.create_background_task", side_effect=schedule_background_task),
        patch("mindroom.thread_summary._generate_summary", new=AsyncMock(return_value="Summary")) as mock_generate,
        patch("mindroom.thread_summary.send_thread_summary_event", new=AsyncMock(return_value="$summary")) as mock_send,
    ):
        await apply_post_response_effects(
            FinalDeliveryOutcome(
                terminal_status="completed",
                event_id="$response",
                is_visible_response=True,
                final_visible_body="response",
                delivery_kind="sent",
            ),
            ResponseOutcome(
                thread_summary_room_id="!room:localhost",
                thread_summary_thread_id="$thread",
                thread_summary_message_count_hint=4,
            ),
            deps,
        )

        assert scheduled_tasks
        await asyncio.gather(*scheduled_tasks)

    conversation_cache.get_strict_thread_history.assert_awaited_once_with(
        "!room:localhost",
        "$thread",
        caller_label="thread_summary_background",
    )
    mock_generate.assert_awaited_once_with(
        thread_history,
        config,
        runtime_paths,
        model_name="default",
        tag_vocabulary=None,
        trusted_sender_ids=current_internal_sender_ids(config, runtime_paths),
    )
    mock_send.assert_awaited_once_with(
        client,
        "!room:localhost",
        "$thread",
        "Summary",
        5,
        "default",
        conversation_cache,
        initial_enrichment_complete=None,
    )


@pytest.mark.asyncio
async def test_post_response_effects_queues_summary_with_entity_model_for_adhoc_room(tmp_path: Path) -> None:
    """Ad-hoc invited rooms should inherit the responding entity's summary model override."""
    config = _config(tmp_path)
    config.room_thread_summary_models["general"] = "qwen"
    config.models["qwen"] = ModelConfig(provider="openai", id="qwen-test-model")
    runtime_paths = runtime_paths_for(config)
    client = AsyncMock(spec=nio.AsyncClient)
    runtime = BotRuntimeState(
        client=client,
        config=config,
        runtime_paths=runtime_paths,
        enable_streaming=False,
        orchestrator=None,
        event_cache=make_event_cache_mock(),
        event_cache_write_coordinator=make_event_cache_write_coordinator_mock(),
    )
    conversation_cache = MagicMock()
    support = PostResponseEffectsSupport(
        runtime=runtime,
        logger=MagicMock(),
        runtime_paths=runtime_paths,
        delivery_gateway=MagicMock(),
        conversation_cache=conversation_cache,
    )
    deps = support.build_deps(
        room_id="!adhoc:localhost",
        interactive_agent_name="general",
    )
    thread_history = [
        ResolvedVisibleMessage.synthetic(
            sender=f"@user{i}:localhost",
            body=f"Message {i}",
            timestamp=i,
            event_id=f"$message{i}",
        )
        for i in range(5)
    ]
    conversation_cache.get_strict_thread_history = AsyncMock(return_value=thread_history)
    scheduled_tasks: list[asyncio.Task[None]] = []

    def schedule_background_task(
        coro: Coroutine[object, object, None],
        *,
        name: str,
        error_handler: object | None = None,  # noqa: ARG001
        owner: object | None = None,  # noqa: ARG001
    ) -> asyncio.Task[None]:
        task = asyncio.create_task(coro, name=name)
        scheduled_tasks.append(task)
        return task

    with (
        patch("mindroom.post_response_effects.create_background_task", side_effect=schedule_background_task),
        patch("mindroom.thread_summary._generate_summary", new=AsyncMock(return_value="Summary")) as mock_generate,
        patch("mindroom.thread_summary.send_thread_summary_event", new=AsyncMock(return_value="$summary")),
    ):
        await apply_post_response_effects(
            FinalDeliveryOutcome(
                terminal_status="completed",
                event_id="$response",
                is_visible_response=True,
                final_visible_body="response",
                delivery_kind="sent",
            ),
            ResponseOutcome(
                thread_summary_room_id="!adhoc:localhost",
                thread_summary_thread_id="$thread",
                thread_summary_message_count_hint=4,
                thread_summary_entity_name="general",
            ),
            deps,
        )

        assert scheduled_tasks
        await asyncio.gather(*scheduled_tasks)

    mock_generate.assert_awaited_once_with(
        thread_history,
        config,
        runtime_paths,
        model_name="qwen",
        tag_vocabulary=None,
        trusted_sender_ids=current_internal_sender_ids(config, runtime_paths),
    )


@pytest.mark.asyncio
async def test_generate_response_sets_queued_signal_for_human_ingress(tmp_path: Path) -> None:
    """A waiting human-authored turn should notify the active turn before blocking on the lock."""
    bot = _bot(tmp_path)
    response_envelope = _envelope()
    response_target = response_envelope.target
    coordinator = unwrap_extracted_collaborator(bot._response_runner)
    lifecycle = coordinator._lifecycle_coordinator
    lifecycle_lock = lifecycle._response_lifecycle_lock(response_target)
    queued_signal = lifecycle._get_or_create_queued_signal(response_target)
    await lifecycle_lock.acquire()

    try:
        with patch.object(
            ResponseRunner,
            "_generate_response_locked",
            new=AsyncMock(return_value="$response"),
        ) as mock_locked:
            task = asyncio.create_task(
                bot._response_runner.generate_response(
                    ResponseRequest(
                        prompt="hello",
                        thread_history=[],
                        user_id="@user:localhost",
                        response_envelope=response_envelope,
                    ),
                ),
            )
            await asyncio.wait_for(queued_signal.wait(), timeout=0.2)
            lifecycle_lock.release()
            assert await task == "$response"
            mock_locked.assert_awaited_once()
    finally:
        if lifecycle_lock.locked():
            lifecycle_lock.release()

    assert not queued_signal.is_set()


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "response_envelope",
    [
        pytest.param(_envelope(source_kind=SCHEDULED_SOURCE_KIND), id="scheduled"),
        pytest.param(_envelope(source_kind=HOOK_SOURCE_KIND), id="hook"),
        pytest.param(_envelope(source_kind=HOOK_DISPATCH_SOURCE_KIND), id="hook-dispatch"),
        pytest.param(
            _envelope(
                sender_id="@mindroom_router:localhost",
                requester_id="@mindroom_router:localhost",
                origin=message_origin(
                    sender_id="@mindroom_router:localhost",
                    requester_id="@mindroom_router:localhost",
                    sender_entity_name="router",
                    requester_entity_name="router",
                    source_kind=MESSAGE_SOURCE_KIND,
                ),
            ),
            id="router-notice",
        ),
    ],
)
async def test_generate_response_skips_signal_for_non_human_prompt_ingress(
    tmp_path: Path,
    response_envelope: MessageEnvelope,
) -> None:
    """Automation and router-authored notices should not interrupt the active turn."""
    bot = _bot(tmp_path)
    response_target = response_envelope.target
    coordinator = unwrap_extracted_collaborator(bot._turn_controller.deps.response_runner)
    lifecycle = coordinator._lifecycle_coordinator
    lifecycle_lock = lifecycle._response_lifecycle_lock(response_target)
    queued_signal = lifecycle._get_or_create_queued_signal(response_target)
    await lifecycle_lock.acquire()

    try:
        with patch.object(
            ResponseRunner,
            "_generate_response_locked",
            new=AsyncMock(return_value="$response"),
        ):
            task = asyncio.create_task(
                bot._response_runner.generate_response(
                    ResponseRequest(
                        prompt="hello",
                        thread_history=[],
                        user_id="@user:localhost",
                        response_envelope=response_envelope,
                    ),
                ),
            )
            with pytest.raises(asyncio.TimeoutError):
                await asyncio.wait_for(queued_signal.wait(), timeout=0.05)
            lifecycle_lock.release()
            await task
    finally:
        if lifecycle_lock.locked():
            lifecycle_lock.release()

    assert not queued_signal.is_set()


def test_forced_compaction_placeholder_check_degrades_on_storage_error(tmp_path: Path) -> None:
    """Storage errors in the placeholder-ordering hint should not abort response generation."""
    bot = _bot(tmp_path)
    coordinator = unwrap_extracted_collaborator(bot._response_runner)
    scope = HistoryScope(kind="agent", scope_id="home")

    with patch.object(
        coordinator.deps.state_writer,
        "create_storage",
        side_effect=RuntimeError("storage unavailable"),
    ):
        result = coordinator._has_queued_forced_compaction(
            session_id="session",
            scope=scope,
            execution_identity=None,
        )

    assert result is False


@pytest.mark.asyncio
async def test_generate_response_sets_queued_signal_for_trusted_router_relay(tmp_path: Path) -> None:
    """A trusted router relay carrying a human requester should notify the active turn."""
    bot = _bot(tmp_path)
    response_envelope = _envelope(
        sender_id="@mindroom_router:localhost",
        requester_id="@user:localhost",
        source_kind=TRUSTED_INTERNAL_RELAY_SOURCE_KIND,
        origin=message_origin(
            sender_id="@mindroom_router:localhost",
            requester_id="@user:localhost",
            sender_entity_name="router",
            requester_entity_name=None,
            source_kind=TRUSTED_INTERNAL_RELAY_SOURCE_KIND,
            original_sender="@user:localhost",
            trusted_user_relay=True,
        ),
    )
    response_target = response_envelope.target
    coordinator = unwrap_extracted_collaborator(bot._response_runner)
    lifecycle = coordinator._lifecycle_coordinator
    lifecycle_lock = lifecycle._response_lifecycle_lock(response_target)
    queued_signal = lifecycle._get_or_create_queued_signal(response_target)
    await lifecycle_lock.acquire()

    try:
        with patch.object(
            ResponseRunner,
            "_generate_response_locked",
            new=AsyncMock(return_value="$response"),
        ):
            task = asyncio.create_task(
                bot._response_runner.generate_response(
                    ResponseRequest(
                        prompt="hello",
                        thread_history=[],
                        user_id="@user:localhost",
                        response_envelope=response_envelope,
                    ),
                ),
            )
            await asyncio.wait_for(queued_signal.wait(), timeout=0.2)
            lifecycle_lock.release()
            assert await task == "$response"
    finally:
        if lifecycle_lock.locked():
            lifecycle_lock.release()

    assert not queued_signal.is_set()


@pytest.mark.asyncio
async def test_generate_response_detects_active_turn_before_lock_is_held(tmp_path: Path) -> None:
    """A second human turn should queue even before the first acquires the lifecycle lock."""
    bot = _bot(tmp_path)
    coordinator = unwrap_extracted_collaborator(bot._response_runner)
    lock = _PrelockBarrierLock()
    first_envelope = _envelope(source_event_id="$first")
    second_envelope = _envelope(source_event_id="$second")

    async def fake_generate_response_locked(
        _self: ResponseRunner,
        request: ResponseRequest,
        *,
        resolved_target: MessageTarget,
        early_placeholder_state: object,
    ) -> str:
        del resolved_target, early_placeholder_state
        return str(request.user_id)

    with (
        patch.object(coordinator._lifecycle_coordinator, "_response_lifecycle_lock", return_value=lock),
        patch.object(ResponseRunner, "_generate_response_locked", new=fake_generate_response_locked),
    ):
        first_task = asyncio.create_task(
            bot._response_runner.generate_response(
                ResponseRequest(
                    prompt="hello",
                    thread_history=[],
                    user_id="first",
                    response_envelope=first_envelope,
                ),
            ),
        )
        await lock.first_waiting.wait()

        second_task = asyncio.create_task(
            bot._response_runner.generate_response(
                ResponseRequest(
                    prompt="stop",
                    thread_history=[],
                    user_id="second",
                    response_envelope=second_envelope,
                ),
            ),
        )

        lock._allow_first_entry.set()
        second_result = await second_task
        first_result = await first_task

    assert second_result == "second"
    assert first_result == "first"


@pytest.mark.asyncio
async def test_generate_response_waits_for_lock_before_starting_placeholder_lifecycle(tmp_path: Path) -> None:
    """A queued scheduled turn should not start the placeholder lifecycle until it owns the lock."""
    bot = _bot(tmp_path)
    response_envelope = _envelope(source_kind=SCHEDULED_SOURCE_KIND)
    response_target = response_envelope.target
    coordinator = unwrap_extracted_collaborator(bot._response_runner)
    lifecycle_lock = coordinator._lifecycle_coordinator._response_lifecycle_lock(response_target)
    await lifecycle_lock.acquire()
    lifecycle_started = asyncio.Event()

    async def fake_run_cancellable_response(*_args: object, **kwargs: object) -> str:
        lifecycle_started.set()
        response_function = kwargs["response_function"]
        await response_function(None)
        return "$response"

    try:
        with (
            patch.object(
                ResponseRunner,
                "_process_and_respond",
                new=AsyncMock(
                    return_value=_ResponseGenerationOutcome(
                        delivery=FinalDeliveryOutcome(
                            terminal_status="completed",
                            event_id="$response",
                            is_visible_response=True,
                            final_visible_body="ok",
                            delivery_kind="sent",
                        ),
                        run_succeeded=True,
                    ),
                ),
            ),
            patch.object(
                ResponseRunner,
                "_run_cancellable_response",
                new=AsyncMock(side_effect=fake_run_cancellable_response),
            ) as mock_run_cancellable_response,
            patch("mindroom.response_runner.should_use_streaming", new_callable=AsyncMock, return_value=False),
            patch("mindroom.response_runner.reprioritize_auto_flush_sessions", new=MagicMock()),
            patch("mindroom.response_lifecycle.apply_post_response_effects", new=AsyncMock()),
        ):
            task = asyncio.create_task(
                bot._response_runner.generate_response(
                    ResponseRequest(
                        prompt="hello",
                        thread_history=[],
                        user_id="@user:localhost",
                        response_envelope=response_envelope,
                    ),
                ),
            )
            await asyncio.sleep(0.05)
            mock_run_cancellable_response.assert_not_awaited()

            lifecycle_lock.release()
            await asyncio.wait_for(lifecycle_started.wait(), timeout=0.2)
            resolution = await task
            assert resolution == "$response"
    finally:
        if lifecycle_lock.locked():
            lifecycle_lock.release()


@pytest.mark.asyncio
async def test_refresh_model_history_after_lock_refreshes_empty_thread_history(tmp_path: Path) -> None:
    """Threaded turns with an empty cached history should still refresh after lock handoff."""
    bot = _bot(tmp_path)
    coordinator = unwrap_extracted_collaborator(bot._response_runner)
    resolver = unwrap_extracted_collaborator(coordinator.deps.resolver)
    fresh_history = ThreadHistoryResult(
        [
            ResolvedVisibleMessage.synthetic(
                sender="@user:localhost",
                body="updated",
                event_id="$reply",
                content={"body": "updated"},
            ),
        ],
        is_full_history=True,
    )

    with patch.object(
        resolver,
        "fetch_thread_history",
        new=AsyncMock(return_value=fresh_history),
    ) as mock_fetch_thread_history:
        request = await coordinator._refresh_model_history_after_lock(
            ResponseRequest(
                thread_history=[],
                prompt="hello",
                user_id="@user:localhost",
                response_envelope=request_envelope(
                    room_id="!room:localhost",
                    reply_to_event_id="$event",
                    thread_id="$thread",
                    prompt="hello",
                    user_id="@user:localhost",
                ),
            ),
        )

    mock_fetch_thread_history.assert_awaited_once_with(
        "!room:localhost",
        "$thread",
        caller_label="dispatch_post_lock_refresh",
    )
    assert request.thread_history == fresh_history


@pytest.mark.asyncio
async def test_refresh_model_history_after_lock_does_not_reprove_room_target(
    tmp_path: Path,
) -> None:
    """Post-lock model refresh must not re-prove a finalized room target."""
    bot = _bot(tmp_path)
    coordinator = unwrap_extracted_collaborator(bot._response_runner)
    resolver = unwrap_extracted_collaborator(coordinator.deps.resolver)
    target = MessageTarget.resolve("!room:localhost", None, "$event", room_mode=True)
    envelope = _envelope(source_event_id="$event", target=target)

    with patch.object(
        resolver,
        "fetch_thread_history",
        new=AsyncMock(side_effect=AssertionError("room targets have no model thread history to refresh")),
    ):
        request = await coordinator._refresh_model_history_after_lock(
            ResponseRequest(
                thread_history=[],
                prompt="hello",
                user_id="@user:localhost",
                response_envelope=envelope,
                requires_model_history_refresh=True,
            ),
        )

    assert request.thread_id is None
    assert request.thread_history == []
    assert request.response_envelope.target.resolved_thread_id is None


@pytest.mark.asyncio
async def test_generate_response_uses_post_lock_reproof_target(tmp_path: Path) -> None:
    """Agent delivery must enter the runner with the finalized stable room target."""
    bot = _bot(tmp_path)
    coordinator = unwrap_extracted_collaborator(bot._response_runner)
    stable_target = MessageTarget.resolve("!room:localhost", None, "$event", room_mode=True)
    observed_run_targets: list[MessageTarget] = []
    observed_delivery_targets: list[MessageTarget | None] = []

    async def fake_run_cancellable_response(**kwargs: object) -> str:
        target = kwargs["target"]
        assert isinstance(target, MessageTarget)
        observed_run_targets.append(target)
        response_function = cast(
            "Callable[[object | None], Awaitable[object]]",
            kwargs["response_function"],
        )
        await response_function(None)
        return "$response"

    async def fake_process_and_respond(
        request: ResponseRequest,
        **_kwargs: object,
    ) -> _ResponseGenerationOutcome:
        observed_delivery_targets.append(request.response_envelope.target)
        return _ResponseGenerationOutcome(
            delivery=FinalDeliveryOutcome(
                terminal_status="completed",
                event_id="$response",
                is_visible_response=True,
                final_visible_body="ok",
                delivery_kind="sent",
            ),
            run_succeeded=True,
        )

    with (
        patch.object(coordinator, "_build_lifecycle", MagicMock(return_value=_NoopResponseLifecycle())),
        patch.object(
            coordinator,
            "_run_cancellable_response",
            new=AsyncMock(side_effect=fake_run_cancellable_response),
        ),
        patch.object(
            coordinator,
            "_process_and_respond",
            new=AsyncMock(side_effect=fake_process_and_respond),
        ),
        patch("mindroom.response_runner.should_use_streaming", AsyncMock(return_value=False)),
    ):
        result = await coordinator.generate_response(
            ResponseRequest(
                thread_history=[],
                prompt="hello",
                user_id="@user:localhost",
                response_envelope=_envelope(source_event_id="$event", target=stable_target),
            ),
        )

    assert result == "$response"
    assert [target.resolved_thread_id for target in observed_run_targets] == [None]
    assert [target.resolved_thread_id if target is not None else None for target in observed_delivery_targets] == [None]
    lock_keys = set(coordinator._lifecycle_coordinator._response_lifecycle_locks)
    assert ("!room:localhost", None) in lock_keys
    assert ("!room:localhost", "$plain_root") not in lock_keys


@pytest.mark.asyncio
async def test_generate_response_keeps_locked_target_when_payload_preparation_retargets(tmp_path: Path) -> None:
    """Post-lock request preparation may refresh context, but it must not retarget delivery."""
    bot = _bot(tmp_path)
    coordinator = unwrap_extracted_collaborator(bot._response_runner)
    stable_target = MessageTarget.resolve("!room:localhost", None, "$event", room_mode=True)
    retarget = MessageTarget.resolve("!room:localhost", "$other_thread", "$event")
    observed_run_targets: list[MessageTarget] = []
    observed_delivery_targets: list[MessageTarget | None] = []
    observed_lifecycle_targets: list[MessageTarget] = []

    async def fake_prepare(request: ResponseRequest) -> ResponseRequest:
        return replace(
            request,
            response_envelope=_envelope(source_event_id="$event", target=retarget),
            payload_preparation=None,
        )

    async def fake_run_cancellable_response(**kwargs: object) -> str:
        target = kwargs["target"]
        assert isinstance(target, MessageTarget)
        observed_run_targets.append(target)
        response_function = cast(
            "Callable[[object | None], Awaitable[object]]",
            kwargs["response_function"],
        )
        await response_function(None)
        return "$response"

    async def fake_process_and_respond(
        request: ResponseRequest,
        **_kwargs: object,
    ) -> _ResponseGenerationOutcome:
        observed_delivery_targets.append(request.response_envelope.target)
        return _ResponseGenerationOutcome(
            delivery=FinalDeliveryOutcome(
                terminal_status="completed",
                event_id="$response",
                is_visible_response=True,
                final_visible_body="ok",
                delivery_kind="sent",
            ),
            run_succeeded=True,
        )

    def fake_build_lifecycle(**kwargs: object) -> _NoopResponseLifecycle:
        request = kwargs["request"]
        assert isinstance(request, ResponseRequest)
        observed_lifecycle_targets.append(request.response_envelope.target)
        return _NoopResponseLifecycle()

    with (
        patch.object(coordinator, "_build_lifecycle", MagicMock(side_effect=fake_build_lifecycle)),
        patch.object(
            coordinator.deps.request_preparer,
            "prepare",
            new=AsyncMock(side_effect=fake_prepare),
        ),
        patch.object(
            coordinator,
            "_run_cancellable_response",
            new=AsyncMock(side_effect=fake_run_cancellable_response),
        ),
        patch.object(
            coordinator,
            "_process_and_respond",
            new=AsyncMock(side_effect=fake_process_and_respond),
        ),
        patch("mindroom.response_runner.should_use_streaming", AsyncMock(return_value=False)),
    ):
        result = await coordinator.generate_response(
            ResponseRequest(
                thread_history=[],
                prompt="hello",
                user_id="@user:localhost",
                response_envelope=_envelope(source_event_id="$event", target=stable_target),
                payload_preparation=_payload_preparation(stable_target),
            ),
        )

    assert result == "$response"
    assert observed_run_targets == [stable_target]
    assert observed_delivery_targets == [stable_target]
    assert observed_lifecycle_targets == [stable_target]
    lock_keys = set(coordinator._lifecycle_coordinator._response_lifecycle_locks)
    assert ("!room:localhost", None) in lock_keys
    assert ("!room:localhost", "$other_thread") not in lock_keys


@pytest.mark.asyncio
async def test_generate_team_response_uses_post_lock_reproof_target(tmp_path: Path) -> None:
    """Team delivery/session setup must enter the runner with the finalized stable room target."""
    bot = _bot(tmp_path)
    bot.client = MagicMock()
    cached_room = MagicMock(encrypted=False)
    bot.client.rooms = {"!room:localhost": cached_room}
    bot.client.room_send = AsyncMock()
    bot.client.room_typing = AsyncMock()
    bot.orchestrator = MagicMock()
    coordinator = unwrap_extracted_collaborator(bot._response_runner)
    stable_target = MessageTarget.resolve("!room:localhost", None, "$event", room_mode=True)
    lifecycle = _NoopResponseLifecycle()
    observed_run_targets: list[MessageTarget] = []
    observed_delivery_targets: list[MessageTarget] = []

    async def fake_run_cancellable_response(**kwargs: object) -> str:
        target = kwargs["target"]
        assert isinstance(target, MessageTarget)
        observed_run_targets.append(target)
        response_function = cast(
            "Callable[[object | None], Awaitable[object]]",
            kwargs["response_function"],
        )
        await response_function(None)
        return "$response"

    async def fake_deliver_final(request: FinalDeliveryRequest) -> FinalDeliveryOutcome:
        observed_delivery_targets.append(request.target)
        return FinalDeliveryOutcome(
            terminal_status="completed",
            event_id="$response",
            is_visible_response=True,
            final_visible_body=request.response_text,
            delivery_kind="sent",
        )

    with (
        patch.object(coordinator, "_build_lifecycle", MagicMock(return_value=lifecycle)),
        patch.object(
            coordinator,
            "_run_cancellable_response",
            new=AsyncMock(side_effect=fake_run_cancellable_response),
        ),
        patch("mindroom.delivery_gateway.DeliveryGateway.deliver_final", new=AsyncMock(side_effect=fake_deliver_final)),
        patch("mindroom.response_runner.should_use_streaming", AsyncMock(return_value=False)),
        patch("mindroom.response_runner.team_response", AsyncMock(return_value="team ok")),
    ):
        result = await coordinator.generate_team_response_helper(
            ResponseRequest(
                thread_history=[],
                prompt="hello",
                user_id="@user:localhost",
                response_envelope=_envelope(source_event_id="$event", target=stable_target),
            ),
            team_agents=[],
            team_mode="coordinate",
        )

    assert result == "$response"
    assert [target.resolved_thread_id for target in observed_run_targets] == [None]
    assert [target.resolved_thread_id for target in observed_delivery_targets] == [None]
    assert lifecycle.session_thread_ids == [None]
    lock_keys = set(coordinator._lifecycle_coordinator._response_lifecycle_locks)
    assert ("!room:localhost", None) in lock_keys
    assert ("!room:localhost", "$plain_root") not in lock_keys


@pytest.mark.asyncio
async def test_generate_team_response_keeps_locked_target_when_payload_preparation_retargets(tmp_path: Path) -> None:
    """Team response setup and delivery should keep the target selected before lock acquisition."""
    bot = _bot(tmp_path)
    bot.client = MagicMock()
    cached_room = MagicMock(encrypted=False)
    bot.client.rooms = {"!room:localhost": cached_room}
    bot.client.room_send = AsyncMock()
    bot.client.room_typing = AsyncMock()
    bot.orchestrator = MagicMock()
    coordinator = unwrap_extracted_collaborator(bot._response_runner)
    stable_target = MessageTarget.resolve("!room:localhost", None, "$event", room_mode=True)
    retarget = MessageTarget.resolve("!room:localhost", "$other_thread", "$event")
    lifecycle = _NoopResponseLifecycle()
    observed_run_targets: list[MessageTarget] = []
    observed_delivery_targets: list[MessageTarget] = []

    async def fake_prepare(request: ResponseRequest) -> ResponseRequest:
        return replace(
            request,
            response_envelope=_envelope(source_event_id="$event", target=retarget),
            payload_preparation=None,
        )

    async def fake_run_cancellable_response(**kwargs: object) -> str:
        target = kwargs["target"]
        assert isinstance(target, MessageTarget)
        observed_run_targets.append(target)
        response_function = cast(
            "Callable[[object | None], Awaitable[object]]",
            kwargs["response_function"],
        )
        await response_function(None)
        return "$response"

    async def fake_deliver_final(request: FinalDeliveryRequest) -> FinalDeliveryOutcome:
        observed_delivery_targets.append(request.target)
        return FinalDeliveryOutcome(
            terminal_status="completed",
            event_id="$response",
            is_visible_response=True,
            final_visible_body=request.response_text,
            delivery_kind="sent",
        )

    with (
        patch.object(coordinator, "_build_lifecycle", MagicMock(return_value=lifecycle)),
        patch.object(
            coordinator.deps.request_preparer,
            "prepare",
            new=AsyncMock(side_effect=fake_prepare),
        ),
        patch.object(
            coordinator,
            "_run_cancellable_response",
            new=AsyncMock(side_effect=fake_run_cancellable_response),
        ),
        patch("mindroom.delivery_gateway.DeliveryGateway.deliver_final", new=AsyncMock(side_effect=fake_deliver_final)),
        patch("mindroom.response_runner.should_use_streaming", AsyncMock(return_value=False)),
        patch("mindroom.response_runner.team_response", AsyncMock(return_value="team ok")),
    ):
        result = await coordinator.generate_team_response_helper(
            ResponseRequest(
                thread_history=[],
                prompt="hello",
                user_id="@user:localhost",
                response_envelope=_envelope(source_event_id="$event", target=stable_target),
                payload_preparation=_payload_preparation(stable_target),
            ),
            team_agents=[],
            team_mode="coordinate",
        )

    assert result == "$response"
    assert observed_run_targets == [stable_target]
    assert observed_delivery_targets == [stable_target]
    assert lifecycle.session_thread_ids == [None]
    lock_keys = set(coordinator._lifecycle_coordinator._response_lifecycle_locks)
    assert ("!room:localhost", None) in lock_keys
    assert ("!room:localhost", "$other_thread") not in lock_keys


@pytest.mark.asyncio
async def test_prepare_request_after_lock_wraps_refresh_failures(tmp_path: Path) -> None:
    """Post-lock refresh failures should route through the normalized preparation error boundary."""
    bot = _bot(tmp_path)
    coordinator = unwrap_extracted_collaborator(bot._response_runner)
    resolver = unwrap_extracted_collaborator(coordinator.deps.resolver)

    with (
        patch.object(
            resolver,
            "fetch_thread_history",
            new=AsyncMock(side_effect=RuntimeError("repair required")),
        ),
        pytest.raises(PostLockRequestPreparationError) as excinfo,
    ):
        await coordinator._prepare_request_after_lock(
            ResponseRequest(
                thread_history=[],
                prompt="hello",
                user_id="@user:localhost",
                response_envelope=request_envelope(
                    room_id="!room:localhost",
                    reply_to_event_id="$event",
                    thread_id="$thread",
                    prompt="hello",
                    user_id="@user:localhost",
                ),
                requires_model_history_refresh=True,
            ),
        )

    assert isinstance(excinfo.value.__cause__, RuntimeError)


@pytest.mark.asyncio
async def test_generate_team_response_helper_sets_queued_signal(tmp_path: Path) -> None:
    """Team responses should raise the same queued-message signal before waiting on the lock."""
    bot = _bot(tmp_path)
    response_envelope = _envelope()
    response_target = response_envelope.target
    coordinator = unwrap_extracted_collaborator(bot._response_runner)
    lifecycle = coordinator._lifecycle_coordinator
    lifecycle_lock = lifecycle._response_lifecycle_lock(response_target)
    queued_signal = lifecycle._get_or_create_queued_signal(response_target)
    await lifecycle_lock.acquire()

    try:
        with patch.object(
            ResponseRunner,
            "_generate_team_response_helper_locked",
            new=AsyncMock(return_value="$team-response"),
        ) as mock_locked:
            task = asyncio.create_task(
                bot._response_runner.generate_team_response_helper(
                    ResponseRequest(
                        thread_history=[],
                        prompt="hello",
                        user_id="@user:localhost",
                        response_envelope=response_envelope,
                    ),
                    team_agents=[],
                    team_mode="coordinate",
                ),
            )
            await asyncio.wait_for(queued_signal.wait(), timeout=0.2)
            lifecycle_lock.release()
            assert await task == "$team-response"
            mock_locked.assert_awaited_once()
    finally:
        if lifecycle_lock.locked():
            lifecycle_lock.release()

    assert not queued_signal.is_set()


@pytest.mark.asyncio
async def test_generate_response_without_reservation_does_not_drain_human_backlog(tmp_path: Path) -> None:
    """Unreserved direct responses should serialize normally instead of owning active follow-up backlog."""
    bot = _bot(tmp_path)
    response_envelope_b = _envelope(source_event_id="$event-b")
    response_envelope_c = _envelope(source_event_id="$event-c")
    response_target = response_envelope_b.target
    coordinator = unwrap_extracted_collaborator(bot._response_runner)
    lifecycle = coordinator._lifecycle_coordinator
    lifecycle_lock = lifecycle._response_lifecycle_lock(response_target)
    queued_signal = lifecycle._get_or_create_queued_signal(response_target)
    observed_pending: list[bool] = []
    second_turn_started = asyncio.Event()
    allow_turns_to_finish = asyncio.Event()

    async def fake_locked(_self: ResponseRunner, *_args: object, **_kwargs: object) -> str:
        observed_pending.append(queued_signal.has_pending_human_messages())
        if len(observed_pending) == 1:
            second_turn_started.set()
        await allow_turns_to_finish.wait()
        return f"$response-{len(observed_pending)}"

    await lifecycle_lock.acquire()
    try:
        with patch.object(ResponseRunner, "_generate_response_locked", new=fake_locked):
            task_b = asyncio.create_task(
                bot._response_runner.generate_response(
                    ResponseRequest(
                        prompt="hello",
                        thread_history=[],
                        user_id="@user:localhost",
                        response_envelope=response_envelope_b,
                    ),
                ),
            )
            await asyncio.wait_for(queued_signal.wait(), timeout=0.2)
            task_c = asyncio.create_task(
                bot._response_runner.generate_response(
                    ResponseRequest(
                        prompt="hello again",
                        thread_history=[],
                        user_id="@user:localhost",
                        response_envelope=response_envelope_c,
                    ),
                ),
            )
            for _ in range(20):
                if queued_signal.pending_human_messages == 2:
                    break
                await asyncio.sleep(0)
            assert queued_signal.pending_human_messages == 2

            lifecycle_lock.release()
            await asyncio.wait_for(second_turn_started.wait(), timeout=0.2)
            assert observed_pending == [True]

            allow_turns_to_finish.set()
            assert await task_b == "$response-1"
            assert await task_c == "$response-2"
    finally:
        if lifecycle_lock.locked():
            lifecycle_lock.release()

    assert observed_pending == [True, False]
    assert not queued_signal.is_set()


@pytest.mark.asyncio
async def test_generate_team_response_without_reservation_does_not_drain_human_backlog(tmp_path: Path) -> None:
    """Unreserved direct team responses should serialize without owning active follow-up backlog."""
    bot = _bot(tmp_path)
    response_envelope_b = _envelope(source_event_id="$team-event-b")
    response_envelope_c = _envelope(source_event_id="$team-event-c")
    response_target = response_envelope_b.target
    coordinator = unwrap_extracted_collaborator(bot._response_runner)
    lifecycle = coordinator._lifecycle_coordinator
    lifecycle_lock = lifecycle._response_lifecycle_lock(response_target)
    queued_signal = lifecycle._get_or_create_queued_signal(response_target)
    observed_pending: list[bool] = []
    second_turn_started = asyncio.Event()
    allow_turns_to_finish = asyncio.Event()

    async def fake_locked(_self: ResponseRunner, *_args: object, **_kwargs: object) -> str:
        observed_pending.append(queued_signal.has_pending_human_messages())
        if len(observed_pending) == 1:
            second_turn_started.set()
        await allow_turns_to_finish.wait()
        return f"$team-response-{len(observed_pending)}"

    await lifecycle_lock.acquire()
    try:
        with patch.object(ResponseRunner, "_generate_team_response_helper_locked", new=fake_locked):
            task_b = asyncio.create_task(
                bot._response_runner.generate_team_response_helper(
                    ResponseRequest(
                        thread_history=[],
                        prompt="hello",
                        user_id="@user:localhost",
                        response_envelope=response_envelope_b,
                    ),
                    team_agents=[],
                    team_mode="coordinate",
                ),
            )
            await asyncio.wait_for(queued_signal.wait(), timeout=0.2)
            task_c = asyncio.create_task(
                bot._response_runner.generate_team_response_helper(
                    ResponseRequest(
                        thread_history=[],
                        prompt="hello again",
                        user_id="@user:localhost",
                        response_envelope=response_envelope_c,
                    ),
                    team_agents=[],
                    team_mode="coordinate",
                ),
            )
            for _ in range(20):
                if queued_signal.pending_human_messages == 2:
                    break
                await asyncio.sleep(0)
            assert queued_signal.pending_human_messages == 2

            lifecycle_lock.release()
            await asyncio.wait_for(second_turn_started.wait(), timeout=0.2)
            assert observed_pending == [True]

            allow_turns_to_finish.set()
            assert await task_b == "$team-response-1"
            assert await task_c == "$team-response-2"
    finally:
        if lifecycle_lock.locked():
            lifecycle_lock.release()

    assert observed_pending == [True, False]
    assert not queued_signal.is_set()


@pytest.mark.asyncio
async def test_reserved_human_follow_up_reaches_active_turn_before_dispatch(tmp_path: Path) -> None:
    """Gate-owned follow-ups should still notify the active response before their dispatch starts."""
    bot = _bot(tmp_path)
    active_envelope = _envelope(source_event_id="$active")
    follow_up_envelope = _envelope(source_event_id="$followup")
    response_target = active_envelope.target
    coordinator = unwrap_extracted_collaborator(bot._response_runner)
    lifecycle = coordinator._lifecycle_coordinator
    queued_signal = lifecycle._get_or_create_queued_signal(response_target)
    active_started = asyncio.Event()
    active_saw_follow_up = asyncio.Event()
    allow_active_to_finish = asyncio.Event()
    follow_up_observed_pending: list[bool] = []

    async def active_operation(_target: MessageTarget) -> str:
        active_started.set()
        await queued_signal.wait()
        if queued_signal.has_pending_human_messages():
            active_saw_follow_up.set()
        await allow_active_to_finish.wait()
        return "$active-response"

    async def follow_up_operation(_target: MessageTarget) -> str:
        follow_up_observed_pending.append(queued_signal.has_pending_human_messages())
        return "$followup-response"

    active_task = asyncio.create_task(
        lifecycle.run_locked_response(
            target=response_target,
            response_envelope=active_envelope,
            queued_notice_reservation=None,
            pipeline_timing=None,
            locked_operation=active_operation,
        ),
    )
    await asyncio.wait_for(active_started.wait(), timeout=0.2)

    reservation = lifecycle.reserve_waiting_human_message(
        target=response_target,
        response_envelope=follow_up_envelope,
    )
    assert reservation is not None
    await asyncio.wait_for(active_saw_follow_up.wait(), timeout=0.2)
    assert queued_signal.pending_human_messages == 1

    follow_up_task = asyncio.create_task(
        lifecycle.run_locked_response(
            target=response_target,
            response_envelope=follow_up_envelope,
            queued_notice_reservation=reservation,
            pipeline_timing=None,
            locked_operation=follow_up_operation,
        ),
    )
    await asyncio.sleep(0)
    assert queued_signal.pending_human_messages == 1

    allow_active_to_finish.set()
    assert await active_task == "$active-response"
    assert await follow_up_task == "$followup-response"

    assert follow_up_observed_pending == [False]
    assert not queued_signal.is_set()


@pytest.mark.asyncio
async def test_response_lifecycle_reservations_clear_individual_notices(tmp_path: Path) -> None:
    """Lifecycle reservations only own queued-message notices, not coalesced follow-up batching."""
    bot = _bot(tmp_path)
    coordinator = unwrap_extracted_collaborator(bot._response_runner)
    lifecycle = coordinator._lifecycle_coordinator
    target = MessageTarget.resolve("!room:localhost", "$thread", "$active")
    active_envelope = _envelope(source_event_id="$active", target=target)
    active_started = asyncio.Event()
    release_active = asyncio.Event()
    follow_up_calls: list[str] = []

    async def active_operation(_target: MessageTarget) -> str:
        active_started.set()
        await release_active.wait()
        return "$active-response"

    async def follow_up_operation(source_event_id: str, _target: MessageTarget) -> str:
        follow_up_calls.append(source_event_id)
        return f"$response-{source_event_id}"

    active_task = asyncio.create_task(
        lifecycle.run_locked_response(
            target=target,
            response_envelope=active_envelope,
            queued_notice_reservation=None,
            pipeline_timing=None,
            locked_operation=active_operation,
        ),
    )
    await asyncio.wait_for(active_started.wait(), timeout=0.2)

    queued_items = []
    for source_event_id, sender_id in (
        ("$a1", "@alice:localhost"),
        ("$b1", "@bob:localhost"),
        ("$a2", "@alice:localhost"),
    ):
        envelope = _envelope(
            source_event_id=source_event_id,
            target=target,
            requester_id=sender_id,
            sender_id=sender_id,
        )
        reservation = lifecycle.reserve_waiting_human_message(target=target, response_envelope=envelope)
        assert reservation is not None
        queued_items.append((source_event_id, envelope, reservation))

    follow_up_tasks = [
        asyncio.create_task(
            lifecycle.run_locked_response(
                target=target,
                response_envelope=envelope,
                queued_notice_reservation=reservation,
                pipeline_timing=None,
                locked_operation=lambda locked_target, source_event_id=source_event_id: follow_up_operation(
                    source_event_id,
                    locked_target,
                ),
            ),
        )
        for source_event_id, envelope, reservation in queued_items
    ]
    await asyncio.sleep(0)
    assert follow_up_calls == []

    release_active.set()
    assert await active_task == "$active-response"
    assert await asyncio.gather(*follow_up_tasks) == [
        "$response-$a1",
        "$response-$b1",
        "$response-$a2",
    ]
    assert follow_up_calls == ["$a1", "$b1", "$a2"]


@pytest.mark.asyncio
async def test_non_human_lock_owner_does_not_clear_pending_human_notice(tmp_path: Path) -> None:
    """Scheduled or hook turns sharing the target lock must not clear human follow-up notices."""
    bot = _bot(tmp_path)
    coordinator = unwrap_extracted_collaborator(bot._response_runner)
    lifecycle = coordinator._lifecycle_coordinator
    target = MessageTarget.resolve("!room:localhost", "$thread", "$active")
    active_envelope = _envelope(source_event_id="$active", target=target)
    human_envelope = _envelope(
        source_event_id="$human",
        target=target,
        dispatch_policy_source_kind=ACTIVE_THREAD_FOLLOW_UP_SOURCE_KIND,
    )
    scheduled_envelope = _envelope(
        source_event_id="$scheduled",
        target=target,
        source_kind=SCHEDULED_SOURCE_KIND,
    )
    active_started = asyncio.Event()
    release_active = asyncio.Event()
    observed_scheduled_pending: list[set[str]] = []
    observed_human_pending: list[set[str]] = []

    async def active_operation(_target: MessageTarget) -> str:
        active_started.set()
        await release_active.wait()
        return "$active-response"

    async def scheduled_operation(_target: MessageTarget) -> str:
        observed_scheduled_pending.append(
            set(lifecycle._get_or_create_queued_signal(target).pending_human_message_event_ids),
        )
        return "$scheduled-response"

    async def human_operation(_target: MessageTarget) -> str:
        observed_human_pending.append(
            set(lifecycle._get_or_create_queued_signal(target).pending_human_message_event_ids),
        )
        return "$human-response"

    active_task = asyncio.create_task(
        lifecycle.run_locked_response(
            target=target,
            response_envelope=active_envelope,
            queued_notice_reservation=None,
            pipeline_timing=None,
            locked_operation=active_operation,
        ),
    )
    await asyncio.wait_for(active_started.wait(), timeout=0.2)

    human_reservation = lifecycle.reserve_waiting_human_message(target=target, response_envelope=human_envelope)
    assert human_reservation is not None
    scheduled_task = asyncio.create_task(
        lifecycle.run_locked_response(
            target=target,
            response_envelope=scheduled_envelope,
            queued_notice_reservation=None,
            pipeline_timing=None,
            locked_operation=scheduled_operation,
        ),
    )
    await asyncio.sleep(0)

    release_active.set()
    assert await active_task == "$active-response"
    assert await scheduled_task == "$scheduled-response"
    assert observed_scheduled_pending == [{"$human"}]
    assert lifecycle._get_or_create_queued_signal(target).pending_human_message_event_ids == {"$human"}

    assert (
        await lifecycle.run_locked_response(
            target=target,
            response_envelope=human_envelope,
            queued_notice_reservation=human_reservation,
            pipeline_timing=None,
            locked_operation=human_operation,
        )
        == "$human-response"
    )
    assert observed_human_pending == [set()]


@pytest.mark.asyncio
async def test_reserved_command_follow_up_cleanup_when_dispatch_returns(tmp_path: Path) -> None:
    """Command-shaped active follow-ups should not leave stale queued-message state."""
    bot = _bot(tmp_path)
    room = MagicMock(spec=nio.MatrixRoom)
    room.room_id = "!room:localhost"
    target = MessageTarget.resolve(room.room_id, "$thread", "$command")
    envelope = _envelope(
        dispatch_policy_source_kind=ACTIVE_THREAD_FOLLOW_UP_SOURCE_KIND,
        source_event_id="$command",
        target=target,
    )
    event = PreparedTextEvent(
        sender="@user:localhost",
        event_id="$command",
        body="!help",
        source={"content": {"body": "!help"}},
        server_timestamp=1234,
    )
    dispatch = PreparedDispatch(
        requester_user_id="@user:localhost",
        context=MessageContext(
            am_i_mentioned=False,
            is_thread=True,
            thread_id="$thread",
            thread_history=[],
            mentioned_agents=[],
            has_non_agent_mentions=False,
        ),
        target=target,
        correlation_id="$command",
        envelope=envelope,
    )
    coordinator = unwrap_extracted_collaborator(bot._response_runner)
    lifecycle = coordinator._lifecycle_coordinator
    queued_signal = lifecycle._get_or_create_queued_signal(target)
    queued_signal.begin_response_turn()
    try:
        reservation = lifecycle.reserve_waiting_human_message(target=target, response_envelope=envelope)
        assert reservation is not None
        with (
            patch.object(
                bot._turn_controller,
                "_prepare_dispatch",
                new=AsyncMock(return_value=prepared_dispatch_result(dispatch)),
            ),
            patch("mindroom.turn_policy.TurnPolicy.plan_turn", new=AsyncMock()) as mock_plan_turn,
        ):
            await bot._turn_controller._dispatch_text_message(
                room,
                _PrecheckedEvent(event=event, requester_user_id="@user:localhost"),
                queued_notice_reservation=reservation,
            )
    finally:
        queued_signal.finish_response_turn()

    mock_plan_turn.assert_not_awaited()
    assert not queued_signal.is_set()


@pytest.mark.asyncio
async def test_reserved_superseded_follow_up_cleanup_when_dispatch_returns(tmp_path: Path) -> None:
    """Superseded active follow-ups should clear their enqueue-time notice when skipped."""
    bot = _bot(tmp_path)
    room = MagicMock(spec=nio.MatrixRoom)
    room.room_id = "!room:localhost"
    target = MessageTarget.resolve(room.room_id, "$thread", "$older")
    envelope = _envelope(
        dispatch_policy_source_kind=ACTIVE_THREAD_FOLLOW_UP_SOURCE_KIND,
        source_event_id="$older",
        target=target,
    )
    event = PreparedTextEvent(
        sender="@user:localhost",
        event_id="$older",
        body="older follow-up",
        source={"content": {"body": "older follow-up"}},
        server_timestamp=1234,
    )
    dispatch = PreparedDispatch(
        requester_user_id="@user:localhost",
        context=MessageContext(
            am_i_mentioned=False,
            is_thread=True,
            thread_id="$thread",
            thread_history=[],
            mentioned_agents=[],
            has_non_agent_mentions=False,
        ),
        target=target,
        correlation_id="$older",
        envelope=envelope,
    )
    coordinator = unwrap_extracted_collaborator(bot._response_runner)
    lifecycle = coordinator._lifecycle_coordinator
    queued_signal = lifecycle._get_or_create_queued_signal(target)
    queued_signal.begin_response_turn()
    try:
        reservation = lifecycle.reserve_waiting_human_message(target=target, response_envelope=envelope)
        assert reservation is not None
        with (
            patch.object(
                bot._turn_controller,
                "_prepare_dispatch",
                new=AsyncMock(return_value=prepared_dispatch_result(dispatch)),
            ),
            patch.object(bot._turn_controller, "_has_newer_unresponded_in_thread", return_value=True),
            patch.object(bot._turn_policy, "plan_turn", new=AsyncMock()) as mock_plan_turn,
        ):
            await bot._turn_controller._dispatch_text_message(
                room,
                _PrecheckedEvent(event=event, requester_user_id="@user:localhost"),
                queued_notice_reservation=reservation,
            )
    finally:
        queued_signal.finish_response_turn()

    mock_plan_turn.assert_not_awaited()
    assert not queued_signal.is_set()


@pytest.mark.asyncio
async def test_reserved_follow_up_cleanup_when_hook_suppression_returns_before_dispatch(tmp_path: Path) -> None:
    """Hook-suppressed active follow-ups should clear the reservation before planning."""
    bot = _bot(tmp_path)
    room = MagicMock(spec=nio.MatrixRoom)
    room.room_id = "!room:localhost"
    case = _reserved_follow_up_case(bot, room, event_id="$suppressed")
    try:
        with (
            patch.object(bot._turn_controller, "_prepare_dispatch", new=AsyncMock(return_value=None)),
            patch("mindroom.turn_policy.TurnPolicy.plan_turn", new=AsyncMock()) as mock_plan_turn,
        ):
            await bot._turn_controller._dispatch_text_message(
                room,
                _PrecheckedEvent(event=case.event, requester_user_id="@user:localhost"),
                queued_notice_reservation=case.reservation,
            )
    finally:
        case.queued_signal.finish_response_turn()

    mock_plan_turn.assert_not_awaited()
    assert not case.queued_signal.is_set()


@pytest.mark.asyncio
async def test_reserved_follow_up_cleanup_when_plan_ignores_before_response(tmp_path: Path) -> None:
    """Ignored active follow-ups should clear the reservation before response lifecycle ownership."""
    bot = _bot(tmp_path)
    room = MagicMock(spec=nio.MatrixRoom)
    room.room_id = "!room:localhost"
    case = _reserved_follow_up_case(bot, room, event_id="$ignored")
    try:
        with (
            patch.object(
                bot._turn_controller,
                "_prepare_dispatch",
                new=AsyncMock(return_value=prepared_dispatch_result(case.dispatch)),
            ),
            patch("mindroom.text_ingress_dispatch.is_dm_room", new=AsyncMock(return_value=False)),
            patch(
                "mindroom.turn_policy.TurnPolicy.plan_turn",
                new=AsyncMock(return_value=_DispatchPlan(kind="ignore")),
            ),
        ):
            await bot._turn_controller._dispatch_text_message(
                room,
                _PrecheckedEvent(event=case.event, requester_user_id="@user:localhost"),
                queued_notice_reservation=case.reservation,
            )
    finally:
        case.queued_signal.finish_response_turn()

    assert not case.queued_signal.is_set()


@pytest.mark.asyncio
async def test_reserved_follow_up_cleanup_when_route_returns_before_response(tmp_path: Path) -> None:
    """Routed active follow-ups should clear the reservation after router handoff."""
    bot = _bot(tmp_path)
    room = MagicMock(spec=nio.MatrixRoom)
    room.room_id = "!room:localhost"
    case = _reserved_follow_up_case(bot, room, event_id="$routed")
    try:
        with (
            patch.object(
                bot._turn_controller,
                "_prepare_dispatch",
                new=AsyncMock(return_value=prepared_dispatch_result(case.dispatch)),
            ),
            patch("mindroom.text_ingress_dispatch.is_dm_room", new=AsyncMock(return_value=False)),
            patch(
                "mindroom.turn_policy.TurnPolicy.plan_turn",
                new=AsyncMock(return_value=_DispatchPlan(kind="route", router_message="route this")),
            ),
            patch.object(bot._turn_controller, "_execute_router_relay", new=AsyncMock()) as mock_route,
        ):
            await bot._turn_controller._dispatch_text_message(
                room,
                _PrecheckedEvent(event=case.event, requester_user_id="@user:localhost"),
                queued_notice_reservation=case.reservation,
            )
    finally:
        case.queued_signal.finish_response_turn()

    mock_route.assert_awaited_once()
    assert not case.queued_signal.is_set()


@pytest.mark.asyncio
async def test_reserved_follow_up_cleanup_when_dispatch_raises_before_lifecycle(tmp_path: Path) -> None:
    """Exceptions before response lifecycle ownership should cancel the reservation."""
    bot = _bot(tmp_path)
    room = MagicMock(spec=nio.MatrixRoom)
    room.room_id = "!room:localhost"
    case = _reserved_follow_up_case(bot, room, event_id="$raises")
    try:
        with (
            patch.object(
                bot._turn_controller,
                "_prepare_dispatch",
                new=AsyncMock(return_value=prepared_dispatch_result(case.dispatch)),
            ),
            patch("mindroom.text_ingress_dispatch.is_dm_room", new=AsyncMock(return_value=False)),
            patch(
                "mindroom.turn_policy.TurnPolicy.plan_turn",
                new=AsyncMock(
                    return_value=_DispatchPlan(
                        kind="respond",
                        response_action=ResponseAction(kind="individual"),
                    ),
                ),
            ),
            patch.object(
                bot._turn_controller,
                "_execute_response_action",
                new=AsyncMock(side_effect=RuntimeError("dispatch failed")),
            ),
            pytest.raises(RuntimeError, match="dispatch failed"),
        ):
            await bot._turn_controller._dispatch_text_message(
                room,
                _PrecheckedEvent(event=case.event, requester_user_id="@user:localhost"),
                queued_notice_reservation=case.reservation,
            )
    finally:
        case.queued_signal.finish_response_turn()

    assert not case.queued_signal.is_set()


@pytest.mark.asyncio
async def test_reserved_follow_up_cleanup_when_dispatch_cancelled_before_lifecycle(tmp_path: Path) -> None:
    """A response task cancelled before lifecycle ownership still cancels the reservation."""
    bot = _bot(tmp_path)
    room = MagicMock(spec=nio.MatrixRoom)
    room.room_id = "!room:localhost"
    case = _reserved_follow_up_case(bot, room, event_id="$cancelled")
    try:
        with (
            patch.object(
                bot._turn_controller,
                "_prepare_dispatch",
                new=AsyncMock(return_value=prepared_dispatch_result(case.dispatch)),
            ),
            patch("mindroom.text_ingress_dispatch.is_dm_room", new=AsyncMock(return_value=False)),
            patch(
                "mindroom.turn_policy.TurnPolicy.plan_turn",
                new=AsyncMock(
                    return_value=_DispatchPlan(
                        kind="respond",
                        response_action=ResponseAction(kind="individual"),
                    ),
                ),
            ),
            patch.object(
                bot._turn_controller,
                "_execute_response_action",
                new=AsyncMock(side_effect=asyncio.CancelledError),
            ),
        ):
            await bot._turn_controller._dispatch_text_message(
                room,
                _PrecheckedEvent(event=case.event, requester_user_id="@user:localhost"),
                queued_notice_reservation=case.reservation,
            )
    finally:
        case.queued_signal.finish_response_turn()

    assert not case.queued_signal.is_set()


@pytest.mark.asyncio
async def test_reserved_follow_up_cleanup_when_text_normalization_raises(tmp_path: Path) -> None:
    """Reserved follow-ups should clean up even before PreparedDispatch can exist."""
    bot = _bot(tmp_path)
    room = MagicMock(spec=nio.MatrixRoom)
    room.room_id = "!room:localhost"
    target = MessageTarget.resolve(room.room_id, "$thread", "$normalize")
    envelope = _envelope(
        dispatch_policy_source_kind=ACTIVE_THREAD_FOLLOW_UP_SOURCE_KIND,
        source_event_id="$normalize",
        target=target,
    )
    event = _prepared_text_event(event_id="$normalize")
    coordinator = unwrap_extracted_collaborator(bot._response_runner)
    lifecycle = coordinator._lifecycle_coordinator
    queued_signal = lifecycle._get_or_create_queued_signal(target)
    queued_signal.begin_response_turn()
    try:
        reservation = lifecycle.reserve_waiting_human_message(target=target, response_envelope=envelope)
        assert reservation is not None
        with (
            patch.object(
                type(bot._turn_controller.deps.normalizer),
                "resolve_text_event",
                new=AsyncMock(side_effect=RuntimeError("normalization failed")),
            ),
            pytest.raises(RuntimeError, match="normalization failed"),
        ):
            await bot._turn_controller._dispatch_text_message(
                room,
                event,
                "@user:localhost",
                queued_notice_reservation=reservation,
            )
    finally:
        queued_signal.finish_response_turn()

    assert not queued_signal.is_set()


@pytest.mark.asyncio
async def test_reserved_follow_up_cleanup_when_prepare_dispatch_raises(tmp_path: Path) -> None:
    """Reserved follow-ups should clean up when dispatch preparation fails."""
    bot = _bot(tmp_path)
    room = MagicMock(spec=nio.MatrixRoom)
    room.room_id = "!room:localhost"
    target = MessageTarget.resolve(room.room_id, "$thread", "$prepare")
    envelope = _envelope(
        dispatch_policy_source_kind=ACTIVE_THREAD_FOLLOW_UP_SOURCE_KIND,
        source_event_id="$prepare",
        target=target,
    )
    event = _prepared_text_event(event_id="$prepare")
    coordinator = unwrap_extracted_collaborator(bot._response_runner)
    lifecycle = coordinator._lifecycle_coordinator
    queued_signal = lifecycle._get_or_create_queued_signal(target)
    queued_signal.begin_response_turn()
    try:
        reservation = lifecycle.reserve_waiting_human_message(target=target, response_envelope=envelope)
        assert reservation is not None
        with (
            patch.object(
                bot._turn_controller,
                "_prepare_dispatch",
                new=AsyncMock(side_effect=RuntimeError("prepare failed")),
            ),
            pytest.raises(RuntimeError, match="prepare failed"),
        ):
            await bot._turn_controller._dispatch_text_message(
                room,
                event,
                "@user:localhost",
                queued_notice_reservation=reservation,
            )
    finally:
        queued_signal.finish_response_turn()

    assert not queued_signal.is_set()


@pytest.mark.asyncio
async def test_reserved_follow_up_cleanup_when_handle_coalesced_batch_fails_before_dispatch(
    tmp_path: Path,
) -> None:
    """Reserved follow-ups claimed by the gate should clean up if handoff fails early."""
    bot = _bot(tmp_path)
    room = MagicMock(spec=nio.MatrixRoom)
    room.room_id = "!room:localhost"
    target = MessageTarget.resolve(room.room_id, "$thread", "$handoff")
    envelope = _envelope(
        dispatch_policy_source_kind=ACTIVE_THREAD_FOLLOW_UP_SOURCE_KIND,
        source_event_id="$handoff",
        target=target,
    )
    event = _prepared_text_event(event_id="$handoff")
    coordinator = unwrap_extracted_collaborator(bot._response_runner)
    lifecycle = coordinator._lifecycle_coordinator
    queued_signal = lifecycle._get_or_create_queued_signal(target)
    queued_signal.begin_response_turn()
    try:
        reservation = lifecycle.reserve_waiting_human_message(target=target, response_envelope=envelope)
        assert reservation is not None
        batch = build_coalesced_batch(
            CoalescingKey(room.room_id, "$thread", "@user:localhost"),
            [
                PendingEvent(
                    event=event,
                    room=room,
                    source_kind=MESSAGE_SOURCE_KIND,
                    dispatch_policy_source_kind=ACTIVE_THREAD_FOLLOW_UP_SOURCE_KIND,
                    dispatch_metadata=_queued_notice_metadata(reservation),
                ),
            ],
        )
        with (
            patch(
                "mindroom.turn_controller.build_dispatch_handoff",
                side_effect=RuntimeError("handoff failed"),
            ),
            pytest.raises(RuntimeError, match="handoff failed"),
        ):
            await bot._turn_controller.handle_coalesced_batch(batch)
    finally:
        queued_signal.finish_response_turn()

    assert not queued_signal.is_set()


@pytest.mark.asyncio
async def test_coalesced_batch_consumes_queued_notice_for_batch_thread(tmp_path: Path) -> None:
    """A mixed batch should consume the notices for its single coalescing target before dispatch."""
    bot = _bot(tmp_path)
    room = MagicMock(spec=nio.MatrixRoom)
    room.room_id = "!room:localhost"
    pre_target = MessageTarget.resolve(room.room_id, "$pre_stt_thread", "$typed")
    post_target = MessageTarget.resolve(room.room_id, "$post_stt_thread", "$voice")
    pre_envelope = _envelope(
        dispatch_policy_source_kind=ACTIVE_THREAD_FOLLOW_UP_SOURCE_KIND,
        source_event_id="$typed",
        target=pre_target,
    )
    post_envelope = _envelope(
        source_kind=VOICE_SOURCE_KIND,
        dispatch_policy_source_kind=ACTIVE_THREAD_FOLLOW_UP_SOURCE_KIND,
        source_event_id="$voice",
        target=post_target,
    )
    coordinator = unwrap_extracted_collaborator(bot._response_runner)
    lifecycle = coordinator._lifecycle_coordinator
    pre_signal = lifecycle._get_or_create_queued_signal(pre_target)
    post_signal = lifecycle._get_or_create_queued_signal(post_target)
    typed_event = _prepared_text_event(event_id="$typed")
    voice_event = replace(
        _prepared_text_event(event_id="$voice"),
        body="🎤 voice transcript",
        source_kind_override=VOICE_SOURCE_KIND,
    )
    captured_dispatches: list[str] = []

    async def capture_dispatch(*_args: object, **kwargs: object) -> None:
        del kwargs
        assert pre_signal.pending_human_messages == 0
        assert post_signal.pending_human_messages == 0
        captured_dispatches.append("dispatched")

    pre_signal.begin_response_turn()
    post_signal.begin_response_turn()
    try:
        pre_reservation = lifecycle.reserve_waiting_human_message(target=pre_target, response_envelope=pre_envelope)
        post_reservation = lifecycle.reserve_waiting_human_message(target=post_target, response_envelope=post_envelope)
        assert pre_reservation is not None
        assert post_reservation is not None
        batch = build_coalesced_batch(
            CoalescingKey(room.room_id, "$post_stt_thread", "@user:localhost"),
            [
                PendingEvent(
                    event=typed_event,
                    room=room,
                    source_kind=MESSAGE_SOURCE_KIND,
                    dispatch_policy_source_kind=ACTIVE_THREAD_FOLLOW_UP_SOURCE_KIND,
                    dispatch_metadata=_targeted_queued_notice_metadata(pre_reservation, pre_target),
                ),
                PendingEvent(
                    event=voice_event,
                    room=room,
                    source_kind=VOICE_SOURCE_KIND,
                    dispatch_policy_source_kind=ACTIVE_THREAD_FOLLOW_UP_SOURCE_KIND,
                    dispatch_metadata=_targeted_queued_notice_metadata(post_reservation, post_target),
                ),
            ],
        )

        with patch.object(bot._turn_controller, "_dispatch_text_message", new=AsyncMock(side_effect=capture_dispatch)):
            await bot._turn_controller.handle_coalesced_batch(batch)
    finally:
        pre_signal.finish_response_turn()
        post_signal.finish_response_turn()

    assert captured_dispatches == ["dispatched"]
    assert pre_signal.pending_human_messages == 0
    assert post_signal.pending_human_messages == 0
    assert not pre_signal.is_set()
    assert not post_signal.is_set()


@pytest.mark.asyncio
async def test_room_scoped_root_voice_consumes_final_target_queued_notice(tmp_path: Path) -> None:
    """A room-scoped voice root should consume the notice for its final target root before dispatch."""
    bot = _bot(tmp_path)
    room = MagicMock(spec=nio.MatrixRoom)
    room.room_id = "!room:localhost"
    voice_event = replace(
        _prepared_text_event(event_id="$voice-root"),
        body="🎤 voice follow-up",
        source_kind_override=VOICE_SOURCE_KIND,
    )
    target = bot._turn_controller.deps.resolver.build_message_target(
        room_id=room.room_id,
        thread_id=None,
        reply_to_event_id=voice_event.event_id,
        event_source=voice_event.source,
    )
    assert target.resolved_thread_id == "$voice-root"
    envelope = _envelope(
        source_kind=VOICE_SOURCE_KIND,
        dispatch_policy_source_kind=ACTIVE_THREAD_FOLLOW_UP_SOURCE_KIND,
        source_event_id=voice_event.event_id,
        target=target,
    )
    coordinator = unwrap_extracted_collaborator(bot._response_runner)
    lifecycle = coordinator._lifecycle_coordinator
    queued_signal = lifecycle._get_or_create_queued_signal(target)
    captured_dispatches: list[str] = []

    async def capture_dispatch(*_args: object, **kwargs: object) -> None:
        del kwargs
        assert queued_signal.pending_human_messages == 0
        captured_dispatches.append("dispatched")

    queued_signal.begin_response_turn()
    try:
        voice_reservation = lifecycle.reserve_waiting_human_message(target=target, response_envelope=envelope)
        assert voice_reservation is not None
        batch = build_coalesced_batch(
            CoalescingKey(room.room_id, None, "@user:localhost"),
            [
                PendingEvent(
                    event=voice_event,
                    room=room,
                    source_kind=VOICE_SOURCE_KIND,
                    dispatch_policy_source_kind=ACTIVE_THREAD_FOLLOW_UP_SOURCE_KIND,
                    dispatch_metadata=_targeted_queued_notice_metadata(voice_reservation, target),
                ),
            ],
        )

        with patch.object(bot._turn_controller, "_dispatch_text_message", new=AsyncMock(side_effect=capture_dispatch)):
            await bot._turn_controller.handle_coalesced_batch(batch)
    finally:
        queued_signal.finish_response_turn()

    assert captured_dispatches == ["dispatched"]
    assert queued_signal.pending_human_messages == 0
    assert not queued_signal.is_set()


@pytest.mark.asyncio
async def test_active_follow_up_reservation_cancelled_when_enqueue_is_cancelled(tmp_path: Path) -> None:
    """Cancellation during enqueue handoff should not leak the reserved notice."""
    bot = _bot(tmp_path)
    room = MagicMock(spec=nio.MatrixRoom)
    room.room_id = "!room:localhost"
    target = MessageTarget.resolve(room.room_id, "$thread", "$cancelled")
    envelope = _envelope(source_kind=MESSAGE_SOURCE_KIND, source_event_id="$cancelled", target=target)
    event = _prepared_text_event(event_id="$cancelled")
    coordinator = unwrap_extracted_collaborator(bot._response_runner)
    lifecycle = coordinator._lifecycle_coordinator
    queued_signal = lifecycle._get_or_create_queued_signal(target)
    queued_signal.begin_response_turn()
    try:
        reservation = lifecycle.reserve_waiting_human_message(target=target, response_envelope=envelope)
        assert reservation is not None
        assert queued_signal.pending_human_messages == 1
        reservation_owner = bot._turn_controller.reserve_prompt_ingress_order(room, "@user:localhost")
        try:
            with (
                patch.object(
                    bot._turn_controller,
                    "_enqueue_for_dispatch",
                    new=AsyncMock(side_effect=asyncio.CancelledError),
                ),
                pytest.raises(asyncio.CancelledError),
            ):
                await bot._turn_controller._enqueue_prepared_text_for_dispatch(
                    room=room,
                    prepared_event=event,
                    dispatch_event=event,
                    envelope=envelope,
                    coalescing_thread_id="$thread",
                    requester_user_id="@user:localhost",
                    reservation_owner=reservation_owner,
                    queued_notice_reservation=reservation,
                )
        finally:
            await reservation_owner.release()
    finally:
        queued_signal.finish_response_turn()

    assert queued_signal.pending_human_messages == 0
    assert not queued_signal.is_set()


def test_queued_human_reservation_is_idempotent(tmp_path: Path) -> None:
    """Reservation consume/cancel operations should clear the notice at most once."""
    bot = _bot(tmp_path)
    target = MessageTarget.resolve("!room:localhost", "$thread", "$event")
    envelope = _envelope(
        dispatch_policy_source_kind=ACTIVE_THREAD_FOLLOW_UP_SOURCE_KIND,
        source_event_id="$event",
        target=target,
    )
    coordinator = unwrap_extracted_collaborator(bot._response_runner)
    lifecycle = coordinator._lifecycle_coordinator
    queued_signal = lifecycle._get_or_create_queued_signal(target)
    queued_signal.begin_response_turn()
    try:
        reservation = lifecycle.reserve_waiting_human_message(target=target, response_envelope=envelope)
        assert reservation is not None
        assert queued_signal.pending_human_messages == 1
        reservation.consume()
        reservation.consume()
        reservation.cancel()
        assert queued_signal.pending_human_messages == 0
    finally:
        queued_signal.finish_response_turn()

    assert not queued_signal.is_set()


def test_managed_message_does_not_reserve_queued_human_notice(tmp_path: Path) -> None:
    """Managed chatter should not count as a queued human follow-up."""
    bot = _bot(tmp_path)
    target = MessageTarget.resolve("!room:localhost", "$thread", "$event")
    envelope = _envelope(
        source_event_id="$event",
        target=target,
        sender_id="@mindroom_general:localhost",
        requester_id="@mindroom_general:localhost",
        origin=message_origin(
            sender_id="@mindroom_general:localhost",
            requester_id="@mindroom_general:localhost",
            sender_entity_name="general",
            requester_entity_name="general",
            source_kind=MESSAGE_SOURCE_KIND,
        ),
    )
    coordinator = unwrap_extracted_collaborator(bot._response_runner)
    lifecycle = coordinator._lifecycle_coordinator
    queued_signal = lifecycle._get_or_create_queued_signal(target)
    queued_signal.begin_response_turn()
    try:
        reservation = lifecycle.reserve_waiting_human_message(target=target, response_envelope=envelope)
        assert reservation is None
        assert queued_signal.pending_human_messages == 0
    finally:
        queued_signal.finish_response_turn()

    assert not queued_signal.is_set()


@pytest.mark.asyncio
async def test_handed_off_reservation_is_cancelled_when_lock_wait_is_cancelled(tmp_path: Path) -> None:
    """A reservation handed to the lifecycle should not leak if lock acquisition is cancelled."""
    bot = _bot(tmp_path)
    target = MessageTarget.resolve("!room:localhost", "$thread", "$event")
    envelope = _envelope(
        dispatch_policy_source_kind=ACTIVE_THREAD_FOLLOW_UP_SOURCE_KIND,
        source_event_id="$event",
        target=target,
    )
    coordinator = unwrap_extracted_collaborator(bot._response_runner)
    lifecycle = coordinator._lifecycle_coordinator
    queued_signal = lifecycle._get_or_create_queued_signal(target)
    queued_signal.begin_response_turn()
    reservation = lifecycle.reserve_waiting_human_message(target=target, response_envelope=envelope)
    assert reservation is not None
    assert queued_signal.pending_human_messages == 1

    lock = lifecycle._response_lifecycle_lock(target)
    await lock.acquire()

    async def locked_operation(_target: MessageTarget) -> str:
        msg = "lock wait should be cancelled before the operation runs"
        raise AssertionError(msg)

    try:
        task = asyncio.create_task(
            lifecycle.run_locked_response(
                target=target,
                response_envelope=envelope,
                queued_notice_reservation=reservation,
                pipeline_timing=None,
                locked_operation=locked_operation,
            ),
        )
        await asyncio.sleep(0)
        task.cancel()
        with pytest.raises(asyncio.CancelledError):
            await task
    finally:
        lock.release()
        queued_signal.finish_response_turn()

    assert queued_signal.pending_human_messages == 0
    assert not queued_signal.is_set()
    assert not queued_signal.has_active_response_turn()


def test_reserved_follow_up_cannot_join_multi_event_batch(tmp_path: Path) -> None:
    """Batch validation should not own cleanup for a reserved active follow-up."""
    bot = _bot(tmp_path)
    room = MagicMock(spec=nio.MatrixRoom)
    room.room_id = "!room:localhost"
    target = MessageTarget.resolve(room.room_id, "$thread", "$reserved")
    envelope = _envelope(
        dispatch_policy_source_kind=ACTIVE_THREAD_FOLLOW_UP_SOURCE_KIND,
        source_event_id="$reserved",
        target=target,
    )
    coordinator = unwrap_extracted_collaborator(bot._response_runner)
    lifecycle = coordinator._lifecycle_coordinator
    queued_signal = lifecycle._get_or_create_queued_signal(target)
    queued_signal.begin_response_turn()
    reservation = None
    try:
        reservation = lifecycle.reserve_waiting_human_message(target=target, response_envelope=envelope)
        assert reservation is not None
        with pytest.raises(ValueError, match="solo batches"):
            build_coalesced_batch(
                CoalescingKey(room.room_id, "$thread", "@user:localhost"),
                [
                    PendingEvent(
                        event=_prepared_text_event(event_id="$reserved"),
                        room=room,
                        source_kind=MESSAGE_SOURCE_KIND,
                        dispatch_policy_source_kind=ACTIVE_THREAD_FOLLOW_UP_SOURCE_KIND,
                        dispatch_metadata=_queued_notice_metadata(reservation),
                    ),
                    PendingEvent(
                        event=_prepared_text_event(event_id="$normal"),
                        room=room,
                        source_kind=MESSAGE_SOURCE_KIND,
                    ),
                ],
            )
        assert queued_signal.is_set()
    finally:
        if reservation is not None:
            reservation.cancel()
        queued_signal.finish_response_turn()

    assert not queued_signal.is_set()


@pytest.mark.asyncio
async def test_coalesced_dispatch_never_creates_queued_signal(tmp_path: Path) -> None:
    """Messages dropped by coalescing should not create false mid-turn notifications."""
    bot = _bot(tmp_path)
    room = MagicMock(spec=nio.MatrixRoom)
    room.room_id = "!room:localhost"
    event = _prepared_text_event(event_id="$older")
    envelope = _envelope()
    dispatch = PreparedDispatch(
        requester_user_id="@user:localhost",
        context=_message_context(),
        target=envelope.target,
        correlation_id="corr",
        envelope=envelope,
    )

    with (
        patch.object(bot._inbound_turn_normalizer, "resolve_text_event", new=AsyncMock(return_value=event)),
        patch.object(
            bot._turn_controller,
            "_prepare_dispatch",
            new=AsyncMock(return_value=prepared_dispatch_result(dispatch)),
        ),
        patch.object(bot._turn_controller, "_has_newer_unresponded_in_thread", return_value=True),
        patch.object(
            bot._turn_policy,
            "plan_turn",
            new=AsyncMock(return_value=_DispatchPlan(kind="ignore")),
        ) as mock_plan,
    ):
        await bot._turn_controller._dispatch_text_message(
            room,
            _PrecheckedEvent(event=event, requester_user_id="@user:localhost"),
        )

    assert not bot._turn_store.is_handled("$older")
    mock_plan.assert_not_awaited()
    coordinator = unwrap_extracted_collaborator(bot._response_runner)
    assert coordinator._lifecycle_coordinator._thread_queued_signals == {}


def test_notice_hook_keeps_single_notice_at_end_and_skips_stop_after_tool_call() -> None:
    """The injected notice should stay unique, remain last, avoid double wrapping, and skip stop-after-tool-call results."""
    model = _FakeModel()
    install_queued_message_notice_hook(model, notice_text=QUEUED_MESSAGE_NOTICE_TEXT)
    install_queued_message_notice_hook(model, notice_text=QUEUED_MESSAGE_NOTICE_TEXT)

    plain_messages = [Message(role="user", content="hello")]
    model.format_function_call_results(
        messages=plain_messages,
        function_call_results=[Message(role="tool", content="result")],
    )
    assert _notice_count(plain_messages) == 0

    with queued_message_signal_context(_StaticQueuedState(pending=True)) as notice_context:
        queued_messages = [Message(role="user", content="hello")]
        model.format_function_call_results(
            messages=queued_messages,
            function_call_results=[Message(role="tool", content="result")],
        )
        model.format_function_call_results(
            messages=queued_messages,
            function_call_results=[Message(role="tool", content="another result")],
        )

        stop_after_messages = [Message(role="user", content="hello")]
        stop_after_model = _FakeModel()
        install_queued_message_notice_hook(stop_after_model, notice_text=QUEUED_MESSAGE_NOTICE_TEXT)
        stop_after_model.format_function_call_results(
            messages=stop_after_messages,
            function_call_results=[Message(role="tool", content="done", stop_after_tool_call=True)],
        )

    with queued_message_signal_context(_StaticQueuedState(pending=True)) as next_notice_context:
        next_turn_messages = [Message(role="user", content="hello")]
        model.format_function_call_results(
            messages=next_turn_messages,
            function_call_results=[Message(role="tool", content="result")],
        )

    assert _notice_count(queued_messages) == 1
    assert queued_messages[-1].content == QUEUED_MESSAGE_NOTICE_TEXT
    assert queued_messages[-1].provider_data == {
        _QUEUED_NOTICE_MARKER_KEY: True,
        _QUEUED_NOTICE_RESPONSE_TURN_ID_KEY: notice_context.response_turn_id,
    }
    assert _notice_count(next_turn_messages) == 1
    assert next_turn_messages[-1].content == QUEUED_MESSAGE_NOTICE_TEXT
    assert next_notice_context.response_turn_id != notice_context.response_turn_id
    assert _notice_count(stop_after_messages) == 0


def test_default_queued_notice_requires_pause_handoff() -> None:
    """The default notice should require an explicit pause handoff and continuation default."""
    assert QUEUED_MESSAGE_NOTICE_TEXT == (
        "[SYSTEM NOTICE — PAUSE FOR A NEWER USER MESSAGE] A newer user message arrived in this thread "
        "while you were working. This is a PAUSE, not a cancellation of the current task. Do not make "
        "any new tool calls. End this turn now with a final text response that explicitly says a newer "
        "message arrived and states: (1) what you completed, (2) what remains unfinished, and (3) the "
        "next step. You cannot see the newer message's contents yet. If work remains, state that you "
        "intend to resume it on the next turn, subject to the newer message's instructions."
    )


def test_notice_hook_uses_configured_notice_text() -> None:
    """Queued-message notices should use the configured hidden prompt text."""
    model = _FakeModel()
    install_queued_message_notice_hook(model, notice_text="Custom queued notice.")

    with queued_message_signal_context(_StaticQueuedState(pending=True)) as notice_context:
        messages = [Message(role="user", content="hello")]
        model.format_function_call_results(
            messages=messages,
            function_call_results=[Message(role="tool", content="result")],
        )

    assert messages[-1].content == "Custom queued notice."
    assert messages[-1].provider_data == {
        _QUEUED_NOTICE_MARKER_KEY: True,
        _QUEUED_NOTICE_RESPONSE_TURN_ID_KEY: notice_context.response_turn_id,
    }


def test_notice_hook_scopes_dedupe_to_response_turn_id() -> None:
    """Tool-round dedupe should remove only notices owned by the active response."""
    model = _FakeModel()
    install_queued_message_notice_hook(model, notice_text=QUEUED_MESSAGE_NOTICE_TEXT)

    with queued_message_signal_context(_StaticQueuedState(pending=True)) as notice_context:
        prior_notice = _queued_notice_message(
            "prior-response",
            marker="persisted",
            content=QUEUED_MESSAGE_NOTICE_TEXT,
            from_history=True,
        )
        current_stale_notice = _queued_notice_message(
            notice_context.response_turn_id,
            marker="persisted",
            content=QUEUED_MESSAGE_NOTICE_TEXT,
        )
        messages = [prior_notice, current_stale_notice]
        model.format_function_call_results(
            messages=messages,
            function_call_results=[Message(role="tool", content="result")],
        )

    assert messages[0] is prior_notice
    assert _notice_count(messages, marker="persisted", response_turn_id="prior-response") == 1
    assert _notice_count(messages, marker="persisted", response_turn_id=notice_context.response_turn_id) == 0
    assert _notice_count(messages, marker=True, response_turn_id=notice_context.response_turn_id) == 1


def test_notice_hook_preserves_prior_persisted_replay() -> None:
    """A prior persisted notice should survive later tool-result formatting byte-for-byte."""
    model = _FakeModel()
    install_queued_message_notice_hook(model, notice_text=QUEUED_MESSAGE_NOTICE_TEXT)
    prior_notice = _queued_notice_message(
        "prior-response",
        marker="persisted",
        content=QUEUED_MESSAGE_NOTICE_TEXT,
        from_history=True,
    )
    original = prior_notice.model_dump()

    with queued_message_signal_context(_StaticQueuedState(pending=True)) as notice_context:
        messages = [prior_notice]
        for index in range(2):
            model.format_function_call_results(
                messages=messages,
                function_call_results=[Message(role="tool", content=f"result {index}")],
            )

    assert prior_notice.model_dump() == original
    assert _notice_count(messages, marker="persisted", response_turn_id="prior-response") == 1
    assert _notice_count(messages, marker=True, response_turn_id=notice_context.response_turn_id) == 1


def test_notice_reinjects_at_end_across_multiple_tool_rounds() -> None:
    """Repeated tool rounds should keep exactly one queued notice at the end of the prompt."""
    model = _FakeModel()
    install_queued_message_notice_hook(model, notice_text=QUEUED_MESSAGE_NOTICE_TEXT)

    with queued_message_signal_context(_StaticQueuedState(pending=True)):
        messages = [Message(role="user", content="hello")]
        for index in range(5):
            model.format_function_call_results(
                messages=messages,
                function_call_results=[Message(role="tool", content=f"result {index}")],
            )

            assert _notice_count(messages) == 1
            assert messages[-1].content == QUEUED_MESSAGE_NOTICE_TEXT


def test_queued_notice_logs_once_per_response_turn() -> None:
    """Repeated successful appends should emit one injection event for the response."""
    model = _FakeModel()
    install_queued_message_notice_hook(model, notice_text=QUEUED_MESSAGE_NOTICE_TEXT)

    with (
        capture_logs() as logs,
        queued_message_signal_context(
            _StaticQueuedState(pending=True),
        ) as notice_context,
    ):
        messages = [Message(role="user", content="hello")]
        for index in range(5):
            model.format_function_call_results(
                messages=messages,
                function_call_results=[Message(role="tool", content=f"result {index}")],
            )

    injection_logs = [entry for entry in logs if entry.get("event") == "queued_message_notice_injected"]
    assert injection_logs == [
        {
            "event": "queued_message_notice_injected",
            "log_level": "info",
            "response_turn_id": notice_context.response_turn_id,
        },
    ]

    with capture_logs() as no_injection_logs:
        with queued_message_signal_context(_StaticQueuedState(pending=False)):
            model.format_function_call_results(
                messages=[],
                function_call_results=[Message(role="tool", content="result")],
            )
        with queued_message_signal_context(_StaticQueuedState(pending=True)):
            model.format_function_call_results(
                messages=[],
                function_call_results=[Message(role="tool", content="done", stop_after_tool_call=True)],
            )

    assert not any(entry.get("event") == "queued_message_notice_injected" for entry in no_injection_logs)


def test_notice_reinjects_after_media_follow_up_message() -> None:
    """Agno appends media follow-up messages after tool formatting, so the queued notice must be reappended."""
    model = _FakeModel()
    install_queued_message_notice_hook(model, notice_text=QUEUED_MESSAGE_NOTICE_TEXT)

    with queued_message_signal_context(_StaticQueuedState(pending=True)):
        messages = [Message(role="user", content="hello")]
        function_call_results = [
            Message(
                role="tool",
                content="generated image",
                images=[Image(url="https://example.com/image.png")],
            ),
        ]
        model.format_function_call_results(
            messages=messages,
            function_call_results=function_call_results,
        )
        model._handle_function_call_media(
            messages=messages,
            function_call_results=function_call_results,
        )

    assert _notice_count(messages) == 1
    assert messages[-2].content == "Take note of the following content"
    assert messages[-1].content == QUEUED_MESSAGE_NOTICE_TEXT


def test_notice_hook_still_installs_when_media_handler_is_missing() -> None:
    """Missing media support must not disable queued notices for formatted tool results."""
    model = _FakeModelWithoutFunctionCallMedia()
    install_queued_message_notice_hook(model, notice_text=QUEUED_MESSAGE_NOTICE_TEXT)

    with queued_message_signal_context(_StaticQueuedState(pending=True)):
        messages = [Message(role="user", content="hello")]
        model.format_function_call_results(
            messages=messages,
            function_call_results=[Message(role="tool", content="result")],
        )

    assert _notice_count(messages) == 1
    assert messages[-1].content == QUEUED_MESSAGE_NOTICE_TEXT


@pytest.mark.asyncio
async def test_ai_response_preserves_stale_notice_before_prepare(tmp_path: Path) -> None:
    """Loaded session history should persist a prior crash-left live notice."""
    config = _config(tmp_path)
    storage = _FakeStorage()
    storage.session = AgentSession(
        session_id="session-1",
        runs=[
            RunOutput(
                run_id="run-0",
                session_id="session-1",
                messages=[_queued_notice_message("prior-response")],
                status=RunStatus.completed,
            ),
        ],
    )
    observed_notice_counts: list[int] = []

    async def fake_prepare(
        _ctx: object,
        *,
        scope_context: object | None = None,
        **_kwargs: object,
    ) -> _PreparedAgentRun:
        assert scope_context is not None
        session = scope_context.session
        assert session is not None
        observed_notice_counts.append(_notice_count(session.runs[0].messages or []))
        agent = MagicMock()
        agent.model = None
        agent.arun = AsyncMock(
            return_value=RunOutput(
                run_id="run-1",
                session_id="session-1",
                content="final answer",
                model="test-model",
                model_provider="openai",
                messages=[],
                status=RunStatus.completed,
                tools=[],
            ),
        )
        return _prepared_run(agent)

    with (
        patch(
            "mindroom.ai.open_resolved_scope_session_context",
            side_effect=lambda **_kwargs: _open_scope(storage),
        ),
        patch("mindroom.ai._prepare_agent_and_prompt", new=AsyncMock(side_effect=fake_prepare)),
        patch("mindroom.ai.close_agent_runtime_state_dbs"),
    ):
        response = await ai_response(
            make_turn_context("general", session_id="session-1"),
            prompt="hello",
            runtime_paths=runtime_paths_for(config),
            config=config,
        )

    assert response == "final answer"
    assert observed_notice_counts == [1]
    assert storage.upserted is True
    assert storage.session is not None
    recovered = _notice_messages(
        storage.session.runs[0].messages or [],
        marker="persisted",
        response_turn_id="prior-response",
    )
    assert [message.content for message in recovered] == [QUEUED_MESSAGE_NOTICE_TEXT]


@pytest.mark.asyncio
async def test_ai_response_preserves_notice_in_session(tmp_path: Path) -> None:
    """Attempt registration should stay live until outer non-stream finalization."""
    config = _config(tmp_path)
    storage = _FakeStorage()
    model = _FakeModel()
    run_output_holder: dict[str, RunOutput] = {}

    async def fake_arun(
        _prompt: str,
        *,
        session_id: str,
        **_kwargs: object,
    ) -> RunOutput:
        messages = [Message(role="user", content="hello")]
        model.format_function_call_results(
            messages=messages,
            function_call_results=[Message(role="tool", content="tool result")],
        )
        stored_messages = [message.model_copy(deep=True) for message in messages]
        storage.session = AgentSession(
            session_id=session_id,
            runs=[
                RunOutput(
                    run_id="run-1",
                    session_id=session_id,
                    messages=stored_messages,
                    status=RunStatus.completed,
                ),
            ],
        )
        run_output = RunOutput(
            run_id="run-1",
            session_id=session_id,
            content="final answer",
            model="test-model",
            model_provider="openai",
            messages=messages,
            status=RunStatus.completed,
            tools=[],
        )
        run_output_holder["run"] = run_output
        return run_output

    agent = MagicMock()
    agent.model = model
    agent.arun = AsyncMock(side_effect=fake_arun)

    with (
        patch(
            "mindroom.ai.open_resolved_scope_session_context",
            side_effect=lambda **_kwargs: _open_scope(storage),
        ),
        patch("mindroom.ai._prepare_agent_and_prompt", new=AsyncMock(return_value=_prepared_run(agent))),
        patch("mindroom.ai.close_agent_runtime_state_dbs"),
        queued_message_signal_context(_StaticQueuedState(pending=True)) as notice_context,
    ):
        response = await ai_response(
            make_turn_context("general", session_id="session-1"),
            prompt="hello",
            runtime_paths=runtime_paths_for(config),
            config=config,
        )
        assert storage.upserted is False
        assert _notice_count(run_output_holder["run"].messages or [], marker=True) == 1
        assert storage.session is not None
        assert _notice_count(storage.session.runs[0].messages or [], marker=True) == 1
        await ai_runtime.finalize_queued_notice_response_turn_async(notice_context)

    assert response == "final answer"
    assert storage.upserted is True
    assert storage.session is not None
    assert _notice_count(storage.session.runs[0].messages or [], marker=True) == 0
    assert _notice_count(storage.session.runs[0].messages or [], marker="persisted") == 1


@pytest.mark.asyncio
async def test_ai_response_preserves_notice_in_session_after_exception(tmp_path: Path) -> None:
    """An error-only response should remove its live notice at the outer boundary."""
    config = _config(tmp_path)
    storage = _FakeStorage()
    model = _FakeModel()

    async def fake_arun(
        _prompt: str,
        *,
        session_id: str,
        **_kwargs: object,
    ) -> RunOutput:
        messages = [Message(role="user", content="hello")]
        model.format_function_call_results(
            messages=messages,
            function_call_results=[Message(role="tool", content="tool result")],
        )
        storage.session = AgentSession(
            session_id=session_id,
            runs=[
                RunOutput(
                    run_id="run-1",
                    session_id=session_id,
                    messages=messages,
                    status=RunStatus.error,
                ),
            ],
        )
        error_message = "boom"
        raise RuntimeError(error_message)

    agent = MagicMock()
    agent.model = model
    agent.arun = AsyncMock(side_effect=fake_arun)

    with (
        patch(
            "mindroom.ai.open_resolved_scope_session_context",
            side_effect=lambda **_kwargs: _open_scope(storage),
        ),
        patch("mindroom.ai._prepare_agent_and_prompt", new=AsyncMock(return_value=_prepared_run(agent))),
        patch("mindroom.ai.close_agent_runtime_state_dbs"),
        queued_message_signal_context(_StaticQueuedState(pending=True)) as notice_context,
    ):
        response = await ai_response(
            make_turn_context("general", session_id="session-1"),
            prompt="hello",
            runtime_paths=runtime_paths_for(config),
            config=config,
        )
        assert storage.session is not None
        assert _notice_count(storage.session.runs[0].messages or [], marker=True) == 1
        await ai_runtime.finalize_queued_notice_response_turn_async(notice_context)

    assert isinstance(response, str)
    assert storage.upserted is True
    assert storage.session is not None
    assert _notice_count(storage.session.runs[0].messages or []) == 0


@pytest.mark.asyncio
async def test_stream_agent_response_preserves_notice_in_session(tmp_path: Path) -> None:
    """Streaming attempt cleanup should wait for the outer response boundary."""
    config = _config(tmp_path)
    storage = _FakeStorage()
    model = _FakeModel()

    async def fake_stream(
        _prompt: str,
        *,
        session_id: str,
        **_kwargs: object,
    ) -> AsyncIterator[object]:
        messages = [Message(role="user", content="hello")]
        model.format_function_call_results(
            messages=messages,
            function_call_results=[Message(role="tool", content="tool result")],
        )
        stored_messages = [message.model_copy(deep=True) for message in messages]
        storage.session = AgentSession(
            session_id=session_id,
            runs=[
                RunOutput(
                    run_id="run-1",
                    session_id=session_id,
                    messages=stored_messages,
                    status=RunStatus.completed,
                ),
            ],
        )
        yield RunContentEvent(content="chunk")
        yield RunCompletedEvent(run_id="run-1", session_id=session_id)

    agent = MagicMock()
    agent.model = model
    agent.arun = fake_stream

    with (
        patch(
            "mindroom.ai.open_resolved_scope_session_context",
            side_effect=lambda **_kwargs: _open_scope(storage),
        ),
        patch("mindroom.ai._prepare_agent_and_prompt", new=AsyncMock(return_value=_prepared_run(agent))),
        patch("mindroom.ai.close_agent_runtime_state_dbs"),
        queued_message_signal_context(_StaticQueuedState(pending=True)) as notice_context,
    ):
        chunks = [
            chunk
            async for chunk in stream_agent_response(
                make_turn_context("general", session_id="session-1"),
                prompt="hello",
                runtime_paths=runtime_paths_for(config),
                config=config,
            )
        ]
        assert storage.upserted is False
        assert storage.session is not None
        assert _notice_count(storage.session.runs[0].messages or [], marker=True) == 1
        await ai_runtime.finalize_queued_notice_response_turn_async(notice_context)

    assert any(isinstance(chunk, RunContentEvent) and chunk.content == "chunk" for chunk in chunks)
    assert storage.upserted is True
    assert storage.session is not None
    assert _notice_count(storage.session.runs[0].messages or [], marker=True) == 0
    assert _notice_count(storage.session.runs[0].messages or [], marker="persisted") == 1


def test_create_team_instance_installs_notice_hook_on_team_model(tmp_path: Path) -> None:
    """Team coordinator models should receive the same queued-message notice hook."""
    config = _config(tmp_path)
    runtime_paths = runtime_paths_for(config)
    model = _FakeModel()
    agent = AgnoAgent(id="general", name="General", model="openai:test-model")

    with (
        open_bound_scope_session_context(
            agents=[agent],
            session_id="session-1",
            runtime_paths=runtime_paths,
            config=config,
            execution_identity=None,
        ) as scope_context,
        patch("mindroom.model_loading.get_model_instance", return_value=model),
        patch("mindroom.teams.Team", side_effect=lambda **kwargs: SimpleNamespace(model=kwargs["model"])),
        queued_message_signal_context(_StaticQueuedState(pending=True)),
    ):
        assert scope_context is not None
        team = _create_team_instance(
            agents=[agent],
            mode=TeamMode.COORDINATE,
            config=config,
            runtime_paths=runtime_paths,
            team_display_name="Queued Notice Team",
            scope_context=scope_context,
            execution_identity=None,
            model_name="default",
        )
        messages = [Message(role="user", content="hello")]
        team.model.format_function_call_results(
            messages=messages,
            function_call_results=[Message(role="tool", content="tool result")],
        )
        assert _notice_count(messages) == 1


@pytest.mark.asyncio
async def test_response_finalization_collapses_nested_current_notice_state() -> None:
    """Outer finalization should leave one exact notice in the top-level completed run."""
    created_storages: list[_FakeStorage] = []
    model = _FakeModel()
    install_queued_message_notice_hook(model, notice_text=QUEUED_MESSAGE_NOTICE_TEXT)

    with queued_message_signal_context(_StaticQueuedState(pending=True)) as notice_context:
        messages: list[Message] = []
        model.format_function_call_results(
            messages=messages,
            function_call_results=[Message(role="tool", content="result")],
        )
        run_output = TeamRunOutput(
            run_id="run-1",
            session_id="session-1",
            messages=[message.model_copy(deep=True) for message in messages],
            member_responses=[
                RunOutput(
                    run_id="member-run-1",
                    session_id="session-1",
                    messages=[message.model_copy(deep=True) for message in messages],
                    status=RunStatus.completed,
                ),
            ],
            status=RunStatus.completed,
        )

        def storage_factory() -> _FakeStorage:
            storage = _FakeStorage()
            storage.session = TeamSession(
                session_id="session-1",
                runs=[
                    TeamRunOutput(
                        run_id="run-1",
                        session_id="session-1",
                        messages=[message.model_copy(deep=True) for message in messages],
                        member_responses=[
                            RunOutput(
                                run_id="member-run-1",
                                session_id="session-1",
                                messages=[message.model_copy(deep=True) for message in messages],
                                status=RunStatus.completed,
                            ),
                        ],
                        status=RunStatus.completed,
                    ),
                ],
            )
            created_storages.append(storage)
            return storage

        register_queued_notice_storage(
            storage_factory=storage_factory,
            session_id="session-1",
            session_type=SessionType.TEAM,
            entity_name="queued-notice-team",
        )
        assert _notice_count(run_output.messages or [], marker=True) == 1
        await ai_runtime.finalize_queued_notice_response_turn_async(notice_context)

    storage = created_storages[0]
    assert storage.upserted is True
    assert storage.session is not None
    stored_team_run = storage.session.runs[0]
    assert isinstance(stored_team_run, TeamRunOutput)
    assert _notice_count(stored_team_run.messages or [], marker=True) == 0
    assert _notice_count(stored_team_run.messages or [], marker="persisted") == 1
    assert stored_team_run.member_responses is not None
    stored_member_run = stored_team_run.member_responses[0]
    assert isinstance(stored_member_run, RunOutput)
    assert _notice_count(stored_member_run.messages or []) == 0
    assert storage.closed is True


@pytest.mark.asyncio
async def test_response_finalization_does_not_copy_notice_to_untouched_target() -> None:
    """Finalization should persist a notice only in storage that contains its live marker."""
    untouched_storage = _FakeStorage()
    noticed_storage = _FakeStorage()
    model = _FakeModel()
    install_queued_message_notice_hook(model, notice_text=QUEUED_MESSAGE_NOTICE_TEXT)

    with queued_message_signal_context(_StaticQueuedState(pending=True)) as notice_context:
        noticed_messages = [Message(role="user", content="Outer request")]
        model.format_function_call_results(
            messages=noticed_messages,
            function_call_results=[Message(role="tool", content="Outer tool result")],
        )
        untouched_run = RunOutput(
            run_id="untouched-run",
            session_id="shared-session",
            messages=[
                Message(role="user", content="Delegated request"),
                Message(role="assistant", content="Delegated answer"),
            ],
            status=RunStatus.completed,
        )
        noticed_run = RunOutput(
            run_id="noticed-run",
            session_id="shared-session",
            messages=noticed_messages,
            status=RunStatus.completed,
        )
        untouched_storage.session = AgentSession(
            session_id="shared-session",
            runs=[untouched_run],
        )
        noticed_storage.session = AgentSession(
            session_id="shared-session",
            runs=[noticed_run],
        )
        register_queued_notice_storage(
            storage_factory=lambda: untouched_storage,
            session_id="shared-session",
            session_type=SessionType.AGENT,
            entity_name="delegated-agent",
        )
        register_queued_notice_storage(
            storage_factory=lambda: noticed_storage,
            session_id="shared-session",
            session_type=SessionType.AGENT,
            entity_name="outer-agent",
        )

        await ai_runtime.finalize_queued_notice_response_turn_async(notice_context)

    assert untouched_storage.upsert_count == 0
    assert untouched_storage.session is not None
    assert [(message.role, message.content) for message in untouched_run.messages or []] == [
        ("user", "Delegated request"),
        ("assistant", "Delegated answer"),
    ]
    assert noticed_storage.upsert_count == 1
    assert _notice_count(noticed_run.messages or [], marker=True) == 0
    assert _notice_count(noticed_run.messages or [], marker="persisted") == 1


@pytest.mark.asyncio
async def test_response_finalization_uses_newest_completed_same_turn_run_and_original_notice_position() -> None:
    """Two completed attempts should leave one exact notice at the newest notice position."""
    model = _FakeModel()
    install_queued_message_notice_hook(model, notice_text=QUEUED_MESSAGE_NOTICE_TEXT)
    storage = _FakeStorage()

    with queued_message_signal_context(_StaticQueuedState(pending=True)) as notice_context:
        first_messages = [
            Message(role="user", content="First request"),
            Message(role="assistant", content="First tool call"),
        ]
        model.format_function_call_results(
            messages=first_messages,
            function_call_results=[Message(role="tool", content="First result")],
        )
        first_messages.append(Message(role="assistant", content="First final"))
        second_messages = [
            Message(role="user", content="Second request"),
            Message(role="assistant", content="Second tool call"),
        ]
        model.format_function_call_results(
            messages=second_messages,
            function_call_results=[Message(role="tool", content="Second result")],
        )
        second_messages.append(Message(role="assistant", content="Second final"))
        first_run = RunOutput(
            run_id="first-completed-run",
            session_id="session-1",
            messages=first_messages,
            status=RunStatus.completed,
        )
        second_run = RunOutput(
            run_id="second-completed-run",
            session_id="session-1",
            messages=second_messages,
            status=RunStatus.completed,
        )
        storage.session = AgentSession(
            session_id="session-1",
            runs=[
                RunOutput(
                    run_id=first_run.run_id,
                    session_id=first_run.session_id,
                    messages=[message.model_copy(deep=True) for message in first_messages],
                    status=first_run.status,
                ),
                RunOutput(
                    run_id=second_run.run_id,
                    session_id=second_run.session_id,
                    messages=[message.model_copy(deep=True) for message in second_messages],
                    status=second_run.status,
                ),
            ],
        )
        register_queued_notice_storage(
            storage_factory=lambda: storage,
            session_id="session-1",
            session_type=SessionType.AGENT,
            entity_name="general",
        )

        await ai_runtime.finalize_queued_notice_response_turn_async(notice_context)

    assert storage.upsert_count == 1
    assert storage.session is not None
    stored_first, stored_second = storage.session.runs
    assert _notice_count(stored_first.messages or []) == 0
    assert [(message.role, message.content) for message in stored_second.messages or []] == [
        ("user", "Second request"),
        ("assistant", "Second tool call"),
        ("tool", "Second result"),
        ("user", QUEUED_MESSAGE_NOTICE_TEXT),
        ("assistant", "Second final"),
    ]
    assert (
        _notice_count(
            stored_second.messages or [],
            marker="persisted",
            response_turn_id=notice_context.response_turn_id,
        )
        == 1
    )


@pytest.mark.asyncio
async def test_async_cleanup_keeps_session_storage_io_off_event_loop() -> None:
    """Slow response-boundary storage finalization must stay off the event loop."""
    request_started = threading.Event()
    release_request = threading.Event()
    created_storages: list[_BlockingStorage] = []
    event_loop_thread_id = threading.get_ident()

    class _BlockingStorage(_FakeStorage):
        def __init__(self) -> None:
            super().__init__()
            self.created_thread_id = threading.get_ident()

        def get_session(self, session_id: str, _session_type: object) -> AgentSession | TeamSession | None:
            request_started.set()
            release_request.wait(timeout=1)
            return super().get_session(session_id, _session_type)

    model = _FakeModel()
    install_queued_message_notice_hook(model, notice_text=QUEUED_MESSAGE_NOTICE_TEXT)
    with queued_message_signal_context(_StaticQueuedState(pending=True)) as notice_context:
        messages: list[Message] = []
        model.format_function_call_results(
            messages=messages,
            function_call_results=[Message(role="tool", content="result")],
        )

        def storage_factory() -> _BlockingStorage:
            storage = _BlockingStorage()
            storage.session = AgentSession(
                session_id="session-1",
                runs=[
                    RunOutput(
                        run_id="run-1",
                        session_id="session-1",
                        messages=[message.model_copy(deep=True) for message in messages],
                        status=RunStatus.completed,
                    ),
                ],
            )
            created_storages.append(storage)
            return storage

        register_queued_notice_storage(
            storage_factory=storage_factory,
            session_id="session-1",
            session_type=SessionType.AGENT,
            entity_name="queued-notice-agent",
        )
        cleanup_task = asyncio.create_task(
            ai_runtime.finalize_queued_notice_response_turn_async(notice_context),
        )
        try:
            assert await asyncio.to_thread(request_started.wait, 1)
            await asyncio.wait_for(asyncio.sleep(0), timeout=0.1)
            assert not cleanup_task.done()
        finally:
            release_request.set()

        await asyncio.wait_for(cleanup_task, timeout=1)
    storage = created_storages[0]
    assert storage.created_thread_id != event_loop_thread_id
    assert storage.upserted is True
    assert storage.closed is True


@pytest.mark.asyncio
async def test_response_turn_finalization_completes_when_cancelled_during_storage_io() -> None:
    """Cancellation during storage I/O must not leave a durable live imperative behind."""
    request_started = threading.Event()
    release_request = threading.Event()
    request_finished = threading.Event()
    model = _FakeModel()
    install_queued_message_notice_hook(model, notice_text=QUEUED_MESSAGE_NOTICE_TEXT)

    class _BlockingStorage(_FakeStorage):
        def get_session(self, session_id: str, _session_type: object) -> AgentSession | TeamSession | None:
            request_started.set()
            release_request.wait(timeout=1)
            return super().get_session(session_id, _session_type)

        def close(self) -> None:
            super().close()
            request_finished.set()

    with queued_message_signal_context(_StaticQueuedState(pending=True)) as notice_context:
        messages: list[Message] = []
        model.format_function_call_results(
            messages=messages,
            function_call_results=[Message(role="tool", content="result")],
        )
        run_output = RunOutput(
            run_id="completed-run",
            session_id="session-1",
            messages=messages,
            status=RunStatus.completed,
        )
        storage = _BlockingStorage()
        storage.session = AgentSession(
            session_id="session-1",
            runs=[
                RunOutput(
                    run_id=run_output.run_id,
                    session_id=run_output.session_id,
                    messages=[message.model_copy(deep=True) for message in messages],
                    status=run_output.status,
                ),
            ],
        )
        register_queued_notice_storage(
            storage_factory=lambda: storage,
            session_id="session-1",
            session_type=SessionType.AGENT,
            entity_name="general",
        )
        finalization_task = asyncio.create_task(
            ai_runtime.finalize_queued_notice_response_turn_async(notice_context),
        )
        assert await asyncio.to_thread(request_started.wait, 1)
        finalization_task.cancel()
        release_request.set()
        with pytest.raises(asyncio.CancelledError):
            await finalization_task
        assert await asyncio.to_thread(request_finished.wait, 1)

    assert storage.session is not None
    assert _notice_count(storage.session.runs[0].messages or [], marker=True) == 0
    assert (
        _notice_count(
            storage.session.runs[0].messages or [],
            marker="persisted",
            response_turn_id=notice_context.response_turn_id,
        )
        == 1
    )


@pytest.mark.asyncio
async def test_response_turn_finalization_defers_repeated_cancellation_during_storage_io() -> None:
    """Repeated cancellation must not release the caller before durable finalization finishes."""
    request_started = threading.Event()
    release_request = threading.Event()
    request_finished = threading.Event()
    first_cancel_observed = asyncio.Event()
    original_shield = asyncio.shield
    model = _FakeModel()
    install_queued_message_notice_hook(model, notice_text=QUEUED_MESSAGE_NOTICE_TEXT)

    class _BlockingStorage(_FakeStorage):
        def get_session(self, session_id: str, _session_type: object) -> AgentSession | TeamSession | None:
            request_started.set()
            release_request.wait(timeout=1)
            return super().get_session(session_id, _session_type)

        def close(self) -> None:
            super().close()
            request_finished.set()

    async def observed_shield(task: asyncio.Task[None]) -> None:
        try:
            await original_shield(task)
        except asyncio.CancelledError:
            first_cancel_observed.set()
            raise

    with queued_message_signal_context(_StaticQueuedState(pending=True)) as notice_context:
        messages: list[Message] = []
        model.format_function_call_results(
            messages=messages,
            function_call_results=[Message(role="tool", content="result")],
        )
        run_output = RunOutput(
            run_id="completed-run",
            session_id="session-1",
            messages=messages,
            status=RunStatus.completed,
        )
        storage = _BlockingStorage()
        storage.session = AgentSession(
            session_id="session-1",
            runs=[
                RunOutput(
                    run_id=run_output.run_id,
                    session_id=run_output.session_id,
                    messages=[message.model_copy(deep=True) for message in messages],
                    status=run_output.status,
                ),
            ],
        )
        register_queued_notice_storage(
            storage_factory=lambda: storage,
            session_id="session-1",
            session_type=SessionType.AGENT,
            entity_name="general",
        )
        with patch.object(ai_runtime.asyncio, "shield", new=observed_shield):
            finalization_task = asyncio.create_task(
                ai_runtime.finalize_queued_notice_response_turn_async(notice_context),
            )
            assert await asyncio.to_thread(request_started.wait, 1)
            finalization_task.cancel()
            await asyncio.wait_for(first_cancel_observed.wait(), timeout=1)
            finalization_task.cancel()
            completed, _ = await asyncio.wait({finalization_task}, timeout=0.05)
            try:
                assert finalization_task not in completed
            finally:
                release_request.set()
            with pytest.raises(asyncio.CancelledError):
                await finalization_task

    assert request_finished.is_set()
    assert storage.session is not None
    assert _notice_count(storage.session.runs[0].messages or [], marker=True) == 0
    assert (
        _notice_count(
            storage.session.runs[0].messages or [],
            marker="persisted",
            response_turn_id=notice_context.response_turn_id,
        )
        == 1
    )


@pytest.mark.asyncio
async def test_storage_registration_deduplicates_one_session_across_fresh_factories() -> None:
    """Fresh storage-factory closures for one session should cause one boundary open."""
    model = _FakeModel()
    install_queued_message_notice_hook(model, notice_text=QUEUED_MESSAGE_NOTICE_TEXT)
    factory_calls = 0

    with queued_message_signal_context(_StaticQueuedState(pending=True)) as notice_context:
        messages: list[Message] = []
        model.format_function_call_results(
            messages=messages,
            function_call_results=[Message(role="tool", content="result")],
        )
        run_output = RunOutput(
            run_id="completed-run",
            session_id="session-1",
            messages=messages,
            status=RunStatus.completed,
        )
        storage = _FakeStorage()
        storage.session = AgentSession(
            session_id="session-1",
            runs=[
                RunOutput(
                    run_id=run_output.run_id,
                    session_id=run_output.session_id,
                    messages=[message.model_copy(deep=True) for message in messages],
                    status=run_output.status,
                ),
            ],
        )

        def fresh_storage_factory() -> Callable[[], _FakeStorage]:
            def storage_factory() -> _FakeStorage:
                nonlocal factory_calls
                factory_calls += 1
                return storage

            return storage_factory

        for _ in range(2):
            register_queued_notice_storage(
                storage_factory=fresh_storage_factory(),
                session_id="session-1",
                session_type=SessionType.AGENT,
                entity_name="general",
            )
        await ai_runtime.finalize_queued_notice_response_turn_async(notice_context)

    assert factory_calls == 1


@pytest.mark.asyncio
async def test_error_responses_never_attach_notices_to_prior_completed_run() -> None:
    """Failed responses must not accumulate their notice facts on an earlier turn."""
    model = _FakeModel()
    install_queued_message_notice_hook(model, notice_text=QUEUED_MESSAGE_NOTICE_TEXT)
    storage = _FakeStorage()
    prior_completed = RunOutput(
        run_id="prior-completed",
        session_id="session-1",
        messages=[
            Message(role="user", content="Prior request"),
            Message(role="assistant", content="Prior answer"),
        ],
        status=RunStatus.completed,
    )
    storage.session = AgentSession(session_id="session-1", runs=[prior_completed])

    for index in range(3):
        with queued_message_signal_context(_StaticQueuedState(pending=True)) as notice_context:
            messages: list[Message] = []
            model.format_function_call_results(
                messages=messages,
                function_call_results=[Message(role="tool", content=f"failed result {index}")],
            )
            error_run = RunOutput(
                run_id=f"error-run-{index}",
                session_id="session-1",
                messages=messages,
                status=RunStatus.error,
            )
            assert storage.session.runs is not None
            storage.session.runs.append(
                RunOutput(
                    run_id=error_run.run_id,
                    session_id=error_run.session_id,
                    messages=[message.model_copy(deep=True) for message in messages],
                    status=error_run.status,
                ),
            )
            register_queued_notice_storage(
                storage_factory=lambda: storage,
                session_id="session-1",
                session_type=SessionType.AGENT,
                entity_name="general",
            )
            await ai_runtime.finalize_queued_notice_response_turn_async(notice_context)

    assert storage.session.runs is not None
    assert _notice_count(prior_completed.messages or []) == 0
    assert (
        sum(
            _notice_count(run.messages or [])
            for run in storage.session.runs
            if isinstance(run, RunOutput | TeamRunOutput)
        )
        == 0
    )


@pytest.mark.asyncio
async def test_response_lifecycle_finalizes_notice_after_locked_operation() -> None:
    """The lifecycle boundary should persist the exact notice shown to the model."""
    lifecycle = ResponseLifecycleCoordinator()
    target = MessageTarget.resolve("!room:localhost", "$thread", "$event")
    envelope = _envelope(source_event_id="$event", target=target)
    storage = _FakeStorage()
    model = _FakeModel()
    notice_text = "Custom queued notice."
    install_queued_message_notice_hook(model, notice_text=notice_text)
    observed_live_counts: list[int] = []

    async def locked_operation(_target: MessageTarget) -> str:
        lifecycle._get_or_create_queued_signal(target).add_waiting_human_message("$newer")
        messages: list[Message] = []
        model.format_function_call_results(
            messages=messages,
            function_call_results=[Message(role="tool", content="result")],
        )
        run = RunOutput(
            run_id="completed-run",
            session_id="session-1",
            messages=messages,
            status=RunStatus.completed,
        )
        storage.session = AgentSession(
            session_id="session-1",
            runs=[
                RunOutput(
                    run_id=run.run_id,
                    session_id=run.session_id,
                    messages=[message.model_copy(deep=True) for message in messages],
                    status=run.status,
                ),
            ],
        )
        register_queued_notice_storage(
            storage_factory=lambda: storage,
            session_id="session-1",
            session_type=SessionType.AGENT,
            entity_name="general",
        )
        observed_live_counts.append(_notice_count(run.messages or [], marker=True))
        return "$response"

    result = await lifecycle.run_locked_response(
        target=target,
        response_envelope=envelope,
        queued_notice_reservation=None,
        pipeline_timing=None,
        locked_operation=locked_operation,
    )

    assert result == "$response"
    assert observed_live_counts == [1]
    assert storage.session is not None
    stored_run = storage.session.runs[0]
    assert _notice_count(stored_run.messages or [], marker=True) == 0
    persisted_notices = _notice_messages(stored_run.messages or [], marker="persisted")
    assert [message.content for message in persisted_notices] == [notice_text]
    assert storage.upsert_count == 1


@pytest.mark.asyncio
async def test_response_lifecycle_finalizes_error_state_when_operation_raises() -> None:
    """The lifecycle finally block should remove live state when the operation raises."""
    lifecycle = ResponseLifecycleCoordinator()
    target = MessageTarget.resolve("!room:localhost", "$thread", "$event")
    envelope = _envelope(source_event_id="$event", target=target)
    storage = _FakeStorage()
    model = _FakeModel()
    install_queued_message_notice_hook(model, notice_text=QUEUED_MESSAGE_NOTICE_TEXT)

    async def locked_operation(_target: MessageTarget) -> str:
        lifecycle._get_or_create_queued_signal(target).add_waiting_human_message("$newer")
        messages: list[Message] = []
        model.format_function_call_results(
            messages=messages,
            function_call_results=[Message(role="tool", content="failed result")],
        )
        storage.session = AgentSession(
            session_id="session-1",
            runs=[
                RunOutput(
                    run_id="error-run",
                    session_id="session-1",
                    messages=messages,
                    status=RunStatus.error,
                ),
            ],
        )
        register_queued_notice_storage(
            storage_factory=lambda: storage,
            session_id="session-1",
            session_type=SessionType.AGENT,
            entity_name="general",
        )
        error_message = "provider failed"
        raise RuntimeError(error_message)

    with pytest.raises(RuntimeError, match="provider failed"):
        await lifecycle.run_locked_response(
            target=target,
            response_envelope=envelope,
            queued_notice_reservation=None,
            pipeline_timing=None,
            locked_operation=locked_operation,
        )

    assert storage.session is not None
    assert _notice_count(storage.session.runs[0].messages or []) == 0


@pytest.mark.asyncio
async def test_response_turn_finalization_is_idempotent_for_one_persisted_notice() -> None:
    """Repeated boundary finalization should not rewrite an already-normalized response."""
    storage = _FakeStorage()
    model = _FakeModel()
    install_queued_message_notice_hook(model, notice_text=QUEUED_MESSAGE_NOTICE_TEXT)

    with queued_message_signal_context(_StaticQueuedState(pending=True)) as notice_context:
        injected_messages: list[Message] = []
        model.format_function_call_results(
            messages=injected_messages,
            function_call_results=[Message(role="tool", content="result")],
        )
        persisted_notice = _queued_notice_message(
            notice_context.response_turn_id,
            marker="persisted",
            content=QUEUED_MESSAGE_NOTICE_TEXT,
        )
        run = RunOutput(
            run_id="completed-run",
            session_id="session-1",
            messages=[persisted_notice],
            status=RunStatus.completed,
        )
        storage.session = AgentSession(
            session_id="session-1",
            runs=[
                RunOutput(
                    run_id=run.run_id,
                    session_id=run.session_id,
                    messages=[persisted_notice.model_copy(deep=True)],
                    status=run.status,
                ),
            ],
        )
        original_stored = (storage.session.runs[0].messages or [])[0].model_dump()
        register_queued_notice_storage(
            storage_factory=lambda: storage,
            session_id="session-1",
            session_type=SessionType.AGENT,
            entity_name="general",
        )
        await ai_runtime.finalize_queued_notice_response_turn_async(notice_context)
        await ai_runtime.finalize_queued_notice_response_turn_async(notice_context)

    assert storage.upsert_count == 0
    assert storage.session is not None
    assert (storage.session.runs[0].messages or [])[0].model_dump() == original_stored


def test_pre_replay_crash_recovery_preserves_prior_persisted_and_active_live_notice() -> None:
    """Recovery should persist prior live state without touching prior records or the active response."""
    storage = _FakeStorage()

    with queued_message_signal_context(_StaticQueuedState(pending=False)) as notice_context:
        prior_persisted = _queued_notice_message(
            "already-persisted",
            marker="persisted",
            content=QUEUED_MESSAGE_NOTICE_TEXT,
        )
        prior_dump = prior_persisted.model_dump()
        active_live = _queued_notice_message(notice_context.response_turn_id)
        storage.session = AgentSession(
            session_id="session-1",
            runs=[
                RunOutput(
                    run_id="completed-run",
                    session_id="session-1",
                    messages=[
                        _queued_notice_message("crash-left"),
                        prior_persisted,
                        active_live,
                    ],
                ),
            ],
        )
        ai_runtime.scrub_queued_notice_session_context(
            scope_context=SimpleNamespace(
                storage=storage,
                session=storage.session,
            ),
            entity_name="general",
        )

    assert storage.session is not None
    stored_messages = storage.session.runs[0].messages or []
    assert prior_persisted.model_dump() == prior_dump
    assert _notice_count(stored_messages, marker="persisted", response_turn_id="already-persisted") == 1
    assert _notice_count(stored_messages, marker="persisted", response_turn_id="crash-left") == 1
    assert _notice_count(stored_messages, marker=True, response_turn_id="crash-left") == 0
    assert _notice_count(stored_messages, marker=True, response_turn_id=notice_context.response_turn_id) == 1


def test_notice_provider_data_survives_mindroom_sqlite_round_trip(tmp_path: Path) -> None:
    """MindRoom's prompt-sanitizing SQLite storage should preserve notice ownership metadata."""
    response_turn_id = "response-1"
    storage = _queued_notice_storage(tmp_path)
    storage.upsert_session(
        AgentSession(
            session_id="session-1",
            agent_id="general",
            runs=[
                RunOutput(
                    run_id="run-1",
                    agent_id="general",
                    session_id="session-1",
                    messages=[_queued_notice_message(response_turn_id)],
                    status=RunStatus.completed,
                ),
            ],
            created_at=1,
            updated_at=1,
        ),
    )
    storage.close()

    reopened_storage = _queued_notice_storage(tmp_path)
    try:
        session = get_agent_session(reopened_storage, "session-1")
    finally:
        reopened_storage.close()

    assert session is not None
    assert session.runs is not None
    stored_run = session.runs[0]
    assert isinstance(stored_run, RunOutput)
    stored_notice = (stored_run.messages or [])[0]
    assert stored_notice.content == QUEUED_MESSAGE_NOTICE_TEXT
    assert stored_notice.provider_data == {
        _QUEUED_NOTICE_MARKER_KEY: True,
        _QUEUED_NOTICE_RESPONSE_TURN_ID_KEY: response_turn_id,
    }
    assert stored_notice.from_history is False


async def _persist_notice_bearing_response(tmp_path: Path) -> str:
    lifecycle = ResponseLifecycleCoordinator()
    target = MessageTarget.resolve("!room:localhost", "$thread", "$event-1")
    envelope = _envelope(source_event_id="$event-1", target=target)
    response_1_model = RecordingModel(id="response-1-model", provider="fake")
    install_queued_message_notice_hook(response_1_model, notice_text=QUEUED_MESSAGE_NOTICE_TEXT)
    response_1_ids: list[str] = []

    async def response_1_operation(_target: MessageTarget) -> str:
        lifecycle._get_or_create_queued_signal(target).add_waiting_human_message("$newer")
        messages = [Message(role="user", content="First request")]
        response_1_model.format_function_call_results(
            messages=messages,
            function_call_results=[Message(role="tool", content="First tool result")],
        )
        notices = _notice_messages(messages, marker=True)
        assert len(notices) == 1
        provider_data = notices[0].provider_data
        assert isinstance(provider_data, dict)
        response_turn_id = provider_data[_QUEUED_NOTICE_RESPONSE_TURN_ID_KEY]
        assert isinstance(response_turn_id, str)
        response_1_ids.append(response_turn_id)
        messages.append(Message(role="assistant", content="First response handoff"))
        run_output = RunOutput(
            run_id="run-1",
            agent_id="general",
            session_id="session-1",
            messages=messages,
            status=RunStatus.completed,
        )
        storage = _queued_notice_storage(tmp_path)
        try:
            storage.upsert_session(
                AgentSession(
                    session_id="session-1",
                    agent_id="general",
                    runs=[run_output],
                    created_at=1,
                    updated_at=1,
                ),
            )
        finally:
            storage.close()
        register_queued_notice_storage(
            storage_factory=lambda: _queued_notice_storage(tmp_path),
            session_id="session-1",
            session_type=SessionType.AGENT,
            entity_name="general",
        )
        return "First response handoff"

    response_1 = await lifecycle.run_locked_response(
        target=target,
        response_envelope=envelope,
        queued_notice_reservation=None,
        pipeline_timing=None,
        locked_operation=response_1_operation,
    )

    assert response_1 == "First response handoff"
    assert len(response_1_ids) == 1
    return response_1_ids[0]


@pytest.mark.asyncio
async def test_prior_notice_survives_actual_next_provider_request_and_tool_round(
    tmp_path: Path,
) -> None:
    """The exact notice should survive real SQLite replay and next-response formatting."""
    response_1_id = await _persist_notice_bearing_response(tmp_path)

    response_2_storage = _queued_notice_storage(tmp_path)
    recording_model = RecordingModel(id="response-2-model", provider="fake")
    install_queued_message_notice_hook(recording_model, notice_text=QUEUED_MESSAGE_NOTICE_TEXT)
    response_2_agent = AgnoAgent(
        id="general",
        name="General",
        model=recording_model,
        db=response_2_storage,
        add_history_to_context=True,
        num_history_runs=10,
        store_history_messages=False,
    )
    try:
        with queued_message_signal_context(_StaticQueuedState(pending=False)) as response_2_context:
            response_2 = await response_2_agent.arun("Second request", session_id="session-1")
            prior = next(
                message for message in recording_model.seen_messages if message.content == QUEUED_MESSAGE_NOTICE_TEXT
            )
            assert prior.from_history is True
            assert prior.provider_data == {
                _QUEUED_NOTICE_MARKER_KEY: "persisted",
                _QUEUED_NOTICE_RESPONSE_TURN_ID_KEY: response_1_id,
            }
            assert [
                (message.role, message.content)
                for message in recording_model.seen_messages
                if message.content
                in {
                    QUEUED_MESSAGE_NOTICE_TEXT,
                    "First response handoff",
                    "Second request",
                }
            ] == [
                ("user", QUEUED_MESSAGE_NOTICE_TEXT),
                ("assistant", "First response handoff"),
                ("user", "Second request"),
            ]
            original_prior = prior.model_dump()

            recording_model.format_function_call_results(
                messages=recording_model.seen_messages,
                function_call_results=[Message(role="tool", content="Second tool result")],
            )

            assert prior.model_dump() == original_prior
            assert (
                _notice_count(
                    recording_model.seen_messages,
                    marker="persisted",
                    response_turn_id=response_1_id,
                )
                == 1
            )
            assert (
                _notice_count(
                    recording_model.seen_messages,
                    response_turn_id=response_2_context.response_turn_id,
                )
                == 0
            )

        assert response_2.content == "ok"
        persisted = get_agent_session(response_2_storage, "session-1")
    finally:
        response_2_storage.close()

    assert persisted is not None
    assert persisted.runs is not None
    response_2_run = persisted.runs[-1]
    assert isinstance(response_2_run, RunOutput)
    assert all(message.content != QUEUED_MESSAGE_NOTICE_TEXT for message in response_2_run.messages or [])
    response_1_run = persisted.runs[0]
    assert isinstance(response_1_run, RunOutput)
    persisted_notices = _notice_messages(
        response_1_run.messages or [],
        marker="persisted",
        response_turn_id=response_1_id,
    )
    assert [message.content for message in persisted_notices] == [QUEUED_MESSAGE_NOTICE_TEXT]
