"""Focused unit tests for TurnController built from its constructor-injected collaborators.

These tests construct ``TurnController`` directly from ``TurnControllerDeps`` —
no ``AgentBot``/orchestrator boot — with real ``TurnPolicy``, ``ConversationResolver``,
``TurnStore`` (disk-backed ledger), and ``CoalescingGate`` collaborators, and typed
recording fakes only at the execution and delivery seams (``ResponseRunner``,
``DeliveryGateway``). They pin the turn contract from
``docs/architecture/bot-runtime.md``: precheck filtering (via ``IngressValidator``),
replay-guard rejection, policy decision routing, ingress-level command dispatch
with terminal TurnStore outcomes (commands are control inputs that never enter
the coalescing gate), durable dedup across restarts, and the controller-owned
interactive selection path.
"""

from __future__ import annotations

import asyncio
import threading
from contextlib import asynccontextmanager
from dataclasses import dataclass, field, fields, replace
from typing import TYPE_CHECKING, Any, cast
from unittest.mock import MagicMock, patch

import nio
import pytest

from mindroom import constants, interactive
from mindroom.attachments import register_local_attachment
from mindroom.bot_runtime_view import BotRuntimeState
from mindroom.coalescing import CoalescingGate, IngressAdmissionClosedError
from mindroom.coalescing_batch import CoalescedBatch, CoalescingKey, PendingEvent, build_coalesced_batch
from mindroom.command_turn_executor import CommandTurnExecutor, CommandTurnExecutorDeps
from mindroom.commands.parsing import CommandType, command_parser
from mindroom.config.agent import AgentConfig
from mindroom.config.auth import AuthorizationConfig
from mindroom.config.main import Config
from mindroom.constants import ROUTER_AGENT_NAME
from mindroom.conversation_resolver import ConversationResolver, ConversationResolverDeps, MessageContext
from mindroom.conversation_state_writer import ConversationStateWriter, ConversationStateWriterDeps
from mindroom.dispatch_callback_outcome import TurnDispatchOutcome
from mindroom.dispatch_obligations import DispatchCallbackKind, DispatchObligationRunner, DispatchObligationStore
from mindroom.dispatch_recovery_context import turn_dispatch_recovery_scope
from mindroom.dispatch_source import (
    EXTERNAL_TRIGGER_SOURCE_KIND,
    SCHEDULED_SOURCE_KIND,
    TRUSTED_INTERNAL_RELAY_SOURCE_KIND,
    ScheduledHistoryBudget,
)
from mindroom.entity_resolution import entity_identity_registry
from mindroom.handled_turns import TurnRecord
from mindroom.hooks import HookContextSupport, HookRegistry, HookRegistryState
from mindroom.inbound_turn_normalizer import (
    InboundTurnNormalizer,
    InboundTurnNormalizerDeps,
    TextNormalizationRequest,
)
from mindroom.ingress_validation import IngressValidator, IngressValidatorDeps
from mindroom.logging_config import get_logger
from mindroom.matrix.cache.thread_history_result import thread_history_result
from mindroom.message_target import MessageTarget
from mindroom.response_admission import ResponseAdmissionRefusedError
from mindroom.response_payload_preparation import DispatchPayloadInputs, ResponsePayloadPreparation
from mindroom.response_runner import ResponseRequest
from mindroom.sync_restart_retry import InterruptedTurnRooms
from mindroom.tool_system.runtime_context import ToolRuntimeSupport
from mindroom.turn_controller import TurnController, TurnControllerDeps
from mindroom.turn_origin import TurnIntent
from mindroom.turn_policy import IngressHookRunner, PreparedDispatch, ResponseAction, TurnPolicy, TurnPolicyDeps
from mindroom.turn_settlement_retry import TurnSettlementRetry
from mindroom.turn_store import TurnStore, TurnStoreDeps
from mindroom.visible_response_reconciliation import VisibleResponseReconciler, VisibleResponseReconcilerDeps
from mindroom.visible_voice_echo import VisibleVoiceEchoDeps, VisibleVoiceEchoLifecycle
from tests.conftest import (
    bind_runtime_paths,
    make_conversation_cache_mock,
    make_matrix_client_mock,
    make_visible_message,
    runtime_paths_for,
    test_runtime_paths,
)

if TYPE_CHECKING:
    from collections.abc import AsyncIterator, Callable, Coroutine
    from pathlib import Path
    from unittest.mock import AsyncMock

    from mindroom.delivery_gateway import DeliveryGateway, EditTextRequest, SendTextRequest
    from mindroom.dispatch_handoff import PreparedTextEvent
    from mindroom.hooks import MessageEnvelope
    from mindroom.matrix.cache import ThreadHistoryResult
    from mindroom.matrix.event_info import EventInfo
    from mindroom.response_lifecycle import QueuedHumanNoticeReservation
    from mindroom.response_runner import ResponseRunner
    from mindroom.turn_policy import _ResponderAvailability as ResponderAvailability

_ROOM_ID = "!focused:localhost"
_SENDER = "@user:localhost"
_EVENT_ID = "$event:localhost"
_THREAD_ROOT = "$root:localhost"


@dataclass
class _RecordingResponseRunner:
    """Typed ResponseRunner stand-in that records the execution-seam requests.

    Mirrors the two runner seams the controller depends on: the request entry
    points (``generate_response`` / ``generate_team_response_helper``) and the
    runner-owned inbox-task ownership (``track_inbox_response``) added by the
    coalescing redesign, where the dispatch handoff completes once the request
    signals ``on_lifecycle_lock_acquired`` and the response keeps running on a
    runner-owned task.
    """

    response_event_id: str | None = "$response:localhost"
    visible_response_event_id: str | None = None
    pre_lock_error: Exception | None = None
    deferred_sync_restart_error: asyncio.CancelledError | None = None
    requests: list[ResponseRequest] = field(default_factory=list)
    team_requests: list[ResponseRequest] = field(default_factory=list)
    inbox_tasks: list[asyncio.Task[None]] = field(default_factory=list)
    recovery_proof_checks: list[Callable[[], bool]] = field(default_factory=list)
    failure_callbacks: list[Callable[[], None] | None] = field(default_factory=list)

    def active_thread_ids_for_room(self, room_id: str) -> frozenset[str | None]:  # noqa: ARG002
        return frozenset()

    def has_active_response_for_target(self, target: MessageTarget) -> bool:  # noqa: ARG002
        return False

    def reserve_waiting_human_message(
        self,
        *,
        target: MessageTarget,  # noqa: ARG002
        response_envelope: MessageEnvelope,  # noqa: ARG002
    ) -> QueuedHumanNoticeReservation:
        msg = "Queued-notice reservations are not part of these focused turn tests"
        raise AssertionError(msg)

    def track_inbox_response(
        self,
        response: Coroutine[Any, Any, None],
        *,
        name: str,
        recovery_proof_ready: Callable[[], bool],
        on_failure: Callable[[], None] | None = None,
    ) -> asyncio.Task[None]:
        self.recovery_proof_checks.append(recovery_proof_ready)
        self.failure_callbacks.append(on_failure)
        task = asyncio.get_running_loop().create_task(response, name=name)
        self.inbox_tasks.append(task)
        return task

    async def settle_inbox_responses(self) -> None:
        """Await every runner-owned response task, surfacing its failures."""
        for task in self.inbox_tasks:
            await task

    async def generate_response(self, request: ResponseRequest) -> str | None:
        self.requests.append(request)
        if self.pre_lock_error is not None:
            raise self.pre_lock_error
        if request.on_lifecycle_lock_acquired is not None:
            request.on_lifecycle_lock_acquired()
        if request.prepare_source_turn is not None and request.prepare_source_turn():
            if request.on_source_turn_suppressed is not None:
                await request.on_source_turn_suppressed()
            return None
        if self.visible_response_event_id is not None and request.on_visible_response is not None:
            await request.on_visible_response(self.visible_response_event_id)
        if self.deferred_sync_restart_error is not None:
            assert self.response_event_id is not None
            assert request.on_interrupted_response_recoverable is not None
            assert request.on_deferred_outcome_handled is not None
            request.on_interrupted_response_recoverable()
            request.on_deferred_outcome_handled(self.response_event_id)
            raise self.deferred_sync_restart_error
        return self.response_event_id

    async def generate_team_response_helper(
        self,
        request: ResponseRequest,
        *,
        team_agents: object,  # noqa: ARG002
        team_mode: str,  # noqa: ARG002
    ) -> str | None:
        self.team_requests.append(request)
        if self.pre_lock_error is not None:
            raise self.pre_lock_error
        if request.on_lifecycle_lock_acquired is not None:
            request.on_lifecycle_lock_acquired()
        if request.prepare_source_turn is not None and request.prepare_source_turn():
            if request.on_source_turn_suppressed is not None:
                await request.on_source_turn_suppressed()
            return None
        if self.visible_response_event_id is not None and request.on_visible_response is not None:
            await request.on_visible_response(self.visible_response_event_id)
        return self.response_event_id


@dataclass
class _RecordingDeliveryGateway:
    """Typed DeliveryGateway stand-in that records visible sends."""

    sent: list[SendTextRequest] = field(default_factory=list)
    edited: list[EditTextRequest] = field(default_factory=list)
    edit_succeeds: bool = True

    async def send_text(self, request: SendTextRequest) -> str | None:
        self.sent.append(request)
        return f"$sent-{len(self.sent)}:localhost"

    async def edit_text(self, request: EditTextRequest) -> bool:
        self.edited.append(request)
        return self.edit_succeeds


@dataclass
class _SpyTurnPolicy:
    """Delegating TurnPolicy wrapper that counts policy evaluations."""

    inner: TurnPolicy
    plan_turn_calls: int = 0

    def can_reply_to_sender(self, sender_id: str) -> bool:
        return self.inner.can_reply_to_sender(sender_id)

    def responder_availability(self) -> ResponderAvailability:
        return self.inner.responder_availability()

    async def responder_candidates_for_room(
        self,
        room: nio.MatrixRoom,
        requester_user_id: str,
        availability: ResponderAvailability | None = None,
    ) -> list[object]:
        return list(await self.inner.responder_candidates_for_room(room, requester_user_id, availability))

    def effective_response_action(self, action: ResponseAction) -> ResponseAction:
        return self.inner.effective_response_action(action)

    async def plan_turn(self, *args: Any, **kwargs: Any) -> Any:  # noqa: ANN401
        self.plan_turn_calls += 1
        return await self.inner.plan_turn(*args, **kwargs)


async def _unused_hook_send_message(*_args: object, **_kwargs: object) -> str | None:
    """Hook message sending must not happen in these tests (the hook registry is empty)."""
    msg = "hook_send_message must not be called in focused turn-controller tests"
    raise AssertionError(msg)


class _UnusedEditRegenerator:
    """Edit-regeneration stub: the edit path must not run in these tests."""

    async def handle_message_edit(
        self,
        room: nio.MatrixRoom,  # noqa: ARG002
        event: nio.RoomMessageText,  # noqa: ARG002
        event_info: EventInfo,  # noqa: ARG002
        requester_user_id: str,  # noqa: ARG002
    ) -> None:
        msg = "Edit regeneration must not run in focused turn-controller tests"
        raise AssertionError(msg)


@dataclass
class _Harness:
    """One directly constructed TurnController plus its observable seams."""

    controller: TurnController
    policy: _SpyTurnPolicy
    runner: _RecordingResponseRunner
    gateway: _RecordingDeliveryGateway
    turn_store: TurnStore
    interrupted_turn_rooms: InterruptedTurnRooms
    gate: CoalescingGate
    gate_batches: list[CoalescedBatch]
    conversation_cache: AsyncMock
    ignored_dispatch_sources: list[tuple[str, ...]]
    retried_dispatch_sources: list[tuple[str, ...]]

    async def deliver(self, room: nio.MatrixRoom, event: nio.RoomMessageText) -> None:
        """Run one inbound text turn end-to-end, settling runner-owned responses.

        Conversational responses keep running on runner-owned inbox tasks after
        the gate handoff completes, so assertions must wait for those tasks too.
        """
        await self.controller.handle_text_event(room, event)
        await self.gate.drain_all()
        await self.runner.settle_inbox_responses()


def _build_harness(
    config: Config,
    storage_path: Path,
    *,
    agent_name: str = "general",
    thread_history: ThreadHistoryResult | None = None,
) -> _Harness:
    """Construct a TurnController from typed collaborators without booting a bot."""
    runtime_paths = runtime_paths_for(config)
    registry = entity_identity_registry(config, runtime_paths)
    matrix_id = registry.current_id(agent_name)
    logger = get_logger("test_turn_controller_focused")
    runtime = BotRuntimeState(
        client=make_matrix_client_mock(user_id=matrix_id.full_id),
        config=config,
        runtime_paths=runtime_paths,
        enable_streaming=True,
        orchestrator=None,
        event_cache=None,
        event_cache_write_coordinator=None,
    )
    conversation_cache = make_conversation_cache_mock()

    @asynccontextmanager
    async def _turn_scope() -> AsyncIterator[None]:
        yield

    conversation_cache.turn_scope = _turn_scope
    if thread_history is not None:
        conversation_cache.get_dispatch_thread_history.return_value = thread_history
    resolver = ConversationResolver(
        ConversationResolverDeps(
            runtime=runtime,
            logger=logger,
            runtime_paths=runtime_paths,
            agent_name=agent_name,
            matrix_id=matrix_id,
            conversation_cache=conversation_cache,
        ),
    )
    normalizer = InboundTurnNormalizer(
        InboundTurnNormalizerDeps(
            runtime=runtime,
            logger=logger,
            storage_path=storage_path,
            runtime_paths=runtime_paths,
        ),
    )
    state_writer = ConversationStateWriter(
        ConversationStateWriterDeps(
            runtime=runtime,
            logger=logger,
            runtime_paths=runtime_paths,
            agent_name=agent_name,
        ),
    )
    hook_context = HookContextSupport(
        runtime=runtime,
        logger=logger,
        runtime_paths=runtime_paths,
        agent_name=agent_name,
        hook_registry_state=HookRegistryState(HookRegistry.empty()),
        hook_send_message=_unused_hook_send_message,
    )
    tool_runtime = ToolRuntimeSupport(
        runtime=runtime,
        logger=logger,
        runtime_paths=runtime_paths,
        storage_path=storage_path,
        agent_name=agent_name,
        matrix_id=matrix_id,
        resolver=resolver,
        hook_context=hook_context,
    )
    turn_store = TurnStore(
        TurnStoreDeps(
            agent_name=agent_name,
            tracking_base_path=storage_path / "tracking",
            state_writer=state_writer,
            resolver=resolver,
            tool_runtime=tool_runtime,
        ),
    )
    policy = _SpyTurnPolicy(
        TurnPolicy(
            TurnPolicyDeps(
                runtime=runtime,
                logger=logger,
                runtime_paths=runtime_paths,
                agent_name=agent_name,
                matrix_id=matrix_id,
            ),
        ),
    )
    runner = _RecordingResponseRunner()
    gateway = _RecordingDeliveryGateway()
    interrupted_turn_rooms = InterruptedTurnRooms()
    controller_ref: list[TurnController] = []
    gate_batches: list[CoalescedBatch] = []
    ignored_dispatch_sources: list[tuple[str, ...]] = []
    retried_dispatch_sources: list[tuple[str, ...]] = []

    async def _dispatch_batch(batch: CoalescedBatch) -> None:
        gate_batches.append(batch)
        await controller_ref[0].handle_coalesced_batch(batch)

    async def _settle_ignored_dispatch_sources(source_event_ids: tuple[str, ...]) -> None:
        ignored_dispatch_sources.append(source_event_ids)

    def _retry_dispatch_sources(source_event_ids: tuple[str, ...]) -> None:
        retried_dispatch_sources.append(source_event_ids)

    async def _recover_config_confirmation_setup(_room_id: str, _preview_event_id: str) -> bool:
        return False

    visible_responses = VisibleResponseReconciler(
        VisibleResponseReconcilerDeps(
            runtime=runtime,
            logger=logger,
            response_sender=matrix_id.full_id,
            turn_store=turn_store,
            delivery_gateway=cast("DeliveryGateway", gateway),
            settle_ignored_sources=_settle_ignored_dispatch_sources,
        ),
    )
    command_executor = CommandTurnExecutor(
        CommandTurnExecutorDeps(
            runtime=runtime,
            logger=logger,
            runtime_paths=runtime_paths,
            agent_name=agent_name,
            normalizer=normalizer,
            conversation_cache=conversation_cache,
            turn_policy=cast("TurnPolicy", policy),
            turn_store=turn_store,
            visible_responses=visible_responses,
            event_cache=lambda: runtime.event_cache,
            recover_config_confirmation_setup=_recover_config_confirmation_setup,
        ),
    )

    gate = CoalescingGate(
        dispatch_batch=_dispatch_batch,
        debounce_seconds=lambda: 0.0,
        is_shutting_down=lambda: False,
    )
    ingress_validator = IngressValidator(
        IngressValidatorDeps(
            runtime=runtime,
            runtime_paths=runtime_paths,
            matrix_id=matrix_id,
            turn_store=turn_store,
            turn_policy=policy,
        ),
    )
    controller = TurnController(
        TurnControllerDeps(
            runtime=runtime,
            logger=logger,
            runtime_paths=runtime_paths,
            agent_name=agent_name,
            matrix_id=matrix_id,
            conversation_cache=conversation_cache,
            resolver=resolver,
            normalizer=normalizer,
            command_executor=command_executor,
            turn_policy=cast("TurnPolicy", policy),
            ingress_hook_runner=IngressHookRunner(hook_context=hook_context),
            response_runner=cast("ResponseRunner", runner),
            delivery_gateway=cast("DeliveryGateway", gateway),
            tool_runtime=tool_runtime,
            turn_store=turn_store,
            coalescing_gate=gate,
            edit_regenerator=_UnusedEditRegenerator(),
            ingress=ingress_validator,
            interrupted_turn_rooms=interrupted_turn_rooms,
            visible_voice_echo=VisibleVoiceEchoLifecycle(
                VisibleVoiceEchoDeps(
                    runtime=runtime,
                    logger=logger,
                    agent_name=agent_name,
                    delivery_gateway=cast("DeliveryGateway", gateway),
                    turn_store=turn_store,
                    ingress=ingress_validator,
                ),
            ),
            visible_responses=visible_responses,
            retry_dispatch_sources=_retry_dispatch_sources,
        ),
    )
    controller_ref.append(controller)
    return _Harness(
        controller=controller,
        policy=policy,
        runner=runner,
        gateway=gateway,
        turn_store=turn_store,
        interrupted_turn_rooms=interrupted_turn_rooms,
        gate=gate,
        gate_batches=gate_batches,
        conversation_cache=conversation_cache,
        ignored_dispatch_sources=ignored_dispatch_sources,
        retried_dispatch_sources=retried_dispatch_sources,
    )


@pytest.fixture
def config(tmp_path: Path) -> Config:
    """Two-agent config bound to isolated runtime paths."""
    return bind_runtime_paths(
        Config(
            agents={
                "general": AgentConfig(display_name="General"),
                "research": AgentConfig(display_name="Research"),
            },
        ),
        test_runtime_paths(tmp_path / "runtime"),
    )


def _entity_user_id(config: Config, entity_name: str) -> str:
    return entity_identity_registry(config, runtime_paths_for(config)).current_id(entity_name).full_id


def _room_with_members(config: Config, *entity_names: str, room_id: str = _ROOM_ID) -> nio.MatrixRoom:
    room = nio.MatrixRoom(room_id, _entity_user_id(config, entity_names[0]) if entity_names else _SENDER)
    room.add_member(_SENDER, _SENDER, None)
    for entity_name in entity_names:
        user_id = _entity_user_id(config, entity_name)
        room.add_member(user_id, user_id, None)
    return room


def _text_event(
    body: str,
    *,
    event_id: str = _EVENT_ID,
    thread_id: str | None = None,
    origin_server_ts: int = 1_000_000,
) -> nio.RoomMessageText:
    content: dict[str, Any] = {"body": body, "msgtype": "m.text"}
    if thread_id is not None:
        content["m.relates_to"] = {"rel_type": "m.thread", "event_id": thread_id}
    return nio.RoomMessageText.from_dict(
        {
            "content": content,
            "event_id": event_id,
            "sender": _SENDER,
            "origin_server_ts": origin_server_ts,
            "room_id": _ROOM_ID,
            "type": "m.room.message",
        },
    )


def _image_event(*, event_id: str) -> nio.RoomMessageImage:
    return cast(
        "nio.RoomMessageImage",
        nio.RoomMessageImage.from_dict(
            {
                "content": {
                    "body": "photo.jpg",
                    "info": {"mimetype": "image/jpeg"},
                    "msgtype": "m.image",
                    "url": "mxc://localhost/photo",
                },
                "event_id": event_id,
                "sender": _SENDER,
                "origin_server_ts": 1_000_000,
                "room_id": _ROOM_ID,
                "type": "m.room.message",
            },
        ),
    )


def _obligation_runner(
    harness: _Harness,
    *,
    tracking_path: Path,
    principal_id: str,
    entity_name: str,
    room: nio.MatrixRoom,
) -> tuple[DispatchObligationRunner, DispatchObligationStore]:
    async def noop(_room: nio.MatrixRoom, _event: nio.Event) -> None:
        pass

    store = DispatchObligationStore(
        tracking_path=tracking_path,
        principal_id=principal_id,
        entity_name=entity_name,
    )
    runner = DispatchObligationRunner(
        store=store,
        callbacks=DispatchObligationRunner.callbacks_for(
            on_message=harness.controller.handle_text_event,
            on_media=harness.controller.handle_media_event,
            on_reaction=cast("Any", noop),
            on_approval=cast("Any", noop),
            on_invite=cast("Any", noop),
            on_room_lifecycle=cast("Any", noop),
            on_redaction=cast("Any", noop),
            on_decryption_failure=cast("Any", noop),
            source_has_live_owner=harness.gate.has_pending_source_event,
        ),
        room_for_id=lambda _room_id: room,
        turn_is_terminal=harness.turn_store.is_durably_handled,
    )
    return runner, store


def _router_relay_event(
    config: Config,
    *,
    event_id: str,
    original_event_id: str,
    body: str,
    origin_server_ts: int,
) -> nio.RoomMessageText:
    """Build one trusted router relay that explicitly replies to its human source."""
    return nio.RoomMessageText.from_dict(
        {
            "content": {
                "body": body,
                "msgtype": "m.text",
                constants.ORIGINAL_SENDER_KEY: _SENDER,
                constants.SOURCE_KIND_KEY: TRUSTED_INTERNAL_RELAY_SOURCE_KIND,
                "m.relates_to": {
                    "rel_type": "m.thread",
                    "event_id": _THREAD_ROOT,
                    "m.in_reply_to": {"event_id": original_event_id},
                },
            },
            "event_id": event_id,
            "sender": _entity_user_id(config, ROUTER_AGENT_NAME),
            "origin_server_ts": origin_server_ts,
            "room_id": _ROOM_ID,
            "type": "m.room.message",
        },
    )


@pytest.mark.asyncio
async def test_replayed_turn_is_rejected_before_policy_and_compacted(config: Config, tmp_path: Path) -> None:
    """A turn superseded by a newer unresponded requester message never reaches policy.

    The dispatch replay guard must consume the exact callback obligation without
    growing the handled-turn ledger, evaluating policy, or sending anything.
    """
    newer_history = thread_history_result(
        [
            make_visible_message(
                sender=_SENDER,
                body="newer follow-up from the same requester",
                event_id="$newer:localhost",
                timestamp=2_000_000,
            ),
        ],
        is_full_history=True,
    )
    harness = _build_harness(config, tmp_path, thread_history=newer_history)
    room = _room_with_members(config, "general")
    event = _text_event("older superseded message", thread_id=_THREAD_ROOT, origin_server_ts=1_000_000)

    await harness.deliver(room, event)

    assert harness.policy.plan_turn_calls == 0
    assert harness.runner.requests == []
    assert harness.gateway.sent == []
    assert harness.turn_store.is_handled(event.event_id) is False
    assert harness.ignored_dispatch_sources == [(event.event_id,)]


@pytest.mark.asyncio
async def test_detached_response_failure_callback_owns_exact_source_retry(config: Config, tmp_path: Path) -> None:
    """The detached response owner must hand its exact sources back to durable retry."""
    harness = _build_harness(config, tmp_path)
    room = _room_with_members(config, "general")
    event = _text_event("please answer", thread_id=_THREAD_ROOT)

    await harness.deliver(room, event)

    assert len(harness.runner.failure_callbacks) == 1
    on_failure = harness.runner.failure_callbacks[0]
    assert on_failure is not None
    on_failure()
    assert harness.retried_dispatch_sources == [(event.event_id,)]


@pytest.mark.asyncio
async def test_recovered_turn_adopts_its_existing_visible_response(config: Config, tmp_path: Path) -> None:
    """Hard-crash replay must edit the original response instead of sending another one."""
    response_event_id = "$thinking:localhost"
    harness = _build_harness(config, tmp_path)
    room = _room_with_members(config, "general")
    event = _text_event("please recover", thread_id=_THREAD_ROOT)
    target = harness.controller.deps.resolver.build_message_target(
        room_id=_ROOM_ID,
        thread_id=_THREAD_ROOT,
        reply_to_event_id=event.event_id,
        event_source=event.source,
    )
    pending_turn = harness.turn_store.attach_response_context(
        TurnRecord.create(
            [event.event_id],
            response_event_id=response_event_id,
            completed=False,
        ),
        history_scope=harness.turn_store.response_history_scope(
            ResponseAction(kind="individual"),
            requester_user_id=_SENDER,
        ),
        conversation_target=target,
    )
    harness.turn_store.record_pending_turn(pending_turn)

    with turn_dispatch_recovery_scope(active=True):
        await harness.deliver(room, event)

    assert len(harness.runner.requests) == 1
    assert harness.runner.requests[0].existing_event_id == response_event_id
    assert harness.runner.requests[0].existing_event_is_placeholder is True


@pytest.mark.asyncio
async def test_incomplete_response_intent_reconciles_matrix_without_recovery_scope(
    config: Config,
    tmp_path: Path,
) -> None:
    """A pending intent reconciles Matrix even when ordinary sync wins the restart race."""
    response_event_id = "$thinking:localhost"
    harness = _build_harness(config, tmp_path)
    room = _room_with_members(config, "general")
    event = _text_event("please recover", thread_id=_THREAD_ROOT)
    target = harness.controller.deps.resolver.build_message_target(
        room_id=_ROOM_ID,
        thread_id=_THREAD_ROOT,
        reply_to_event_id=event.event_id,
        event_source=event.source,
    )
    pending_turn = harness.turn_store.attach_response_context(
        TurnRecord.create([event.event_id], completed=False),
        history_scope=harness.turn_store.response_history_scope(
            ResponseAction(kind="individual"),
            requester_user_id=_SENDER,
        ),
        conversation_target=target,
    )
    harness.turn_store.record_pending_turn(pending_turn)

    with patch(
        "mindroom.visible_response_reconciliation.find_response_event_ids_via_room_messages",
        return_value=frozenset({response_event_id}),
    ) as find_response_event_ids:
        await harness.deliver(room, event)

    assert len(harness.runner.requests) == 1
    assert harness.runner.requests[0].existing_event_id == response_event_id
    assert harness.runner.requests[0].existing_event_is_placeholder is True
    find_response_event_ids.assert_awaited_once()
    assert find_response_event_ids.await_args.args == (harness.controller._client(), _ROOM_ID)
    assert find_response_event_ids.await_args.kwargs["response_sender"] == _entity_user_id(config, "general")
    assert find_response_event_ids.await_args.kwargs["source_event_ids"] == (_EVENT_ID,)


@pytest.mark.asyncio
async def test_recovery_prefers_untracked_terminal_fallback_over_persisted_placeholder(
    config: Config,
    tmp_path: Path,
) -> None:
    """A hard crash after fallback send must adopt that terminal event, not its failed placeholder."""
    harness = _build_harness(config, tmp_path)
    pending_turn = TurnRecord.create(
        [_EVENT_ID],
        response_event_id="$placeholder:localhost",
        completed=False,
    )

    with patch(
        "mindroom.visible_response_reconciliation.find_response_event_ids_via_room_messages",
        return_value=frozenset({"$fallback:localhost"}),
    ) as find_response_event_ids:
        recovered = await harness.controller.deps.visible_responses.recovered_response_event_id(
            pending_turn,
            room_id=_ROOM_ID,
        )

    assert recovered == "$fallback:localhost"
    response_filter = find_response_event_ids.await_args.kwargs["response_source_filter"]
    assert response_filter({"content": {constants.STREAM_STATUS_KEY: constants.STREAM_STATUS_COMPLETED}})
    assert not response_filter(
        {
            "content": {
                constants.STREAM_STATUS_KEY: constants.STREAM_STATUS_COMPLETED,
                constants.VISIBLE_ROUTER_VOICE_ECHO_KEY: True,
            },
        },
    )


@pytest.mark.asyncio
async def test_recovery_lookup_excludes_visible_voice_echo(config: Config, tmp_path: Path) -> None:
    """Router recovery must distinguish its relay from an earlier visible voice echo."""
    harness = _build_harness(config, tmp_path, agent_name=ROUTER_AGENT_NAME)
    pending_turn = TurnRecord.create(
        [_EVENT_ID],
        response_event_id="$voice-echo:localhost",
        completed=False,
    )
    harness.turn_store.record_pending_turn(pending_turn)

    with patch(
        "mindroom.visible_response_reconciliation.find_response_event_ids_via_room_messages",
        return_value=frozenset({"$voice-echo:localhost", "$relay:localhost"}),
    ):
        recovered = await harness.controller.deps.visible_responses.recovered_response_event_id(
            pending_turn,
            room_id=_ROOM_ID,
            excluded_event_ids=("$voice-echo:localhost",),
        )

    assert recovered == "$relay:localhost"


@pytest.mark.asyncio
async def test_visible_response_identity_is_durable_before_generation_finishes(
    config: Config,
    tmp_path: Path,
) -> None:
    """The pending ledger must own the placeholder before a hard crash can lose it."""
    harness = _build_harness(config, tmp_path)
    harness.runner.visible_response_event_id = "$thinking:localhost"
    harness.runner.response_event_id = None
    room = _room_with_members(config, "general")
    event = _text_event("please persist the visible response")

    await harness.deliver(room, event)

    record = harness.turn_store.get_turn_record(event.event_id)
    assert record is not None
    assert record.response_event_id == "$thinking:localhost"
    assert record.completed is False


@pytest.mark.asyncio
async def test_response_intent_is_durable_before_any_matrix_response(config: Config, tmp_path: Path) -> None:
    """A restart can recognize incomplete work even if the first visible send wins the crash race."""
    harness = _build_harness(config, tmp_path)
    harness.runner.response_event_id = None
    room = _room_with_members(config, "general")
    event = _text_event("please persist response intent")

    await harness.deliver(room, event)

    record = harness.turn_store.get_turn_record(event.event_id)
    assert record is not None
    assert record.response_event_id is None
    assert record.completed is False


@pytest.mark.asyncio
async def test_redacted_pending_turn_settles_deferred_source_before_response(
    config: Config,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """A redaction racing the pending-intent write must terminally settle its callback owner."""
    harness = _build_harness(config, tmp_path)
    room = _room_with_members(config, "general")
    event = _text_event("redact before response")
    real_record_pending = harness.turn_store.record_pending_turn

    def record_then_redact(turn_record: TurnRecord) -> TurnRecord | None:
        pending_turn = real_record_pending(turn_record)
        assert pending_turn is not None
        return harness.turn_store.mark_source_redacted(event.event_id)

    monkeypatch.setattr(harness.turn_store, "record_pending_turn", record_then_redact)

    await harness.deliver(room, event)

    assert harness.runner.requests == []
    assert harness.ignored_dispatch_sources == [(event.event_id,)]


@pytest.mark.asyncio
async def test_locked_coalesced_redaction_settles_every_suppressed_source(
    config: Config,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """The locked redaction check must release every physical source in a suppressed batch."""
    harness = _build_harness(config, tmp_path)
    room = _room_with_members(config, "general", ROUTER_AGENT_NAME)
    mention = _entity_user_id(config, "general")
    relay_event_ids = ("$relay-one:localhost", "$relay-two:localhost")
    relay_events = [
        _router_relay_event(
            config,
            event_id=event_id,
            original_event_id=human_event_id,
            body=f"{mention} {body}",
            origin_server_ts=timestamp,
        )
        for event_id, human_event_id, body, timestamp in (
            (relay_event_ids[0], "$human-one:localhost", "first", 1_000_000),
            (relay_event_ids[1], "$human-two:localhost", "second", 1_000_001),
        )
    ]
    batch = build_coalesced_batch(
        CoalescingKey(_ROOM_ID, _THREAD_ROOT, _SENDER),
        [
            PendingEvent(
                event=event,
                room=room,
                source_kind=TRUSTED_INTERNAL_RELAY_SOURCE_KIND,
                requester_user_id=_SENDER,
                trust_internal_payload_metadata=True,
                discovery_event_id=human_event_id,
            )
            for event, human_event_id in zip(
                relay_events,
                ("$human-one:localhost", "$human-two:localhost"),
                strict=True,
            )
        ],
    )
    real_generate_response = harness.runner.generate_response

    async def redact_before_locked_check(request: ResponseRequest) -> str | None:
        harness.turn_store.mark_source_redacted(relay_event_ids[0])
        return await real_generate_response(request)

    monkeypatch.setattr(harness.runner, "generate_response", redact_before_locked_check)

    await harness.controller.handle_coalesced_batch(batch)
    await harness.runner.settle_inbox_responses()

    assert len(harness.runner.requests) == 1
    assert harness.ignored_dispatch_sources == [relay_event_ids]


@pytest.mark.asyncio
async def test_coalesced_router_relays_index_every_human_source_for_edit_lookup(
    config: Config,
    tmp_path: Path,
) -> None:
    """Every human source of coalesced relays must discover the shared response."""
    harness = _build_harness(config, tmp_path)
    room = _room_with_members(config, "general", ROUTER_AGENT_NAME)
    mention = _entity_user_id(config, "general")
    relay_events = [
        _router_relay_event(
            config,
            event_id="$relay-one:localhost",
            original_event_id="$human-one:localhost",
            body=f"{mention} first",
            origin_server_ts=1_000_000,
        ),
        _router_relay_event(
            config,
            event_id="$relay-two:localhost",
            original_event_id="$human-two:localhost",
            body=f"{mention} second",
            origin_server_ts=1_000_001,
        ),
    ]
    batch = build_coalesced_batch(
        CoalescingKey(_ROOM_ID, _THREAD_ROOT, _SENDER),
        [
            PendingEvent(
                event=event,
                room=room,
                source_kind=TRUSTED_INTERNAL_RELAY_SOURCE_KIND,
                requester_user_id=_SENDER,
                trust_internal_payload_metadata=True,
                discovery_event_id=human_event_id,
            )
            for event, human_event_id in zip(
                relay_events,
                ("$human-one:localhost", "$human-two:localhost"),
                strict=True,
            )
        ],
    )

    await harness.controller.handle_coalesced_batch(batch)
    await harness.runner.settle_inbox_responses()

    first_lookup = harness.turn_store.get_turn_record("$human-one:localhost")
    second_lookup = harness.turn_store.get_turn_record("$human-two:localhost")
    assert first_lookup is not None
    assert first_lookup == second_lookup
    assert first_lookup.source_event_ids == ("$relay-one:localhost", "$relay-two:localhost")
    assert first_lookup.discovery_event_ids == ("$human-one:localhost", "$human-two:localhost")
    assert first_lookup.source_event_metadata is not None
    assert first_lookup.source_event_metadata["$relay-one:localhost"].discovery_event_id == "$human-one:localhost"
    assert first_lookup.source_event_metadata["$relay-two:localhost"].discovery_event_id == "$human-two:localhost"


@pytest.mark.asyncio
async def test_single_router_relay_persists_human_prompt_ownership(config: Config, tmp_path: Path) -> None:
    """A routed singleton must retain the alias-to-relay prompt mapping for later edits."""
    harness = _build_harness(config, tmp_path)
    room = _room_with_members(config, "general", ROUTER_AGENT_NAME)
    mention = _entity_user_id(config, "general")
    relay = _router_relay_event(
        config,
        event_id="$relay:localhost",
        original_event_id="$human:localhost",
        body=f"{mention} single",
        origin_server_ts=1_000_000,
    )
    batch = build_coalesced_batch(
        CoalescingKey(_ROOM_ID, _THREAD_ROOT, _SENDER),
        [
            PendingEvent(
                event=relay,
                room=room,
                source_kind=TRUSTED_INTERNAL_RELAY_SOURCE_KIND,
                requester_user_id=_SENDER,
                trust_internal_payload_metadata=True,
                discovery_event_id="$human:localhost",
            ),
        ],
    )

    await harness.controller.handle_coalesced_batch(batch)
    await harness.runner.settle_inbox_responses()

    record = harness.turn_store.get_turn_record("$human:localhost")
    assert record is not None
    assert record.source_event_metadata is not None
    assert record.source_event_metadata["$relay:localhost"].discovery_event_id == "$human:localhost"
    assert record.prompt_source_event_id("$human:localhost") == "$relay:localhost"


@pytest.mark.asyncio
async def test_router_relay_is_admitted_while_original_ingress_claim_is_active(
    config: Config,
    tmp_path: Path,
) -> None:
    """A human alias claim cannot veto its distinct physical router relay."""
    harness = _build_harness(config, tmp_path)
    room = _room_with_members(config, "general", ROUTER_AGENT_NAME)
    original_event_id = "$human-active:localhost"
    original_claim = TurnRecord.create([original_event_id], completed=False)
    relay = _router_relay_event(
        config,
        event_id="$relay-active:localhost",
        original_event_id=original_event_id,
        body=f"{_entity_user_id(config, 'general')} respond",
        origin_server_ts=1_000_000,
    )
    assert harness.turn_store.try_claim_turn(original_claim) is True

    try:
        await harness.deliver(room, relay)
    finally:
        harness.turn_store.release_pending_turn_claim(original_claim)

    assert len(harness.runner.requests) == 1
    metadata = harness.runner.requests[0].matrix_run_metadata
    assert metadata is not None
    assert metadata[constants.MATRIX_SOURCE_EVENT_IDS_METADATA_KEY] == ["$relay-active:localhost"]


@pytest.mark.asyncio
async def test_completed_router_alias_rejects_later_physical_relay(config: Config, tmp_path: Path) -> None:
    """A new relay event cannot replay a human source already completed through another relay."""
    harness = _build_harness(config, tmp_path)
    room = _room_with_members(config, "general", ROUTER_AGENT_NAME)
    original_event_id = "$human-complete:localhost"
    harness.turn_store.record_turn(
        TurnRecord.create(
            ["$relay-complete:localhost"],
            discovery_event_ids=[original_event_id],
            response_event_id="$response-complete:localhost",
        ),
    )
    duplicate_relay = _router_relay_event(
        config,
        event_id="$relay-duplicate:localhost",
        original_event_id=original_event_id,
        body=f"{_entity_user_id(config, 'general')} duplicate",
        origin_server_ts=1_000_001,
    )

    await harness.deliver(room, duplicate_relay)

    assert harness.runner.requests == []
    assert harness.turn_store.get_turn_record("$relay-duplicate:localhost") is None


@pytest.mark.asyncio
async def test_duplicate_router_relay_claim_settles_without_restart(config: Config, tmp_path: Path) -> None:
    """A relay blocked by the same human alias must not remain deferred without an owner."""
    harness = _build_harness(config, tmp_path)
    room = _room_with_members(config, "general", ROUTER_AGENT_NAME)
    original_event_id = "$human-duplicate:localhost"
    mention = _entity_user_id(config, "general")
    first = _router_relay_event(
        config,
        event_id="$relay-one:localhost",
        original_event_id=original_event_id,
        body=f"{mention} first",
        origin_server_ts=1_000_000,
    )
    second = _router_relay_event(
        config,
        event_id="$relay-two:localhost",
        original_event_id=original_event_id,
        body=f"{mention} duplicate",
        origin_server_ts=1_000_001,
    )
    obligation_runner, obligation_store = _obligation_runner(
        harness,
        tracking_path=tmp_path / "dispatch-tracking",
        principal_id=_entity_user_id(config, "general"),
        entity_name="general",
        room=room,
    )
    turn_settlement_retry = TurnSettlementRetry(store=obligation_store)
    harness.turn_store.deps = replace(
        harness.turn_store.deps,
        on_terminal_turn_persisted=turn_settlement_retry.retry,
    )
    normalization_started = asyncio.Event()
    release_normalization = asyncio.Event()
    resolve_text_event = harness.controller.deps.normalizer.resolve_text_event

    async def resolve_with_barrier(
        _normalizer: InboundTurnNormalizer,
        request: TextNormalizationRequest,
    ) -> PreparedTextEvent:
        if request.event.event_id == first.event_id:
            normalization_started.set()
            await release_normalization.wait()
        return await resolve_text_event(request)

    with patch.object(InboundTurnNormalizer, "resolve_text_event", new=resolve_with_barrier):
        first_dispatch = asyncio.create_task(
            obligation_runner.dispatch(room, first, DispatchCallbackKind.MESSAGE),
        )
        await normalization_started.wait()
        second_dispatch = asyncio.create_task(
            obligation_runner.dispatch(room, second, DispatchCallbackKind.MESSAGE),
        )
        await asyncio.sleep(0)
        assert not second_dispatch.done()
        release_normalization.set()
        await asyncio.gather(first_dispatch, second_dispatch)

    await harness.gate.drain_all()
    await harness.runner.settle_inbox_responses()
    if turn_settlement_retry._task is not None:
        await turn_settlement_retry._task

    assert not obligation_store.has_pending(first.event_id, DispatchCallbackKind.MESSAGE)
    assert not obligation_store.has_pending(second.event_id, DispatchCallbackKind.MESSAGE)


@pytest.mark.asyncio
async def test_router_alias_claim_loser_reclaims_after_failed_owner(config: Config, tmp_path: Path) -> None:
    """A relay retries ingress when the competing alias owner exits without terminal truth."""
    harness = _build_harness(config, tmp_path)
    room = _room_with_members(config, "general", ROUTER_AGENT_NAME)
    original_event_id = "$human-retry:localhost"
    active_claim = TurnRecord.create(
        ["$failed-relay:localhost"],
        discovery_event_ids=[original_event_id],
        completed=False,
    )
    event = _router_relay_event(
        config,
        event_id="$retry-relay:localhost",
        original_event_id=original_event_id,
        body=f"{_entity_user_id(config, 'general')} retry",
        origin_server_ts=1_000_000,
    )
    assert harness.turn_store.try_claim_turn(active_claim)

    dispatch = asyncio.create_task(harness.controller.handle_text_event(room, event))
    await asyncio.sleep(0)
    assert not dispatch.done()
    harness.turn_store.release_pending_turn_claim(active_claim)

    assert await dispatch is TurnDispatchOutcome.DEFERRED
    await harness.gate.drain_all()
    await harness.runner.settle_inbox_responses()
    assert len(harness.runner.requests) == 1


@pytest.mark.asyncio
async def test_router_relay_ignored_by_this_agent_compacts_exact_callback(
    config: Config,
    tmp_path: Path,
) -> None:
    """A relay routed elsewhere settles locally without indexing a fake response turn."""
    harness = _build_harness(config, tmp_path)
    room = _room_with_members(config, "general", "research", ROUTER_AGENT_NAME)
    relay = _router_relay_event(
        config,
        event_id="$relay-research:localhost",
        original_event_id="$human-research:localhost",
        body=f"{_entity_user_id(config, 'research')} investigate",
        origin_server_ts=1_000_000,
    )

    await harness.deliver(room, relay)

    assert harness.runner.requests == []
    assert harness.turn_store.get_turn_record("$relay-research:localhost") is None
    assert harness.turn_store.get_turn_record("$human-research:localhost") is None
    assert harness.ignored_dispatch_sources == [("$relay-research:localhost",)]


@pytest.mark.asyncio
async def test_failed_gate_admission_releases_ingress_claim_once(
    config: Config,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Rejected gate admission transfers and closes the pending claim exactly once."""
    harness = _build_harness(config, tmp_path)
    room = _room_with_members(config, "general")
    event = _text_event("admission closes")
    release_claim = MagicMock(wraps=harness.turn_store.release_pending_turn_claim)
    monkeypatch.setattr(harness.turn_store, "release_pending_turn_claim", release_claim)
    monkeypatch.setattr(
        harness.gate,
        "submit_lane_slot",
        MagicMock(side_effect=IngressAdmissionClosedError),
    )

    obligation_runner, obligation_store = _obligation_runner(
        harness,
        tracking_path=tmp_path / "dispatch-tracking",
        principal_id=_entity_user_id(config, "general"),
        entity_name="general",
        room=room,
    )
    obligation = await obligation_runner.persist(room, event, DispatchCallbackKind.MESSAGE)
    assert obligation is not None

    with pytest.raises(IngressAdmissionClosedError):
        await obligation_runner._run_persisted(obligation, room=room, event=event)

    release_claim.assert_called_once()
    assert obligation_store.has_pending(event.event_id, DispatchCallbackKind.MESSAGE)
    competing_claim = TurnRecord.create([event.event_id], completed=False)
    assert harness.turn_store.try_claim_turn(competing_claim) is True
    harness.turn_store.release_pending_turn_claim(competing_claim)


@pytest.mark.asyncio
async def test_failed_media_admission_remains_a_pending_exact_callback(
    config: Config,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """A closed media lane must raise through the durable boundary for recovery."""
    harness = _build_harness(config, tmp_path)
    room = _room_with_members(config, "general")
    event = _image_event(event_id="$media-admission-closed:localhost")
    monkeypatch.setattr(
        harness.gate,
        "submit_lane_slot",
        MagicMock(side_effect=IngressAdmissionClosedError),
    )
    obligation_runner, obligation_store = _obligation_runner(
        harness,
        tracking_path=tmp_path / "dispatch-tracking",
        principal_id=_entity_user_id(config, "general"),
        entity_name="general",
        room=room,
    )
    obligation = await obligation_runner.persist(room, event, DispatchCallbackKind.MEDIA)
    assert obligation is not None

    with pytest.raises(IngressAdmissionClosedError):
        await obligation_runner._run_persisted(obligation, room=room, event=event)

    assert obligation_store.has_pending(event.event_id, DispatchCallbackKind.MEDIA)


@pytest.mark.asyncio
async def test_router_silent_ignore_compacts_exact_callback(config: Config, tmp_path: Path) -> None:
    """A router ignore without a visible voice echo explicitly settles its source."""
    harness = _build_harness(config, tmp_path, agent_name=ROUTER_AGENT_NAME)
    room = _room_with_members(config, ROUTER_AGENT_NAME, "general")
    event = _text_event("hello there", event_id="$router-ignore:localhost")

    obligation_runner, obligation_store = _obligation_runner(
        harness,
        tracking_path=tmp_path / "dispatch-tracking",
        principal_id=_entity_user_id(config, ROUTER_AGENT_NAME),
        entity_name=ROUTER_AGENT_NAME,
        room=room,
    )
    harness.controller.deps = replace(
        harness.controller.deps,
        visible_responses=VisibleResponseReconciler(
            replace(
                harness.controller.deps.visible_responses.deps,
                settle_ignored_sources=obligation_runner.settle_intentionally_ignored_turn_sources,
            ),
        ),
    )

    await obligation_runner.dispatch(room, event, DispatchCallbackKind.MESSAGE)
    await harness.gate.drain_all()

    assert harness.policy.plan_turn_calls == 1
    assert harness.runner.requests == []
    assert harness.turn_store.get_turn_record(event.event_id) is None
    assert not obligation_store.has_pending(event.event_id, DispatchCallbackKind.MESSAGE)


@pytest.mark.asyncio
async def test_sender_outside_reply_allowlist_is_dropped_at_precheck(tmp_path: Path) -> None:
    """An unauthorized sender is filtered at precheck and the turn gets a terminal record."""
    config = bind_runtime_paths(
        Config(
            agents={"general": AgentConfig(display_name="General")},
            authorization=AuthorizationConfig(agent_reply_permissions={"general": ["@owner:localhost"]}),
        ),
        test_runtime_paths(tmp_path / "runtime"),
    )
    harness = _build_harness(config, tmp_path)
    room = _room_with_members(config, "general")
    event = _text_event("hello from a stranger")

    await harness.deliver(room, event)

    assert harness.policy.plan_turn_calls == 0
    assert harness.runner.requests == []
    assert harness.gateway.sent == []
    # The drop is terminal: the turn is recorded so restarts never resurrect it.
    assert harness.turn_store.is_handled(event.event_id) is True


@pytest.mark.asyncio
async def test_policy_ignore_sends_nothing_and_compacts_callback(config: Config, tmp_path: Path) -> None:
    """An ignore plan compacts exact callback truth without growing the turn ledger."""
    harness = _build_harness(config, tmp_path)
    room = _room_with_members(config, "general", "research")
    event = _text_event("untagged message with multiple visible responders")

    await harness.deliver(room, event)

    assert harness.policy.plan_turn_calls == 1
    assert harness.runner.requests == []
    assert harness.gateway.sent == []
    assert harness.turn_store.is_handled(event.event_id) is False
    assert harness.ignored_dispatch_sources == [(event.event_id,)]


@pytest.mark.asyncio
async def test_policy_respond_crosses_seam_as_immutable_values(config: Config, tmp_path: Path) -> None:
    """A respond plan invokes the runner once with value-only ResponseRequest inputs.

    Pins the one-way ingress→execution seam: the request carries an immutable
    ``ResponsePayloadPreparation`` value (no callbacks back into the controller),
    and the recorded outcome lands in the durable turn store.
    """
    harness = _build_harness(config, tmp_path)
    room = _room_with_members(config, "general")
    event = _text_event("please summarize the build failure")

    await harness.deliver(room, event)

    # Conversational turns flow through the coalescing gate, unlike commands.
    assert len(harness.gate_batches) == 1
    assert harness.policy.plan_turn_calls == 1
    assert len(harness.runner.requests) == 1
    request = harness.runner.requests[0]

    assert request.prompt == "please summarize the build failure"
    assert request.user_id == _SENDER
    assert request.correlation_id == event.event_id
    assert request.current_timestamp_ms == float(event.server_timestamp)
    assert request.current_prompt_is_structured is False
    assert request.response_envelope.requester_id == _SENDER
    assert request.response_envelope.target.room_id == _ROOM_ID
    # A rootable room-level message becomes its own thread root.
    assert request.response_envelope.target.resolved_thread_id == event.event_id

    preparation = request.payload_preparation
    assert isinstance(preparation, ResponsePayloadPreparation)
    assert preparation.prompt == "please summarize the build failure"
    assert preparation.action_kind == "individual"
    assert preparation.target_member_names is None
    assert preparation.payload_inputs == DispatchPayloadInputs(
        message_attachment_ids=(),
        trusted_attachment_ids=(),
        media_events=(),
    )
    # The envelope crosses the seam as the same value the dispatch was planned with.
    assert request.response_envelope is preparation.dispatch.envelope

    # Payload data crosses the seam as values, not closures: the old
    # prepare-after-lock callback is gone and nothing callable rides inside the
    # preparation value. The only callback on the request is the redesign's
    # inbox handoff signal, fired by the runner when it takes the lifecycle lock.
    request_field_names = {request_field.name for request_field in fields(ResponseRequest)}
    assert "prepare_after_lock" not in request_field_names
    assert request.on_lifecycle_lock_acquired is not None
    assert not any(callable(getattr(preparation, preparation_field.name)) for preparation_field in fields(preparation))

    metadata = request.matrix_run_metadata
    assert metadata is not None
    assert metadata[constants.MATRIX_RESPONSE_OWNER_METADATA_KEY] == "general"
    assert harness.turn_store.is_handled(event.event_id) is True


@pytest.mark.asyncio
async def test_response_waits_for_pending_context_persistence_before_generation(
    config: Config,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Session-backed generation must not start before its cleanup context is durable."""
    harness = _build_harness(config, tmp_path)
    room = _room_with_members(config, "general")
    event = _text_event("persist my response context first")
    real_persist = harness.turn_store._ledger._persist_records
    persist_started = threading.Event()
    release_persist = threading.Event()

    def persist_with_barrier(turn_records: tuple[TurnRecord, ...]) -> None:
        if any(event.event_id in record.indexed_event_ids and not record.completed for record in turn_records):
            persist_started.set()
            if not release_persist.wait(timeout=5):
                msg = "test did not release pending-context persistence"
                raise TimeoutError(msg)
        real_persist(turn_records)

    monkeypatch.setattr(harness.turn_store._ledger, "_persist_records", persist_with_barrier)
    delivery = asyncio.create_task(harness.deliver(room, event))
    try:
        assert await asyncio.to_thread(persist_started.wait, 5)
        assert harness.runner.requests == []
    finally:
        release_persist.set()

    await asyncio.wait_for(delivery, timeout=5)

    assert len(harness.runner.requests) == 1
    persisted = harness.turn_store.get_turn_record(event.event_id)
    assert persisted is not None
    assert persisted.completed is True


def _scheduled_fire_event(
    config: Config,
    *,
    extra_content: dict[str, Any],
    event_id: str = "$scheduled:localhost",
    new_thread: bool = False,
) -> nio.RoomMessageText:
    """Build a self-authored scheduled-fire event as the scheduling executor sends it."""
    content: dict[str, Any] = {
        "body": "Poll the queue" if new_thread else "⏰ [Automated Task]\nPoll the queue",
        "msgtype": "m.text",
        constants.SOURCE_KIND_KEY: SCHEDULED_SOURCE_KIND,
        constants.ORIGINAL_SENDER_KEY: _SENDER,
        **extra_content,
    }
    if new_thread:
        content[constants.PER_FIRE_THREAD_ROOT_KEY] = True
    return nio.RoomMessageText.from_dict(
        {
            "content": content,
            "event_id": event_id,
            "sender": _entity_user_id(config, "general"),
            "origin_server_ts": 1_000_000,
            "room_id": _ROOM_ID,
            "type": "m.room.message",
        },
    )


def _single_agent_config(tmp_path: Path, thread_mode: str) -> Config:
    """One-agent config with the given thread mode, bound to isolated runtime paths."""
    return bind_runtime_paths(
        Config(agents={"general": AgentConfig(display_name="General", thread_mode=thread_mode)}),
        test_runtime_paths(tmp_path / "runtime"),
    )


@pytest.mark.asyncio
async def test_scheduled_fire_history_limit_reaches_response_request(config: Config, tmp_path: Path) -> None:
    """The history limit annotated on a trusted scheduled fire lands on the ResponseRequest."""
    harness = _build_harness(config, tmp_path)
    room = _room_with_members(config, "general")
    event = _scheduled_fire_event(config, extra_content={constants.SCHEDULED_HISTORY_LIMIT_KEY: 3})

    await harness.deliver(room, event)

    assert len(harness.runner.requests) == 1
    request = harness.runner.requests[0]
    assert request.scheduled_history_budget == ScheduledHistoryBudget(
        limit=3,
        source_event_id="$scheduled:localhost",
    )
    assert request.response_envelope.origin.intent is TurnIntent.SCHEDULED_FIRE


@pytest.mark.asyncio
async def test_scheduled_router_handoff_history_limit_reaches_response_request(
    config: Config,
    tmp_path: Path,
) -> None:
    """A trusted router handoff preserves the scheduled fire's history cap for the target agent."""
    harness = _build_harness(config, tmp_path)
    room = _room_with_members(config, "general", ROUTER_AGENT_NAME)
    event = nio.RoomMessageText.from_dict(
        {
            "content": {
                "body": "@general could you help with this?",
                "msgtype": "m.text",
                constants.SOURCE_KIND_KEY: TRUSTED_INTERNAL_RELAY_SOURCE_KIND,
                constants.ORIGINAL_SENDER_KEY: _SENDER,
                constants.SCHEDULED_HISTORY_LIMIT_KEY: 2,
                "m.relates_to": {"m.in_reply_to": {"event_id": "$scheduled:localhost"}},
            },
            "event_id": "$scheduled-router-handoff:localhost",
            "sender": _entity_user_id(config, ROUTER_AGENT_NAME),
            "origin_server_ts": 1_000_000,
            "room_id": _ROOM_ID,
            "type": "m.room.message",
        },
    )

    await harness.deliver(room, event)

    assert len(harness.runner.requests) == 1
    request = harness.runner.requests[0]
    assert request.scheduled_history_budget == ScheduledHistoryBudget(
        limit=2,
        source_event_id="$scheduled:localhost",
    )
    assert request.response_envelope.origin.intent is TurnIntent.ROUTER_HANDOFF


@pytest.mark.asyncio
async def test_router_relay_keeps_original_alias_unsettled_through_gate_handoff(
    config: Config,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """The routed source alias stays claimed while dispatch normalizes the relay."""
    harness = _build_harness(config, tmp_path)
    room = _room_with_members(config, "general", ROUTER_AGENT_NAME)
    original_event_id = "$routed-original:localhost"
    event = nio.RoomMessageText.from_dict(
        {
            "content": {
                "body": "@general could you help with this?",
                "msgtype": "m.text",
                constants.ORIGINAL_SENDER_KEY: _SENDER,
                constants.SOURCE_KIND_KEY: TRUSTED_INTERNAL_RELAY_SOURCE_KIND,
                "m.relates_to": {
                    "rel_type": "m.thread",
                    "event_id": _THREAD_ROOT,
                    "m.in_reply_to": {"event_id": original_event_id},
                },
            },
            "event_id": "$router-relay:localhost",
            "sender": _entity_user_id(config, ROUTER_AGENT_NAME),
            "origin_server_ts": 1_000_000,
            "room_id": _ROOM_ID,
            "type": "m.room.message",
        },
    )
    normalization_started = asyncio.Event()
    release_normalization = asyncio.Event()
    resolve_text_event = InboundTurnNormalizer.resolve_text_event

    async def normalize_with_barrier(
        normalizer: InboundTurnNormalizer,
        request: TextNormalizationRequest,
    ) -> PreparedTextEvent:
        normalization_started.set()
        await release_normalization.wait()
        return await resolve_text_event(normalizer, request)

    monkeypatch.setattr(InboundTurnNormalizer, "resolve_text_event", normalize_with_barrier)
    delivery = asyncio.create_task(harness.deliver(room, event))
    await normalization_started.wait()

    wait_started = asyncio.Event()

    async def wait_for_alias() -> None:
        wait_started.set()
        await harness.turn_store.wait_for_turn_settled((original_event_id,))

    waiter = asyncio.create_task(wait_for_alias())
    await wait_started.wait()
    try:
        assert not waiter.done()
    finally:
        release_normalization.set()
        await delivery
    await waiter


@pytest.mark.asyncio
async def test_scheduled_new_thread_survives_router_handoff_in_room_mode(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """A router relay preserves the scheduled fire's per-fire thread and session."""
    config = bind_runtime_paths(
        Config(
            agents={
                "general": AgentConfig(display_name="General", thread_mode="room"),
                "research": AgentConfig(display_name="Research"),
            },
        ),
        test_runtime_paths(tmp_path / "runtime"),
    )
    room = _room_with_members(config, ROUTER_AGENT_NAME, "general", "research")
    scheduled_event = _scheduled_fire_event(config, extra_content={}, new_thread=True)
    router_harness = _build_harness(config, tmp_path / "router", agent_name=ROUTER_AGENT_NAME)

    async def _route_to_general(*_args: object, **_kwargs: object) -> str:
        return "general"

    monkeypatch.setattr("mindroom.turn_controller.suggest_responder_for_message", _route_to_general)

    await router_harness.deliver(room, scheduled_event)

    assert len(router_harness.gateway.sent) == 1
    handoff = router_harness.gateway.sent[0]
    assert handoff.target.resolved_thread_id == scheduled_event.event_id
    assert handoff.extra_content is not None
    assert handoff.extra_content[constants.SOURCE_KIND_KEY] == TRUSTED_INTERNAL_RELAY_SOURCE_KIND
    assert handoff.extra_content[constants.PER_FIRE_THREAD_ROOT_KEY] is True
    assert handoff.extra_content[constants.PER_FIRE_THREAD_ROOT_EVENT_ID_KEY] == scheduled_event.event_id

    relay_content = {
        "body": handoff.response_text,
        "msgtype": "m.text",
        **handoff.extra_content,
        "m.relates_to": {"rel_type": "m.thread", "event_id": scheduled_event.event_id},
    }
    relay_event = nio.RoomMessageText.from_dict(
        {
            "content": relay_content,
            "event_id": "$scheduled-router-handoff:localhost",
            "sender": _entity_user_id(config, ROUTER_AGENT_NAME),
            "origin_server_ts": 1_000_001,
            "room_id": _ROOM_ID,
            "type": "m.room.message",
        },
    )
    agent_harness = _build_harness(config, tmp_path / "agent")

    await agent_harness.deliver(room, relay_event)

    assert len(agent_harness.runner.requests) == 1
    target = agent_harness.runner.requests[0].response_envelope.target
    assert target.resolved_thread_id == scheduled_event.event_id
    assert target.session_id == f"{_ROOM_ID}:{scheduled_event.event_id}"


@pytest.mark.asyncio
async def test_scheduled_fire_without_annotation_keeps_full_history(config: Config, tmp_path: Path) -> None:
    """A scheduled fire without the annotation must not cap history for that turn."""
    harness = _build_harness(config, tmp_path)
    room = _room_with_members(config, "general")
    event = _scheduled_fire_event(config, extra_content={})

    await harness.deliver(room, event)

    assert len(harness.runner.requests) == 1
    assert harness.runner.requests[0].scheduled_history_budget is None


@pytest.mark.asyncio
async def test_user_message_cannot_spoof_scheduled_history_limit(config: Config, tmp_path: Path) -> None:
    """History-limit annotations on untrusted user messages are ignored."""
    harness = _build_harness(config, tmp_path)
    room = _room_with_members(config, "general")
    event = nio.RoomMessageText.from_dict(
        {
            "content": {
                "body": "please summarize the build failure",
                "msgtype": "m.text",
                constants.SOURCE_KIND_KEY: SCHEDULED_SOURCE_KIND,
                constants.SCHEDULED_HISTORY_LIMIT_KEY: 0,
            },
            "event_id": _EVENT_ID,
            "sender": _SENDER,
            "origin_server_ts": 1_000_000,
            "room_id": _ROOM_ID,
            "type": "m.room.message",
        },
    )

    await harness.deliver(room, event)

    assert len(harness.runner.requests) == 1
    request = harness.runner.requests[0]
    assert request.scheduled_history_budget is None
    assert request.response_envelope.origin.intent is not TurnIntent.SCHEDULED_FIRE


@pytest.mark.asyncio
@pytest.mark.parametrize("thread_mode", ["thread", "room"])
async def test_scheduled_fire_response_starts_per_fire_thread_session(tmp_path: Path, thread_mode: str) -> None:
    """A room-level scheduled fire roots a per-fire thread and session in both thread modes."""
    config = _single_agent_config(tmp_path, thread_mode)
    harness = _build_harness(config, tmp_path)
    room = _room_with_members(config, "general")
    event = _scheduled_fire_event(config, extra_content={}, new_thread=True)

    await harness.deliver(room, event)

    assert len(harness.runner.requests) == 1
    target = harness.runner.requests[0].response_envelope.target
    assert target.resolved_thread_id == "$scheduled:localhost"
    assert target.session_id == f"{_ROOM_ID}:$scheduled:localhost"


@pytest.mark.asyncio
async def test_external_trigger_fire_response_starts_per_fire_thread_session(tmp_path: Path) -> None:
    """A room-level external trigger delivery roots a per-fire thread and session in room mode."""
    config = _single_agent_config(tmp_path, "room")
    harness = _build_harness(config, tmp_path)
    room = _room_with_members(config, "general")
    event = nio.RoomMessageText.from_dict(
        {
            "content": {
                "body": "@general poll the webhook queue",
                "msgtype": "m.text",
                constants.SOURCE_KIND_KEY: EXTERNAL_TRIGGER_SOURCE_KIND,
                constants.ORIGINAL_SENDER_KEY: _SENDER,
                constants.PER_FIRE_THREAD_ROOT_KEY: True,
            },
            "event_id": "$trigger:localhost",
            "sender": _entity_user_id(config, "general"),
            "origin_server_ts": 1_000_000,
            "room_id": _ROOM_ID,
            "type": "m.room.message",
        },
    )

    await harness.deliver(room, event)

    assert len(harness.runner.requests) == 1
    target = harness.runner.requests[0].response_envelope.target
    assert target.resolved_thread_id == "$trigger:localhost"
    assert target.session_id == f"{_ROOM_ID}:$trigger:localhost"


@pytest.mark.asyncio
async def test_consecutive_scheduled_fires_resolve_distinct_sessions(tmp_path: Path) -> None:
    """Two fires of one recurring room-mode schedule never share a thread root or session."""
    config = _single_agent_config(tmp_path, "room")
    harness = _build_harness(config, tmp_path)
    room = _room_with_members(config, "general")

    await harness.deliver(
        room,
        _scheduled_fire_event(config, extra_content={}, event_id="$fire-one:localhost", new_thread=True),
    )
    await harness.deliver(
        room,
        _scheduled_fire_event(config, extra_content={}, event_id="$fire-two:localhost", new_thread=True),
    )

    assert len(harness.runner.requests) == 2
    first_target = harness.runner.requests[0].response_envelope.target
    second_target = harness.runner.requests[1].response_envelope.target
    assert first_target.resolved_thread_id == "$fire-one:localhost"
    assert second_target.resolved_thread_id == "$fire-two:localhost"
    assert first_target.session_id == f"{_ROOM_ID}:$fire-one:localhost"
    assert second_target.session_id == f"{_ROOM_ID}:$fire-two:localhost"
    assert first_target.session_id != second_target.session_id


@pytest.mark.asyncio
async def test_concurrent_scheduled_fires_use_distinct_coalescing_threads(tmp_path: Path) -> None:
    """Concurrent per-fire deliveries receive distinct keys before entering the gate."""
    config = _single_agent_config(tmp_path, "room")
    harness = _build_harness(config, tmp_path)
    room = _room_with_members(config, "general")
    first_event = _scheduled_fire_event(
        config,
        extra_content={},
        event_id="$fire-one:localhost",
        new_thread=True,
    )
    second_event = _scheduled_fire_event(
        config,
        extra_content={},
        event_id="$fire-two:localhost",
        new_thread=True,
    )

    first_thread_id, second_thread_id = await asyncio.gather(
        harness.controller.deps.resolver.coalescing_thread_id(room, first_event),
        harness.controller.deps.resolver.coalescing_thread_id(room, second_event),
    )

    assert first_thread_id == first_event.event_id
    assert second_thread_id == second_event.event_id
    assert first_thread_id != second_thread_id


@pytest.mark.asyncio
async def test_scheduled_fire_into_persisted_thread_keeps_thread_session(config: Config, tmp_path: Path) -> None:
    """A scheduled fire delivered into its persisted thread keeps that thread's session."""
    persisted_thread_history = thread_history_result(
        [
            make_visible_message(
                sender=_SENDER,
                body="original schedule request",
                event_id=_THREAD_ROOT,
                timestamp=500_000,
            ),
        ],
        is_full_history=True,
    )
    harness = _build_harness(config, tmp_path, thread_history=persisted_thread_history)
    room = _room_with_members(config, "general")
    event = _scheduled_fire_event(
        config,
        extra_content={"m.relates_to": {"rel_type": "m.thread", "event_id": _THREAD_ROOT}},
        event_id="$scheduled-in-thread:localhost",
    )

    await harness.deliver(room, event)

    assert len(harness.runner.requests) == 1
    target = harness.runner.requests[0].response_envelope.target
    assert target.resolved_thread_id == _THREAD_ROOT
    assert target.session_id == f"{_ROOM_ID}:{_THREAD_ROOT}"


@pytest.mark.asyncio
async def test_room_mode_plain_user_message_keeps_room_session(tmp_path: Path) -> None:
    """Non-scheduled room-level messages keep the shared room session in room mode."""
    config = _single_agent_config(tmp_path, "room")
    harness = _build_harness(config, tmp_path)
    room = _room_with_members(config, "general")
    event = _text_event("hello there")

    await harness.deliver(room, event)

    assert len(harness.runner.requests) == 1
    target = harness.runner.requests[0].response_envelope.target
    assert target.resolved_thread_id is None
    assert target.session_id == _ROOM_ID
    harness.interrupted_turn_rooms.register(event.event_id, room_id=room.room_id)
    assert harness.runner.recovery_proof_checks[0]() is False


@pytest.mark.asyncio
async def test_write_behind_handled_thread_does_not_prove_recovery(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """In-memory handled state cannot prove that shutdown recovery is durable."""
    config = _single_agent_config(tmp_path, "thread")
    harness = _build_harness(config, tmp_path)
    room = _room_with_members(config, "general")
    event = _text_event("hello there")
    response_started = asyncio.Event()
    release_response = asyncio.Event()
    generate_response = harness.runner.generate_response

    async def generate_with_barrier(request: ResponseRequest) -> str | None:
        response_started.set()
        await release_response.wait()
        return await generate_response(request)

    monkeypatch.setattr(harness.runner, "generate_response", generate_with_barrier)
    delivery = asyncio.create_task(harness.deliver(room, event))
    await response_started.wait()

    real_persist = harness.turn_store._ledger._persist_records
    persist_started = threading.Event()
    release_persist = threading.Event()

    def persist_with_barrier(turn_records: tuple[TurnRecord, ...]) -> None:
        persist_started.set()
        if not release_persist.wait(timeout=5):
            msg = "test did not release terminal-turn persistence"
            raise TimeoutError(msg)
        real_persist(turn_records)

    monkeypatch.setattr(harness.turn_store._ledger, "_persist_records", persist_with_barrier)
    release_response.set()

    try:
        await delivery
        assert await asyncio.to_thread(persist_started.wait, 5)
        assert harness.runner.requests[0].response_envelope.target.resolved_thread_id == event.event_id
        assert harness.turn_store.is_handled(event.event_id)
        assert not harness.interrupted_turn_rooms.pending_room_ids
        assert harness.runner.recovery_proof_checks[0]() is False
    finally:
        release_persist.set()
        await asyncio.to_thread(harness.turn_store._ledger.flush)


@pytest.mark.asyncio
async def test_room_mode_schedule_without_new_thread_keeps_room_session(tmp_path: Path) -> None:
    """A relation-free room-scope schedule keeps the shared room session."""
    config = _single_agent_config(tmp_path, "room")
    harness = _build_harness(config, tmp_path)
    room = _room_with_members(config, "general")

    await harness.deliver(room, _scheduled_fire_event(config, extra_content={}))

    assert len(harness.runner.requests) == 1
    target = harness.runner.requests[0].response_envelope.target
    assert target.resolved_thread_id is None
    assert target.session_id == _ROOM_ID


@pytest.mark.asyncio
async def test_user_message_cannot_spoof_scheduled_thread_promotion(tmp_path: Path) -> None:
    """A scheduled marker on an untrusted user message must not force a per-fire thread."""
    config = _single_agent_config(tmp_path, "room")
    harness = _build_harness(config, tmp_path)
    room = _room_with_members(config, "general")
    event = nio.RoomMessageText.from_dict(
        {
            "content": {
                "body": "pretend to be a scheduled fire",
                "msgtype": "m.text",
                constants.SOURCE_KIND_KEY: SCHEDULED_SOURCE_KIND,
                constants.PER_FIRE_THREAD_ROOT_KEY: True,
                constants.PER_FIRE_THREAD_ROOT_EVENT_ID_KEY: "$spoofed-root:localhost",
            },
            "event_id": _EVENT_ID,
            "sender": _SENDER,
            "origin_server_ts": 1_000_000,
            "room_id": _ROOM_ID,
            "type": "m.room.message",
        },
    )

    await harness.deliver(room, event)

    assert len(harness.runner.requests) == 1
    target = harness.runner.requests[0].response_envelope.target
    assert target.resolved_thread_id is None
    assert target.session_id == _ROOM_ID


@pytest.mark.asyncio
async def test_deferred_sync_restart_records_handled_outcome_before_rethrow(
    config: Config,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """A turn interrupted by bot replacement must settle durably before rethrowing."""
    harness = _build_harness(config, tmp_path)
    harness.runner.deferred_sync_restart_error = asyncio.CancelledError("sync_restart")
    room = _room_with_members(config, "general")
    event = _text_event("please survive the sync restart")
    response_started = asyncio.Event()
    release_response = asyncio.Event()
    generate_response = harness.runner.generate_response

    async def generate_with_barrier(request: ResponseRequest) -> str | None:
        response_started.set()
        await release_response.wait()
        return await generate_response(request)

    monkeypatch.setattr(harness.runner, "generate_response", generate_with_barrier)
    delivery = asyncio.create_task(harness.deliver(room, event))
    await response_started.wait()
    recovery_ready = harness.runner.recovery_proof_checks[0]
    assert recovery_ready() is False
    harness.interrupted_turn_rooms.register("$different", room_id=room.room_id)
    assert recovery_ready() is False

    release_response.set()
    with pytest.raises(asyncio.CancelledError, match="sync_restart"):
        await delivery

    assert recovery_ready() is True
    assert harness.interrupted_turn_rooms.contains(event.event_id)
    assert harness.interrupted_turn_rooms.pending_room_ids == {room.room_id}
    assert harness.runner.requests[0].sync_restart_retry_source_event_id is None
    assert harness.turn_store.is_handled(event.event_id) is True
    record = harness.turn_store.get_turn_record(event.event_id)
    assert record is not None
    assert record.response_event_id == "$response:localhost"


@pytest.mark.asyncio
async def test_interrupted_router_relay_detaches_its_advisory_human_alias(
    config: Config,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """An interrupted relay must drop an alias whose owner settled independently."""
    harness = _build_harness(config, tmp_path)
    harness.runner.deferred_sync_restart_error = asyncio.CancelledError("sync_restart")
    room = _room_with_members(config, "general", ROUTER_AGENT_NAME)
    human_event_id = "$human:localhost"
    relay = _router_relay_event(
        config,
        event_id="$relay:localhost",
        original_event_id=human_event_id,
        body=f"{_entity_user_id(config, 'general')} respond",
        origin_server_ts=1_000_000,
    )
    response_started = asyncio.Event()
    release_response = asyncio.Event()
    generate_response = harness.runner.generate_response

    async def generate_with_barrier(request: ResponseRequest) -> str | None:
        response_started.set()
        await release_response.wait()
        return await generate_response(request)

    monkeypatch.setattr(harness.runner, "generate_response", generate_with_barrier)
    delivery = asyncio.create_task(harness.deliver(room, relay))
    await response_started.wait()
    harness.turn_store.record_turn(
        TurnRecord.create([human_event_id], response_event_id="$original-response:localhost"),
    )
    release_response.set()
    with pytest.raises(asyncio.CancelledError, match="sync_restart"):
        await delivery

    relay_record = harness.turn_store.get_turn_record(relay.event_id)
    assert relay_record is not None
    assert relay_record.discovery_event_ids == ()
    assert harness.interrupted_turn_rooms.pending_room_ids == {room.room_id}


@pytest.mark.asyncio
async def test_pre_lock_replacement_refusal_poisons_coalescing_drain(config: Config, tmp_path: Path) -> None:
    """A refused source callback must invalidate checkpoint certification."""
    harness = _build_harness(config, tmp_path)
    harness.runner.pre_lock_error = ResponseAdmissionRefusedError()
    room = _room_with_members(config, "general")
    event = _text_event("please survive replacement")

    await harness.controller.handle_text_event(room, event)
    drain_result = await harness.gate.drain_all()
    await asyncio.gather(*harness.runner.inbox_tasks, return_exceptions=True)

    assert drain_result.completed is False
    assert drain_result.dispatch_failure_count == 1
    record = harness.turn_store.get_turn_record(event.event_id)
    assert record is not None
    assert record.completed is False


@pytest.mark.asyncio
async def test_command_turn_records_terminal_outcome_through_turn_store(config: Config, tmp_path: Path) -> None:
    """A ``!command`` turn executes on the router and records a terminal TurnStore outcome.

    Since the coalescing redesign, command-shaped human text is a control input:
    it dispatches directly at ingress and must never enter the coalescing gate.
    """
    harness = _build_harness(config, tmp_path, agent_name=ROUTER_AGENT_NAME)
    room = _room_with_members(config, ROUTER_AGENT_NAME, "general", "research")
    event = _text_event("!help", event_id="$command:localhost")

    await harness.deliver(room, event)

    # Commands bypass turn-policy planning, AI execution, and the coalescing gate.
    assert harness.policy.plan_turn_calls == 0
    assert harness.runner.requests == []
    assert harness.gate_batches == []
    assert len(harness.gateway.sent) == 1
    assert "command" in harness.gateway.sent[0].response_text.lower()
    assert harness.turn_store.is_handled(event.event_id) is True

    record = harness.turn_store.get_turn_record(event.event_id)
    assert record is not None
    assert record.response_event_id == "$sent-1:localhost"


@pytest.mark.asyncio
async def test_command_replay_adopts_durable_response(config: Config, tmp_path: Path) -> None:
    """A command replay must reuse the response persisted before terminal settlement."""
    harness = _build_harness(config, tmp_path, agent_name=ROUTER_AGENT_NAME)
    room = _room_with_members(config, ROUTER_AGENT_NAME, "general", "research")
    event = _text_event("!help", event_id="$command-replay:localhost")
    command = command_parser.parse(event.body)
    assert command is not None
    target = harness.controller.deps.resolver.build_message_target(
        room_id=room.room_id,
        thread_id=None,
        reply_to_event_id=event.event_id,
        event_source=event.source,
    )
    handled_turn = TurnRecord.create([event.event_id])

    with patch.object(
        harness.controller.deps.visible_responses,
        "record_pending_visible_response",
        wraps=harness.controller.deps.visible_responses.record_pending_visible_response,
    ) as record_visible_response:
        with patch.object(harness.turn_store, "record_responded_turn"):
            await harness.controller.deps.command_executor.execute(
                room,
                event,
                _SENDER,
                command,
                target=target,
                handled_turn=handled_turn,
            )

        pending_turn = harness.turn_store.get_turn_record(event.event_id)
        assert pending_turn is not None
        assert pending_turn.completed is False
        assert pending_turn.response_event_id == "$sent-1:localhost"

        await harness.controller.deps.command_executor.execute(
            room,
            event,
            _SENDER,
            command,
            target=target,
            handled_turn=handled_turn,
        )

    assert record_visible_response.await_count == 2

    assert len(harness.gateway.sent) == 1
    record = harness.turn_store.get_turn_record(event.event_id)
    assert record is not None
    assert record.completed is True
    assert record.response_event_id == "$sent-1:localhost"


@pytest.mark.asyncio
async def test_config_command_replay_recovers_confirmation_before_terminal_settlement(
    config: Config,
    tmp_path: Path,
) -> None:
    """A visible config preview is terminal only after its confirmation state is recovered."""
    harness = _build_harness(config, tmp_path, agent_name=ROUTER_AGENT_NAME)
    recoveries: list[tuple[str, str]] = []

    async def recover_setup(room_id: str, preview_event_id: str) -> bool:
        recoveries.append((room_id, preview_event_id))
        return True

    harness.controller.deps = replace(
        harness.controller.deps,
        command_executor=CommandTurnExecutor(
            replace(
                harness.controller.deps.command_executor.deps,
                recover_config_confirmation_setup=recover_setup,
            ),
        ),
    )
    command_turn = TurnRecord.create(["$config-command"])

    recovered = await harness.controller.deps.command_executor._recover_visible_response(
        room_id=_ROOM_ID,
        command_type=CommandType.CONFIG,
        command_turn=command_turn,
        response_event_id="$preview",
    )

    assert recovered is True
    assert recoveries == [(_ROOM_ID, "$preview")]
    assert harness.turn_store.is_handled("$config-command")


@pytest.mark.asyncio
async def test_command_replay_does_not_repeat_mutation_after_visible_response(config: Config, tmp_path: Path) -> None:
    """A recovered command response must settle without rerunning its handler."""
    harness = _build_harness(config, tmp_path, agent_name=ROUTER_AGENT_NAME)
    room = _room_with_members(config, ROUTER_AGENT_NAME, "general")
    event = _text_event("!model default", event_id="$model-command-replay:localhost")
    command = command_parser.parse(event.body)
    assert command is not None
    target = harness.controller.deps.resolver.build_message_target(
        room_id=room.room_id,
        thread_id=None,
        reply_to_event_id=event.event_id,
        event_source=event.source,
    )
    handled_turn = TurnRecord.create([event.event_id])

    with (
        patch("mindroom.commands.handler.handle_model_command", return_value="✅ Model changed") as mutate,
        patch.object(harness.turn_store, "record_responded_turn"),
    ):
        await harness.controller.deps.command_executor.execute(
            room,
            event,
            _SENDER,
            command,
            target=target,
            handled_turn=handled_turn,
        )

    await harness.controller.deps.command_executor.execute(
        room,
        event,
        _SENDER,
        command,
        target=target,
        handled_turn=handled_turn,
    )

    mutate.assert_called_once()
    assert len(harness.gateway.sent) == 1


@pytest.mark.asyncio
async def test_command_send_failure_keeps_turn_pending(config: Config, tmp_path: Path) -> None:
    """A command with no visible Matrix event must remain retry-owned."""
    harness = _build_harness(config, tmp_path, agent_name=ROUTER_AGENT_NAME)
    room = _room_with_members(config, ROUTER_AGENT_NAME, "general")
    event = _text_event("!help", event_id="$command-send-failure:localhost")
    command = command_parser.parse(event.body)
    assert command is not None
    target = harness.controller.deps.resolver.build_message_target(
        room_id=room.room_id,
        thread_id=None,
        reply_to_event_id=event.event_id,
        event_source=event.source,
    )

    async def fail_send(_request: SendTextRequest) -> None:
        return None

    with (
        patch.object(harness.gateway, "send_text", new=fail_send),
        pytest.raises(RuntimeError, match="did not return a Matrix event ID"),
    ):
        await harness.controller.deps.command_executor.execute(
            room,
            event,
            _SENDER,
            command,
            target=target,
            handled_turn=TurnRecord.create([event.event_id]),
        )

    record = harness.turn_store.get_turn_record(event.event_id)
    assert record is not None
    assert record.completed is False
    assert record.command_result_text is None


@pytest.mark.asyncio
async def test_mutating_command_retries_checkpointed_result_without_reapplying(config: Config, tmp_path: Path) -> None:
    """A failed response send must replay the exact result, not repeat its mutation."""
    harness = _build_harness(config, tmp_path, agent_name=ROUTER_AGENT_NAME)
    room = _room_with_members(config, ROUTER_AGENT_NAME, "general")
    event = _text_event("!model default", event_id="$model-send-failure:localhost")
    command = command_parser.parse(event.body)
    assert command is not None
    target = harness.controller.deps.resolver.build_message_target(
        room_id=room.room_id,
        thread_id=None,
        reply_to_event_id=event.event_id,
        event_source=event.source,
    )

    async def fail_send(_request: SendTextRequest) -> None:
        return None

    with patch("mindroom.commands.handler.handle_model_command", return_value="✅ Model changed") as mutate:
        with (
            patch.object(harness.gateway, "send_text", new=fail_send),
            pytest.raises(RuntimeError, match="did not return a Matrix event ID"),
        ):
            await harness.controller.deps.command_executor.execute(
                room,
                event,
                _SENDER,
                command,
                target=target,
                handled_turn=TurnRecord.create([event.event_id]),
            )

        pending_turn = harness.turn_store.get_turn_record(event.event_id)
        assert pending_turn is not None
        assert pending_turn.command_execution_started
        assert pending_turn.command_result_text == "✅ Model changed"

        await harness.controller.deps.command_executor.execute(
            room,
            event,
            _SENDER,
            command,
            target=target,
            handled_turn=pending_turn,
        )

    mutate.assert_called_once()
    assert harness.gateway.sent[0].response_text == "✅ Model changed"
    assert harness.turn_store.is_handled(event.event_id)


@pytest.mark.asyncio
async def test_failed_mutating_command_execution_requires_manual_retry(config: Config, tmp_path: Path) -> None:
    """A failed admitted attempt must report uncertainty instead of rerunning automatically."""
    harness = _build_harness(config, tmp_path, agent_name=ROUTER_AGENT_NAME)
    room = _room_with_members(config, ROUTER_AGENT_NAME, "general")
    event = _text_event("!model default", event_id="$model-ambiguous:localhost")
    command = command_parser.parse(event.body)
    assert command is not None
    target = harness.controller.deps.resolver.build_message_target(
        room_id=room.room_id,
        thread_id=None,
        reply_to_event_id=event.event_id,
        event_source=event.source,
    )
    with (
        patch(
            "mindroom.commands.handler.handle_model_command",
            side_effect=RuntimeError("transient failure"),
        ) as attempt,
        pytest.raises(RuntimeError, match="transient failure"),
    ):
        await harness.controller.deps.command_executor.execute(
            room,
            event,
            _SENDER,
            command,
            target=target,
            handled_turn=TurnRecord.create([event.event_id]),
        )

    pending_turn = harness.turn_store.get_turn_record(event.event_id)
    assert pending_turn is not None
    assert pending_turn.command_execution_started
    assert pending_turn.command_result_text is None

    with patch("mindroom.commands.handler.handle_model_command") as replay:
        await harness.controller.deps.command_executor.execute(
            room,
            event,
            _SENDER,
            command,
            target=target,
            handled_turn=pending_turn,
        )

    attempt.assert_called_once()
    replay.assert_not_called()
    assert "outcome is uncertain" in harness.gateway.sent[0].response_text
    assert harness.turn_store.is_handled(event.event_id)


@pytest.mark.asyncio
async def test_rejection_replay_adopts_durable_response(config: Config, tmp_path: Path) -> None:
    """A rejected-turn replay must not send a second visible rejection."""
    harness = _build_harness(config, tmp_path)
    room = _room_with_members(config, "general")
    event = _text_event("unsupported request", event_id="$reject-replay:localhost")
    target = MessageTarget.resolve(room.room_id, None, event.event_id)
    envelope = harness.controller.deps.resolver.build_ingress_envelope(
        event=event,
        requester_user_id=_SENDER,
        target=target,
    )
    dispatch = PreparedDispatch(
        requester_user_id=_SENDER,
        context=MessageContext(
            am_i_mentioned=True,
            is_thread=False,
            thread_id=None,
            thread_history=(),
            mentioned_agents=[],
            has_non_agent_mentions=False,
        ),
        target=target,
        correlation_id=event.event_id,
        envelope=envelope,
    )
    action = ResponseAction(kind="reject", rejection_message="Unsupported request")
    handled_turn = harness.turn_store.attach_response_context(
        TurnRecord.create([event.event_id], completed=False),
        history_scope=None,
        conversation_target=target,
    )
    pending_turn = harness.turn_store.record_pending_turn(handled_turn)
    assert pending_turn is not None

    with patch.object(harness.turn_store, "record_responded_turn"):
        await harness.controller._execute_response_action(
            room,
            event,
            dispatch,
            action,
            DispatchPayloadInputs((), (), ()),
            processing_log="rejecting",
            dispatch_started_at=0.0,
            handled_turn=pending_turn,
        )

    pending_turn = harness.turn_store.get_turn_record(event.event_id)
    assert pending_turn is not None
    assert pending_turn.completed is False
    assert pending_turn.response_event_id == "$sent-1:localhost"

    await harness.controller._execute_response_action(
        room,
        event,
        dispatch,
        action,
        DispatchPayloadInputs((), (), ()),
        processing_log="rejecting",
        dispatch_started_at=0.0,
        handled_turn=pending_turn,
        reconcile_visible_response=True,
    )

    assert len(harness.gateway.sent) == 1
    record = harness.turn_store.get_turn_record(event.event_id)
    assert record is not None
    assert record.completed is True
    assert record.response_event_id == "$sent-1:localhost"


@pytest.mark.asyncio
async def test_router_relay_replay_adopts_durable_response(config: Config, tmp_path: Path) -> None:
    """A router replay must reuse the relay persisted before terminal settlement."""
    harness = _build_harness(config, tmp_path, agent_name=ROUTER_AGENT_NAME)
    room = _room_with_members(config, ROUTER_AGENT_NAME, "general")
    event = _text_event("route this", event_id="$router-replay:localhost")
    handled_turn = TurnRecord.create([event.event_id], completed=False)

    with patch.object(harness.turn_store, "record_responded_turn"):
        await harness.controller._execute_router_relay(
            room,
            event,
            (),
            requester_user_id=_SENDER,
            handled_turn=handled_turn,
        )

    pending_turn = harness.turn_store.get_turn_record(event.event_id)
    assert pending_turn is not None
    assert pending_turn.completed is False
    assert pending_turn.response_event_id == "$sent-1:localhost"

    await harness.controller._execute_router_relay(
        room,
        event,
        (),
        requester_user_id=_SENDER,
        handled_turn=handled_turn,
    )

    assert len(harness.gateway.sent) == 1
    record = harness.turn_store.get_turn_record(event.event_id)
    assert record is not None
    assert record.completed is True
    assert record.response_event_id == "$sent-1:localhost"


@pytest.mark.asyncio
async def test_command_dispatch_failure_propagates_without_recording_terminal_outcome(
    config: Config,
    tmp_path: Path,
) -> None:
    """A failed owner command must retain only retryable pending intent."""
    harness = _build_harness(config, tmp_path, agent_name=ROUTER_AGENT_NAME)
    room = _room_with_members(config, ROUTER_AGENT_NAME, "general", "research")
    event = _text_event("!help", event_id="$failed-command:localhost")

    with (
        patch("mindroom.command_turn_executor.handle_command", side_effect=RuntimeError("command failed")),
        pytest.raises(RuntimeError, match="command failed"),
    ):
        await harness.deliver(room, event)

    record = harness.turn_store.get_turn_record(event.event_id)
    assert record is not None
    assert record.completed is False
    assert record.response_event_id is None


@pytest.mark.asyncio
async def test_non_router_agent_consumes_command_without_responding(config: Config, tmp_path: Path) -> None:
    """Plain agents stay silent on command turns; the router owns command replies."""
    harness = _build_harness(config, tmp_path, agent_name="general")
    room = _room_with_members(config, "general")
    event = _text_event("!help", event_id="$command:localhost")

    await harness.deliver(room, event)

    assert harness.policy.plan_turn_calls == 0
    assert harness.runner.requests == []
    assert harness.gateway.sent == []
    # Commands are control inputs: even a silently consumed one never enters the gate.
    assert harness.gate_batches == []
    # Exact callback truth stays compact when only the router owns the visible reply.
    assert harness.turn_store.is_handled(event.event_id) is False
    assert harness.ignored_dispatch_sources == [(event.event_id,)]


@pytest.mark.asyncio
async def test_normal_desktop_agent_owns_desktop_command(tmp_path: Path) -> None:
    """A requester can pair Desktop directly in a normal agent's private room."""
    config = bind_runtime_paths(
        Config(
            agents={
                "computer": AgentConfig(
                    display_name="Computer",
                    tools=["desktop"],
                ),
            },
        ),
        test_runtime_paths(tmp_path / "runtime"),
    )
    harness = _build_harness(config, tmp_path, agent_name="computer")
    room = _room_with_members(config, "computer")
    event = _text_event("!desktop status", event_id="$desktop-command:localhost")

    await harness.deliver(room, event)

    assert harness.policy.plan_turn_calls == 0
    assert harness.runner.requests == []
    assert harness.gate_batches == []
    assert len(harness.gateway.sent) == 1
    assert "Desktop setup is required" in harness.gateway.sent[0].response_text
    assert harness.turn_store.is_handled(event.event_id) is True


@pytest.mark.asyncio
async def test_same_turn_after_restart_produces_exactly_one_response(config: Config, tmp_path: Path) -> None:
    """Replaying one handled turn through a fresh controller never responds twice.

    The second harness shares only the on-disk handled-turn ledger, simulating a
    process restart that re-delivers the same Matrix event.
    """
    room = _room_with_members(config, "general")
    event = _text_event("respond exactly once")

    first_run = _build_harness(config, tmp_path)
    await first_run.deliver(room, event)
    assert len(first_run.runner.requests) == 1
    assert first_run.turn_store.is_handled(event.event_id) is True

    restarted_run = _build_harness(config, tmp_path)
    await restarted_run.deliver(room, _text_event("respond exactly once"))

    assert restarted_run.runner.requests == []
    assert restarted_run.gateway.sent == []
    assert restarted_run.policy.plan_turn_calls == 0
    assert restarted_run.turn_store.is_handled(event.event_id) is True


@pytest.mark.asyncio
async def test_interactive_selection_acks_generates_and_records_once(config: Config, tmp_path: Path) -> None:
    """The controller-owned selection path acks, runs generation, and records the turn."""
    harness = _build_harness(config, tmp_path)
    room = nio.MatrixRoom(_ROOM_ID, _entity_user_id(config, "general"))
    selection = interactive.InteractiveSelection(
        question_event_id="$question:localhost",
        question_text="Which option should I use?",
        selection_key="1",
        selected_label="Option 1",
        selected_value="Option 1",
        thread_id="$thread-root:localhost",
    )

    await harness.controller.handle_interactive_selection(
        room,
        selection=selection,
        user_id=_SENDER,
        source_event_id="$selection:localhost",
    )

    assert len(harness.gateway.sent) == 1
    ack_request = harness.gateway.sent[0]
    assert ack_request.response_text.startswith("You selected: 1 Option 1")
    assert ack_request.target.resolved_thread_id == selection.thread_id
    assert ack_request.target.reply_to_event_id == selection.question_event_id

    assert len(harness.runner.requests) == 1
    request = harness.runner.requests[0]
    assert request.prompt == interactive.build_selection_prompt(selection)
    assert request.existing_event_id == "$sent-1:localhost"
    assert request.existing_event_is_placeholder is True
    assert request.response_envelope.target.reply_to_event_id == selection.question_event_id
    assert request.response_envelope.target.resolved_thread_id == selection.thread_id
    metadata = request.matrix_run_metadata
    assert metadata is not None
    assert metadata[constants.MATRIX_SOURCE_EVENT_IDS_METADATA_KEY] == [selection.question_event_id]
    assert metadata[constants.MATRIX_TURN_DISCOVERY_EVENT_IDS_METADATA_KEY] == ["$selection:localhost"]

    assert harness.turn_store.is_handled(selection.question_event_id) is True
    assert harness.turn_store.is_handled("$selection:localhost") is True


@pytest.mark.asyncio
async def test_interactive_selection_replay_adopts_durable_ack(config: Config, tmp_path: Path) -> None:
    """A retry after visible delivery must reuse the first acknowledgment instead of sending again."""
    harness = _build_harness(config, tmp_path)
    harness.runner.response_event_id = None
    room = nio.MatrixRoom(_ROOM_ID, _entity_user_id(config, "general"))
    selection = interactive.InteractiveSelection(
        question_event_id="$question:localhost",
        question_text="Which option should I use?",
        selection_key="1",
        selected_label="Option 1",
        selected_value="Option 1",
        thread_id="$thread-root:localhost",
    )
    selection_event_id = "$selection:localhost"

    with pytest.raises(RuntimeError, match="no durable terminal outcome"):
        await harness.controller.handle_interactive_selection(
            room,
            selection=selection,
            user_id=_SENDER,
            source_event_id=selection_event_id,
        )

    pending_turn = harness.turn_store.get_turn_record(selection.question_event_id)
    assert pending_turn is not None
    assert pending_turn.completed is False
    assert pending_turn.response_event_id == "$sent-1:localhost"

    harness.runner.response_event_id = "$sent-1:localhost"
    await harness.controller.handle_interactive_selection(
        room,
        selection=selection,
        user_id=_SENDER,
        source_event_id=selection_event_id,
    )

    assert len(harness.gateway.sent) == 1
    assert len(harness.runner.requests) == 2
    assert harness.runner.requests[1].existing_event_id == "$sent-1:localhost"
    assert harness.turn_store.is_handled(selection.question_event_id) is True


@pytest.mark.asyncio
async def test_interactive_selection_persistence_failure_prevents_ack_and_generation(
    config: Config,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """A failed pending-context write must escape before any visible or persisted response."""
    harness = _build_harness(config, tmp_path)
    room = nio.MatrixRoom(_ROOM_ID, _entity_user_id(config, "general"))
    selection = interactive.InteractiveSelection(
        question_event_id="$question:localhost",
        question_text="Which option should I use?",
        selection_key="1",
        selected_label="Option 1",
        selected_value="Option 1",
        thread_id="$thread-root:localhost",
    )
    real_persist = harness.turn_store._ledger._persist_records

    def fail_pending_persist(turn_records: tuple[TurnRecord, ...]) -> None:
        if any(
            selection.question_event_id in record.indexed_event_ids and not record.completed for record in turn_records
        ):
            msg = "pending context write failed"
            raise OSError(msg)
        real_persist(turn_records)

    monkeypatch.setattr(harness.turn_store._ledger, "_persist_records", fail_pending_persist)

    with pytest.raises(OSError, match="pending context write failed"):
        await harness.controller.handle_interactive_selection(
            room,
            selection=selection,
            user_id=_SENDER,
            source_event_id="$selection:localhost",
        )

    assert harness.gateway.sent == []
    assert harness.runner.requests == []


@pytest.mark.asyncio
async def test_interactive_commit_failure_restores_selection_claim(
    config: Config,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Commit persistence failure must restore interactive ownership for retry."""
    harness = _build_harness(config, tmp_path)
    room = nio.MatrixRoom(_ROOM_ID, _entity_user_id(config, "general"))
    selection = interactive.InteractiveSelection(
        question_event_id="$question:localhost",
        question_text="Which option should I use?",
        selection_key="1",
        selected_label="Option 1",
        selected_value="Option 1",
        thread_id="$thread-root:localhost",
    )
    restored: list[interactive.InteractiveSelection] = []

    async def execute_selection(*_args: object, **_kwargs: object) -> None:
        return None

    def fail_commit(_selection: interactive.InteractiveSelection) -> None:
        msg = "selection commit failed"
        raise OSError(msg)

    monkeypatch.setattr(harness.controller, "_execute_interactive_selection", execute_selection)
    monkeypatch.setattr(interactive, "commit_selection", fail_commit)
    monkeypatch.setattr(interactive, "restore_selection", restored.append)

    with pytest.raises(OSError, match="selection commit failed"):
        await harness.controller.handle_interactive_selection(
            room,
            selection=selection,
            user_id=_SENDER,
            source_event_id="$selection:localhost",
        )

    assert restored == [selection]


@pytest.mark.asyncio
async def test_interactive_selection_claim_conflict_accepts_racing_terminal_turn(
    config: Config,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """A completed owner that wins the pending-record race is a clean terminal outcome."""
    harness = _build_harness(config, tmp_path)
    room = nio.MatrixRoom(_ROOM_ID, _entity_user_id(config, "general"))
    selection = interactive.InteractiveSelection(
        question_event_id="$question:localhost",
        question_text="Which option should I use?",
        selection_key="1",
        selected_label="Option 1",
        selected_value="Option 1",
        thread_id="$thread-root:localhost",
    )
    real_record_pending = harness.turn_store.record_pending_turn

    def lose_claim_race(_turn_record: TurnRecord) -> None:
        monkeypatch.setattr(harness.turn_store, "record_pending_turn", real_record_pending)
        harness.turn_store.record_turn(
            TurnRecord.create(
                [selection.question_event_id],
                response_event_id="$racing-response:localhost",
            ),
        )
        harness.turn_store._ledger.flush()

    monkeypatch.setattr(harness.turn_store, "record_pending_turn", lose_claim_race)

    await harness.controller.handle_interactive_selection(
        room,
        selection=selection,
        user_id=_SENDER,
        source_event_id="$selection:localhost",
    )

    assert harness.gateway.sent == []
    assert harness.runner.requests == []


@pytest.mark.asyncio
async def test_interactive_selection_replacement_refusal_uses_checkpoint_replay(
    config: Config,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """A replaced runtime must bubble refusal so its Matrix event is replayed."""
    harness = _build_harness(config, tmp_path)
    room = nio.MatrixRoom(_ROOM_ID, _entity_user_id(config, "general"))
    selection = interactive.InteractiveSelection(
        question_event_id="$question:localhost",
        question_text="Which option should I use?",
        selection_key="1",
        selected_label="Option 1",
        selected_value="Option 1",
        thread_id="$thread-root:localhost",
    )
    selection_event_id = "$selection:localhost"

    async def refuse(request: ResponseRequest) -> str | None:
        harness.runner.requests.append(request)
        raise ResponseAdmissionRefusedError

    monkeypatch.setattr(harness.runner, "generate_response", refuse)

    with pytest.raises(ResponseAdmissionRefusedError):
        await harness.controller.handle_interactive_selection(
            room,
            selection=selection,
            user_id=_SENDER,
            source_event_id=selection_event_id,
        )

    # Only the durably tracked acknowledgement was sent. Refusal performs no
    # untimed Matrix settlement inside replacement shutdown.
    assert len(harness.gateway.sent) == 1
    assert harness.gateway.edited == []

    record = harness.turn_store.get_turn_record(selection_event_id)
    assert record is not None
    assert record.response_event_id == "$sent-1:localhost"
    assert record.completed is False


@pytest.mark.asyncio
async def test_interactive_selection_redacted_after_ack_is_suppressed_under_lock(
    config: Config,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """The selection must be pending before ack and recheck aliases at response startup."""
    harness = _build_harness(config, tmp_path)
    room = nio.MatrixRoom(_ROOM_ID, _entity_user_id(config, "general"))
    selection = interactive.InteractiveSelection(
        question_event_id="$question:localhost",
        question_text="Which option should I use?",
        selection_key="1",
        selected_label="Option 1",
        selected_value="Option 1",
        thread_id="$thread-root:localhost",
    )
    selection_event_id = "$selection:localhost"

    async def send_ack_then_redact(request: SendTextRequest) -> str:
        harness.gateway.sent.append(request)
        marked = harness.turn_store.mark_source_redacted(selection_event_id)
        assert marked is not None
        assert marked.conversation_target is not None
        assert marked.history_scope is not None
        return "$ack:localhost"

    monkeypatch.setattr(harness.gateway, "send_text", send_ack_then_redact)

    await harness.controller.handle_interactive_selection(
        room,
        selection=selection,
        user_id=_SENDER,
        source_event_id=selection_event_id,
    )

    assert len(harness.runner.requests) == 1
    record = harness.turn_store.get_turn_record(selection_event_id)
    assert record is not None
    assert record.redacted_source_event_ids == (selection_event_id,)
    assert record.pending_redaction_cleanup_event_ids == ()
    assert record.response_event_id == "$ack:localhost"
    assert harness.turn_store.is_handled(selection_event_id) is True
    assert harness.turn_store.is_handled(selection.question_event_id) is False


@pytest.mark.asyncio
async def test_interactive_selection_rehydrates_attachment_context_from_thread(
    config: Config,
    tmp_path: Path,
) -> None:
    """A selection callback turn reaches the attachments of the conversation that asked the question.

    The selection is a synthetic turn with no Matrix message of its own, so it
    must rebuild the attachment context from the originating thread; before the
    fix the callback request carried no attachment IDs and ``get_attachment``
    rejected IDs that were available when the question was asked.
    """
    harness = _build_harness(config, tmp_path)
    media_path = tmp_path / "incoming_media" / "report.pdf"
    media_path.parent.mkdir(parents=True, exist_ok=True)
    media_path.write_bytes(b"%PDF-1.4 fake report")
    record = register_local_attachment(
        tmp_path,
        media_path,
        kind="file",
        filename="report.pdf",
        room_id=_ROOM_ID,
        thread_id="$thread-root:localhost",
        sender=_SENDER,
    )
    assert record is not None
    triggering_message = make_visible_message(
        sender=_SENDER,
        body="here is the report",
        content={
            "msgtype": "m.text",
            "body": "here is the report",
            constants.ATTACHMENT_IDS_KEY: [record.attachment_id],
        },
        thread_id="$thread-root:localhost",
    )
    harness.conversation_cache.get_strict_thread_history.return_value = thread_history_result(
        [triggering_message],
        is_full_history=True,
    )
    room = nio.MatrixRoom(_ROOM_ID, _entity_user_id(config, "general"))
    selection = interactive.InteractiveSelection(
        question_event_id="$question:localhost",
        question_text="Process the attached report?",
        selection_key="1",
        selected_label="Yes",
        selected_value="Yes",
        thread_id="$thread-root:localhost",
    )

    await harness.controller.handle_interactive_selection(
        room,
        selection=selection,
        user_id=_SENDER,
        source_event_id="$selection:localhost",
    )

    assert len(harness.runner.requests) == 1
    request = harness.runner.requests[0]
    assert request.attachment_ids == (record.attachment_id,)
    assert request.response_envelope.attachment_ids == (record.attachment_id,)


@pytest.mark.asyncio
@pytest.mark.parametrize(("edit_succeeds", "expected_send_count"), [(True, 1), (False, 2)])
async def test_interactive_selection_attachment_setup_failure_finalizes_ack(
    config: Config,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
    edit_succeeds: bool,
    expected_send_count: int,
) -> None:
    """An attachment-resolution failure visibly terminates the processing acknowledgment."""
    harness = _build_harness(config, tmp_path)
    harness.gateway.edit_succeeds = edit_succeeds

    async def fail_attachment_resolution(
        _normalizer: InboundTurnNormalizer,
        _request: object,
    ) -> object:
        msg = "attachment lookup failed"
        raise RuntimeError(msg)

    monkeypatch.setattr(
        InboundTurnNormalizer,
        "build_dispatch_payload_with_attachments",
        fail_attachment_resolution,
    )
    room = nio.MatrixRoom(_ROOM_ID, _entity_user_id(config, "general"))
    selection = interactive.InteractiveSelection(
        question_event_id="$question:localhost",
        question_text="Process the attached report?",
        selection_key="1",
        selected_label="Yes",
        selected_value="Yes",
        thread_id="$thread-root:localhost",
    )

    await harness.controller.handle_interactive_selection(
        room,
        selection=selection,
        user_id=_SENDER,
        source_event_id="$selection:localhost",
    )

    assert harness.runner.requests == []
    assert len(harness.gateway.sent) == expected_send_count
    assert len(harness.gateway.edited) == 1
    edit_request = harness.gateway.edited[0]
    assert edit_request.event_id == "$sent-1:localhost"
    assert edit_request.new_text == "[general] ⚠️ Error: attachment lookup failed"
    assert edit_request.extra_content == {constants.STREAM_STATUS_KEY: constants.STREAM_STATUS_COMPLETED}
    if not edit_succeeds:
        fallback_request = harness.gateway.sent[1]
        assert fallback_request.response_text == edit_request.new_text
        assert fallback_request.extra_content == edit_request.extra_content
    handled_turn = harness.turn_store.get_turn_record(selection.question_event_id)
    assert handled_turn is not None
    assert handled_turn.response_event_id == ("$sent-1:localhost" if edit_succeeds else "$sent-2:localhost")
    assert harness.turn_store.is_handled(selection.question_event_id) is True
    assert harness.turn_store.is_handled("$selection:localhost") is True


@pytest.mark.asyncio
async def test_interactive_selection_failure_persists_fallback_before_terminal_record(
    config: Config,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """A crash after fallback delivery must leave its exact event ID recoverable."""
    harness = _build_harness(config, tmp_path)
    harness.gateway.edit_succeeds = False

    async def fail_attachment_resolution(
        _normalizer: InboundTurnNormalizer,
        _request: object,
    ) -> object:
        msg = "attachment lookup failed"
        raise RuntimeError(msg)

    def crash_before_terminal_record(_handled_turn: TurnRecord) -> None:
        msg = "simulated crash"
        raise RuntimeError(msg)

    monkeypatch.setattr(
        InboundTurnNormalizer,
        "build_dispatch_payload_with_attachments",
        fail_attachment_resolution,
    )
    monkeypatch.setattr(harness.turn_store, "record_responded_turn", crash_before_terminal_record)
    room = nio.MatrixRoom(_ROOM_ID, _entity_user_id(config, "general"))
    selection = interactive.InteractiveSelection(
        question_event_id="$question:localhost",
        question_text="Process the attached report?",
        selection_key="1",
        selected_label="Yes",
        selected_value="Yes",
        thread_id="$thread-root:localhost",
    )

    with pytest.raises(RuntimeError, match="simulated crash"):
        await harness.controller.handle_interactive_selection(
            room,
            selection=selection,
            user_id=_SENDER,
            source_event_id="$selection:localhost",
        )

    pending_turn = harness.turn_store.get_turn_record(selection.question_event_id)
    assert pending_turn is not None
    assert pending_turn.completed is False
    assert pending_turn.response_event_id == "$sent-2:localhost"


@pytest.mark.asyncio
async def test_interactive_selection_without_response_stays_retryable(config: Config, tmp_path: Path) -> None:
    """A selection whose generation yields no visible response must not be marked handled."""
    harness = _build_harness(config, tmp_path)
    harness.runner.response_event_id = None
    room = nio.MatrixRoom(_ROOM_ID, _entity_user_id(config, "general"))
    selection = interactive.InteractiveSelection(
        question_event_id="$question:localhost",
        question_text="Which option should I use?",
        selection_key="2",
        selected_label="Option 2",
        selected_value="Option 2",
        thread_id="$thread-root:localhost",
    )

    with pytest.raises(RuntimeError, match="no durable terminal outcome"):
        await harness.controller.handle_interactive_selection(
            room,
            selection=selection,
            user_id=_SENDER,
            source_event_id="$selection:localhost",
        )

    assert len(harness.runner.requests) == 1
    assert harness.turn_store.is_handled(selection.question_event_id) is False
    assert harness.turn_store.is_handled("$selection:localhost") is False


@pytest.mark.asyncio
async def test_interactive_selection_interruption_registers_exact_source_before_settlement(
    config: Config,
    tmp_path: Path,
) -> None:
    """A landed interruption must register the selection event before its durable handled write."""
    harness = _build_harness(config, tmp_path)
    harness.runner.deferred_sync_restart_error = asyncio.CancelledError("sync_restart")
    room = nio.MatrixRoom(_ROOM_ID, _entity_user_id(config, "general"))
    selection = interactive.InteractiveSelection(
        question_event_id="$question:localhost",
        question_text="Which option should I use?",
        selection_key="2",
        selected_label="Option 2",
        selected_value="Option 2",
        thread_id="$thread-root:localhost",
    )
    selection_event_id = "$selection:localhost"

    with pytest.raises(asyncio.CancelledError, match="sync_restart"):
        await harness.controller.handle_interactive_selection(
            room,
            selection=selection,
            user_id=_SENDER,
            source_event_id=selection_event_id,
        )

    assert harness.interrupted_turn_rooms.contains(selection_event_id)
    record = harness.turn_store.get_turn_record(selection_event_id)
    assert record is not None
    assert record.response_event_id == "$response:localhost"
    assert record.completed is True
