"""Test that the 🛑 emoji can be reused for other purposes when not stopping generation."""

from __future__ import annotations

import asyncio
import json
from pathlib import Path  # noqa: TC003
from unittest.mock import AsyncMock, MagicMock, patch

import nio
import pytest

from mindroom.bot import AgentBot
from mindroom.cancellation import USER_STOP_CANCEL_MSG
from mindroom.config.main import Config
from mindroom.handled_turns import TurnRecord
from mindroom.logging_config import setup_logging
from mindroom.matrix.users import AgentMatrixUser
from mindroom.message_target import MessageTarget
from mindroom.stop import StopManager
from tests.bot_helpers import dispatch_reaction_durably
from tests.conftest import (
    bind_runtime_paths,
    install_send_response_mock,
    orchestrator_runtime_paths,
    runtime_paths_for,
    test_runtime_paths,
)
from tests.identity_helpers import entity_ids, persist_entity_accounts


async def _drain_stop_cleanup(stop_manager: StopManager) -> None:
    """Wait for any background stop-manager cleanup tasks."""
    if stop_manager.cleanup_tasks:
        await asyncio.gather(*list(stop_manager.cleanup_tasks), return_exceptions=True)


def _stop_test_config(tmp_path: Path, *, include_helper: bool = False) -> Config:
    agents: dict[str, dict[str, object]] = {
        "test_agent": {"display_name": "Test Agent", "rooms": ["!test:example.com"]},
    }
    if include_helper:
        agents["helper"] = {"display_name": "Helper Agent", "rooms": ["!test:example.com"]}
    config = bind_runtime_paths(
        Config(agents=agents, authorization={"default_room_access": True}),
        test_runtime_paths(tmp_path),
    )
    persist_entity_accounts(config, runtime_paths_for(config))
    return config


def _stop_test_agent_user(config: Config) -> AgentMatrixUser:
    matrix_id = entity_ids(config, runtime_paths_for(config))["test_agent"]
    return AgentMatrixUser(
        agent_name="test_agent",
        user_id=matrix_id.full_id,
        display_name="Test Agent",
        password="test_password",  # noqa: S106
    )


def _record_pending_response(bot: AgentBot, message_id: str, target: MessageTarget) -> None:
    """Mirror the durable response intent that owns every real stop button."""
    bot._turn_store.record_pending_turn(
        TurnRecord.create(
            [f"{message_id}-source"],
            response_event_id=message_id,
            completed=False,
            response_owner=bot.agent_name,
            requester_id="@user:example.com",
            conversation_target=target,
        ),
    )
    bot._delivery_gateway.finalize_user_stopped_response = AsyncMock(return_value=True)
    bot._dispatch_obligation_runner.receipt_order = AsyncMock(return_value=1)


@pytest.mark.asyncio
async def test_stop_emoji_only_stops_during_generation(tmp_path: Path) -> None:
    """Test that 🛑 reaction only acts as stop button during message generation."""
    config = _stop_test_config(tmp_path)
    agent_user = _stop_test_agent_user(config)

    bot = AgentBot(
        agent_user=agent_user,
        storage_path=tmp_path,
        config=config,
        runtime_paths=runtime_paths_for(config),
        rooms=["!test:example.com"],
    )

    # Set up the bot with necessary mocks
    bot.client = AsyncMock(spec=nio.AsyncClient)
    bot.client.user_id = agent_user.user_id
    bot.logger = MagicMock()
    send_response = AsyncMock(return_value="$stopping:example.com")
    install_send_response_mock(bot, send_response)

    # Create a room and reaction event
    room = nio.MatrixRoom(room_id="!test:example.com", own_user_id=agent_user.user_id)

    # Create a 🛑 reaction event
    reaction_event = nio.ReactionEvent.from_dict(
        {
            "content": {
                "m.relates_to": {
                    "rel_type": "m.annotation",
                    "event_id": "$message:example.com",
                    "key": "🛑",
                },
            },
            "event_id": "$reaction:example.com",
            "sender": "@user:example.com",
            "origin_server_ts": 1000000,
            "type": "m.reaction",
            "room_id": "!test:example.com",
        },
    )

    # Mock interactive.handle_reaction so the test only exercises stop-vs-fallthrough behavior.
    with patch("mindroom.bot.interactive.handle_reaction") as mock_handle_reaction:
        mock_handle_reaction.return_value = None

        # Case 1: Message is NOT being generated - should handle as interactive
        await dispatch_reaction_durably(bot, room, reaction_event)

        # Should have called interactive.handle_reaction since message wasn't being tracked
        mock_handle_reaction.assert_called_once()

        # Reset the mock
        mock_handle_reaction.reset_mock()

        # Case 2: Message IS being generated - should handle as stop button
        # Track a message as being generated
        task = MagicMock()  # Use MagicMock instead of AsyncMock for the task
        task.done = MagicMock(return_value=False)  # done() is a regular method, not async
        target = MessageTarget.resolve("!test:example.com", None, "$message:example.com")
        bot.stop_manager.set_current(
            message_id="$message:example.com",
            target=target,
            task=task,
        )
        _record_pending_response(bot, "$message:example.com", target)

        # A second physical reaction reaches the same STOP target.
        active_reaction_event = nio.ReactionEvent.from_dict(
            {
                "content": reaction_event.source["content"],
                "event_id": "$active-reaction:example.com",
                "sender": reaction_event.sender,
                "origin_server_ts": 1000001,
                "type": "m.reaction",
                "room_id": room.room_id,
            },
        )
        await dispatch_reaction_durably(bot, room, active_reaction_event)

        # Should NOT have called interactive.handle_reaction since it was handled as stop
        mock_handle_reaction.assert_not_called()
        send_response.assert_not_awaited()

        # The task should have been cancelled
        task.cancel.assert_called_once_with(msg=USER_STOP_CANCEL_MSG)


@pytest.mark.asyncio
async def test_stop_emoji_hard_cancels_and_schedules_agno_cleanup_when_run_id_present(tmp_path: Path) -> None:
    """Tracked Agno runs should hard-cancel immediately and clean up Agno state in the background."""
    config = _stop_test_config(tmp_path)
    agent_user = _stop_test_agent_user(config)

    bot = AgentBot(
        agent_user=agent_user,
        storage_path=tmp_path,
        config=config,
        runtime_paths=runtime_paths_for(config),
        rooms=["!test:example.com"],
    )

    bot.client = AsyncMock(spec=nio.AsyncClient)
    bot.client.user_id = agent_user.user_id
    bot.logger = MagicMock()
    send_response = AsyncMock(return_value="$stopping:example.com")
    install_send_response_mock(bot, send_response)

    room = nio.MatrixRoom(room_id="!test:example.com", own_user_id=agent_user.user_id)
    reaction_event = nio.ReactionEvent.from_dict(
        {
            "content": {
                "m.relates_to": {
                    "rel_type": "m.annotation",
                    "event_id": "$message:example.com",
                    "key": "🛑",
                },
            },
            "event_id": "$reaction:example.com",
            "sender": "@user:example.com",
            "origin_server_ts": 1000000,
            "type": "m.reaction",
            "room_id": "!test:example.com",
        },
    )

    task = MagicMock()
    task.done = MagicMock(return_value=False)
    target = MessageTarget.resolve("!test:example.com", None, "$message:example.com")
    bot.stop_manager.set_current(
        message_id="$message:example.com",
        target=target,
        task=task,
        run_id="run-123",
    )
    _record_pending_response(bot, "$message:example.com", target)

    with patch.object(bot.stop_manager, "_schedule_graceful_run_cancel") as mock_schedule_cancel:
        await dispatch_reaction_durably(bot, room, reaction_event)

    mock_schedule_cancel.assert_called_once_with("$message:example.com", "run-123")
    task.cancel.assert_called_once_with(msg=USER_STOP_CANCEL_MSG)
    send_response.assert_not_awaited()


@pytest.mark.asyncio
async def test_stop_emoji_threaded_target_sends_no_acknowledgement(tmp_path: Path) -> None:
    """Threaded stop reactions should not send a separate acknowledgement message."""
    config = _stop_test_config(tmp_path)
    agent_user = _stop_test_agent_user(config)

    bot = AgentBot(
        agent_user=agent_user,
        storage_path=tmp_path,
        config=config,
        runtime_paths=runtime_paths_for(config),
        rooms=["!test:example.com"],
    )

    bot.client = AsyncMock(spec=nio.AsyncClient)
    bot.client.user_id = agent_user.user_id
    bot.logger = MagicMock()
    send_response = AsyncMock(return_value="$stopping:example.com")
    install_send_response_mock(bot, send_response)

    room = nio.MatrixRoom(room_id="!test:example.com", own_user_id=agent_user.user_id)
    reaction_event = nio.ReactionEvent.from_dict(
        {
            "content": {
                "m.relates_to": {
                    "rel_type": "m.annotation",
                    "event_id": "$message:example.com",
                    "key": "🛑",
                },
            },
            "event_id": "$reaction:example.com",
            "sender": "@user:example.com",
            "origin_server_ts": 1000000,
            "type": "m.reaction",
            "room_id": "!test:example.com",
        },
    )

    task = MagicMock()
    task.done = MagicMock(return_value=False)
    target = MessageTarget.resolve("!test:example.com", "$thread:example.com", "$message:example.com")
    bot.stop_manager.set_current(
        message_id="$message:example.com",
        target=target,
        task=task,
        run_id="run-123",
    )
    _record_pending_response(bot, "$message:example.com", target)

    with patch.object(bot.stop_manager, "_schedule_graceful_run_cancel") as mock_schedule_cancel:
        await dispatch_reaction_durably(bot, room, reaction_event)

    mock_schedule_cancel.assert_called_once_with("$message:example.com", "run-123")
    task.cancel.assert_called_once_with(msg=USER_STOP_CANCEL_MSG)
    send_response.assert_not_awaited()


@pytest.mark.asyncio
async def test_stop_manager_force_cancels_task_when_run_never_becomes_cancellable() -> None:
    """A stop request must hard-cancel quickly when the Agno run is not live yet."""
    stop_manager = StopManager(graceful_cancel_fallback_seconds=0.01)
    started = asyncio.Event()
    completed = asyncio.Event()
    task_cancelled = asyncio.Event()

    async def response_that_would_complete() -> None:
        try:
            started.set()
            await asyncio.sleep(0.1)
            completed.set()
        except asyncio.CancelledError:
            task_cancelled.set()
            raise

    task = asyncio.create_task(response_that_would_complete())
    await started.wait()

    stop_manager.set_current(
        message_id="$message:example.com",
        target=MessageTarget.resolve("!test:example.com", None, "$message:example.com"),
        task=task,
        run_id="run-123",
    )

    with patch("mindroom.stop.acancel_run", new=AsyncMock(return_value=False)):
        assert stop_manager.request_stop_if("$message:example.com", lambda: True) is True
        await asyncio.wait_for(task_cancelled.wait(), timeout=0.2)
        await _drain_stop_cleanup(stop_manager)

    with pytest.raises(asyncio.CancelledError):
        await task
    assert not completed.is_set()


@pytest.mark.asyncio
async def test_stop_manager_add_and_remove_button_notifies_cache_bookkeeping() -> None:
    """Stop-button add/remove should preserve cache bookkeeping for the synthetic reaction."""
    stop_manager = StopManager()
    client = AsyncMock(spec=nio.AsyncClient)
    client.user_id = "@agent:example.com"
    client.rooms = {
        "!test:example.com": nio.MatrixRoom("!test:example.com", "@agent:example.com"),
    }
    client.room_send = AsyncMock(
        return_value=nio.RoomSendResponse(room_id="!test:example.com", event_id="$reaction:example.com"),
    )
    client.room_redact = AsyncMock(return_value=MagicMock())
    notify_outbound_event = MagicMock()
    notify_outbound_redaction = MagicMock()
    task = MagicMock()
    task.done = MagicMock(return_value=False)
    stop_manager.set_current(
        message_id="$message:example.com",
        target=MessageTarget.resolve("!test:example.com", "$thread:example.com", "$message:example.com"),
        task=task,
    )

    added_event_id = await stop_manager.add_stop_button(
        client,
        "$message:example.com",
        notify_outbound_event=notify_outbound_event,
    )

    assert added_event_id == "$reaction:example.com"
    notify_outbound_event.assert_called_once_with(
        "!test:example.com",
        {
            "type": "m.reaction",
            "room_id": "!test:example.com",
            "event_id": "$reaction:example.com",
            "sender": "@agent:example.com",
            "content": {
                "m.relates_to": {
                    "rel_type": "m.annotation",
                    "event_id": "$message:example.com",
                    "key": "🛑",
                },
            },
        },
    )

    await stop_manager.remove_stop_button(
        client,
        "$message:example.com",
        notify_outbound_redaction=notify_outbound_redaction,
    )

    notify_outbound_redaction.assert_called_once_with("!test:example.com", "$reaction:example.com")


@pytest.mark.asyncio
async def test_stop_manager_force_cancels_task_when_graceful_cancel_errors() -> None:
    """Cancellation-manager failures must not disable the hard-cancel fallback."""
    stop_manager = StopManager(graceful_cancel_fallback_seconds=0.01)
    started = asyncio.Event()
    task_cancelled = asyncio.Event()

    async def hung_response() -> None:
        started.set()
        try:
            await asyncio.sleep(999)
        except asyncio.CancelledError:
            task_cancelled.set()
            raise

    task = asyncio.create_task(hung_response())
    await started.wait()

    stop_manager.set_current(
        message_id="$message:example.com",
        target=MessageTarget.resolve("!test:example.com", None, "$message:example.com"),
        task=task,
        run_id="run-123",
    )

    with patch("mindroom.stop.acancel_run", new=AsyncMock(side_effect=RuntimeError("redis down"))):
        assert stop_manager.request_stop_if("$message:example.com", lambda: True) is True
        await asyncio.wait_for(task_cancelled.wait(), timeout=0.2)
        await _drain_stop_cleanup(stop_manager)

    with pytest.raises(asyncio.CancelledError):
        await task


@pytest.mark.asyncio
async def test_stop_manager_logs_tracked_thread_context(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
    capsys: pytest.CaptureFixture[str],
) -> None:
    """Stop-manager logs should include tracked room/thread metadata."""
    monkeypatch.setenv("MINDROOM_LOG_FORMAT", "json")
    setup_logging(level="INFO", runtime_paths=test_runtime_paths(tmp_path))
    capsys.readouterr()

    stop_manager = StopManager()
    task = MagicMock()
    task.done.return_value = False
    target = MessageTarget.resolve("!room:example.org", "$thread:example.org", "$message:example.org")

    stop_manager.set_current(
        message_id="$message:example.org",
        target=target,
        task=task,
    )
    assert stop_manager.request_stop_if("$message:example.org", lambda: True) is True

    payloads = [json.loads(line) for line in capsys.readouterr().err.strip().splitlines()]
    tracking_payload = next(payload for payload in payloads if payload["event"] == "Tracking message generation")
    stop_payload = next(payload for payload in payloads if payload["event"] == "Handling stop reaction")

    assert tracking_payload["room_id"] == "!room:example.org"
    assert tracking_payload["thread_id"] == "$thread:example.org"
    assert stop_payload["room_id"] == "!room:example.org"
    assert stop_payload["thread_id"] == "$thread:example.org"


@pytest.mark.asyncio
async def test_stop_manager_immediately_cancels_task_even_when_acancel_run_succeeds() -> None:
    """A successful Agno cleanup request must not delay hard task cancellation."""
    stop_manager = StopManager(graceful_cancel_fallback_seconds=1.0)
    started = asyncio.Event()
    allow_task_to_finish = asyncio.Event()
    cleanup_requested = asyncio.Event()
    task_cancelled = asyncio.Event()

    async def hung_response() -> None:
        started.set()
        try:
            await asyncio.sleep(999)
        except asyncio.CancelledError:
            task_cancelled.set()
            await allow_task_to_finish.wait()
            raise

    task = asyncio.create_task(hung_response())
    await started.wait()

    stop_manager.set_current(
        message_id="$message:example.com",
        target=MessageTarget.resolve("!test:example.com", None, "$message:example.com"),
        task=task,
        run_id="run-123",
    )

    async def graceful_cancel_run(_run_id: str) -> bool:
        cleanup_requested.set()
        allow_task_to_finish.set()
        return True

    with patch("mindroom.stop.acancel_run", new=graceful_cancel_run):
        assert stop_manager.request_stop_if("$message:example.com", lambda: True) is True
        await asyncio.wait_for(task_cancelled.wait(), timeout=0.1)
        await asyncio.wait_for(cleanup_requested.wait(), timeout=0.2)
        await _drain_stop_cleanup(stop_manager)

    with pytest.raises(asyncio.CancelledError):
        await task


@pytest.mark.asyncio
async def test_stop_manager_immediately_cancels_task_when_acancel_run_is_slow() -> None:
    """A slow Agno cleanup call must not trigger a second task.cancel()."""
    stop_manager = StopManager(graceful_cancel_fallback_seconds=1.0)
    started = asyncio.Event()
    allow_task_to_finish = asyncio.Event()
    task_cancelled = asyncio.Event()
    cancellation_manager_started = asyncio.Event()

    async def hung_response() -> None:
        started.set()
        try:
            await asyncio.sleep(999)
        except asyncio.CancelledError:
            task_cancelled.set()
            await allow_task_to_finish.wait()
            raise

    async def hanging_cancel_run(_run_id: str) -> bool:
        cancellation_manager_started.set()
        await asyncio.sleep(999)
        return True

    task = asyncio.create_task(hung_response())
    await started.wait()

    stop_manager.set_current(
        message_id="$message:example.com",
        target=MessageTarget.resolve("!test:example.com", None, "$message:example.com"),
        task=task,
        run_id="run-123",
    )

    with patch("mindroom.stop.acancel_run", new=hanging_cancel_run):
        assert stop_manager.request_stop_if("$message:example.com", lambda: True) is True
        await asyncio.wait_for(task_cancelled.wait(), timeout=0.1)
        await asyncio.wait_for(cancellation_manager_started.wait(), timeout=0.2)
        await _drain_stop_cleanup(stop_manager)
        assert not task.done()
        allow_task_to_finish.set()

    with pytest.raises(asyncio.CancelledError):
        await task


@pytest.mark.asyncio
async def test_stop_manager_retries_until_run_becomes_cancellable() -> None:
    """Hard cancellation should happen immediately even if Agno needs a retry before cleanup succeeds."""
    stop_manager = StopManager(graceful_cancel_fallback_seconds=0.3)
    started = asyncio.Event()
    allow_task_to_finish = asyncio.Event()
    hard_cancelled = asyncio.Event()
    task_cancelled = asyncio.Event()
    cancel_attempts: list[str] = []

    async def graceful_response() -> None:
        try:
            started.set()
            await asyncio.sleep(999)
        except asyncio.CancelledError:
            hard_cancelled.set()
            await allow_task_to_finish.wait()
            task_cancelled.set()
            raise

    async def fake_acancel_run(run_id: str) -> bool:
        cancel_attempts.append(run_id)
        if len(cancel_attempts) == 1:
            return False
        allow_task_to_finish.set()
        return True

    task = asyncio.create_task(graceful_response())
    await started.wait()

    stop_manager.set_current(
        message_id="$message:example.com",
        target=MessageTarget.resolve("!test:example.com", None, "$message:example.com"),
        task=task,
        run_id="run-123",
    )

    with patch("mindroom.stop.acancel_run", new=fake_acancel_run):
        assert stop_manager.request_stop_if("$message:example.com", lambda: True) is True
        await asyncio.wait_for(hard_cancelled.wait(), timeout=0.1)
        await _drain_stop_cleanup(stop_manager)

    with pytest.raises(asyncio.CancelledError):
        await asyncio.wait_for(task, timeout=0.2)
    assert cancel_attempts == ["run-123", "run-123"]
    assert task_cancelled.is_set()


@pytest.mark.asyncio
async def test_stop_manager_reprobes_when_retry_updates_run_id() -> None:
    """A stop request should keep probing updated run IDs after the hard cancel is sent."""
    stop_manager = StopManager(graceful_cancel_fallback_seconds=0.3)
    started = asyncio.Event()
    allow_task_to_finish = asyncio.Event()
    hard_cancelled = asyncio.Event()
    first_cancel_attempt = asyncio.Event()
    second_cancel_attempt = asyncio.Event()
    cancel_attempts: list[str] = []

    async def graceful_response() -> None:
        try:
            started.set()
            await asyncio.sleep(999)
        except asyncio.CancelledError:
            hard_cancelled.set()
            await allow_task_to_finish.wait()
            raise

    async def fake_acancel_run(run_id: str) -> bool:
        cancel_attempts.append(run_id)
        if run_id == "run-123":
            first_cancel_attempt.set()
        if run_id == "run-456":
            second_cancel_attempt.set()
            allow_task_to_finish.set()
        return True

    task = asyncio.create_task(graceful_response())
    await started.wait()

    stop_manager.set_current(
        message_id="$message:example.com",
        target=MessageTarget.resolve("!test:example.com", None, "$message:example.com"),
        task=task,
        run_id="run-123",
    )

    with patch("mindroom.stop.acancel_run", new=fake_acancel_run):
        assert stop_manager.request_stop_if("$message:example.com", lambda: True) is True
        await asyncio.wait_for(hard_cancelled.wait(), timeout=0.1)
        await asyncio.wait_for(first_cancel_attempt.wait(), timeout=0.2)
        stop_manager.update_run_id("$message:example.com", "run-456")
        await asyncio.wait_for(second_cancel_attempt.wait(), timeout=0.2)
        await _drain_stop_cleanup(stop_manager)

    with pytest.raises(asyncio.CancelledError):
        await asyncio.wait_for(task, timeout=0.2)

    assert cancel_attempts == ["run-123", "run-456"]


@pytest.mark.asyncio
async def test_stop_manager_cleanup_uses_captured_run_id_after_task_finishes() -> None:
    """Cleanup should still cancel the Agno run even if the response task is already done."""
    stop_manager = StopManager(graceful_cancel_fallback_seconds=0.1)
    started = asyncio.Event()

    async def short_lived_response() -> None:
        started.set()
        await asyncio.sleep(999)

    task = asyncio.create_task(short_lived_response())
    await started.wait()

    stop_manager.set_current(
        message_id="$message:example.com",
        target=MessageTarget.resolve("!test:example.com", None, "$message:example.com"),
        task=task,
        run_id="run-123",
    )

    cancel_run = AsyncMock(return_value=True)
    with patch("mindroom.stop.acancel_run", new=cancel_run):
        assert stop_manager.request_stop_if("$message:example.com", lambda: True) is True
        with pytest.raises(asyncio.CancelledError):
            await task
        await _drain_stop_cleanup(stop_manager)

    cancel_run.assert_awaited_once_with("run-123")


@pytest.mark.asyncio
async def test_stop_emoji_from_agent_falls_through(tmp_path: Path) -> None:
    """Test that 🛑 reactions from agents fall through to other handlers."""
    config = bind_runtime_paths(
        Config(
            agents={
                "test_agent": {"display_name": "Test Agent", "rooms": ["!test:localhost"]},
                "helper": {"display_name": "Helper Agent", "rooms": ["!test:localhost"]},
            },
            authorization={"default_room_access": True},
        ),
        test_runtime_paths(tmp_path),
    )
    persist_entity_accounts(config, runtime_paths_for(config))
    ids = entity_ids(config, runtime_paths_for(config))

    agent_user = AgentMatrixUser(
        agent_name="test_agent",
        user_id=ids["test_agent"].full_id,
        display_name="Test Agent",
        password="test_password",  # noqa: S106
    )

    bot = AgentBot(
        agent_user=agent_user,
        storage_path=tmp_path,
        config=config,
        runtime_paths=runtime_paths_for(config),
        rooms=["!test:localhost"],
    )

    # Set up the bot
    bot.client = AsyncMock(spec=nio.AsyncClient)
    bot.client.user_id = agent_user.user_id
    bot.logger = MagicMock()

    room = nio.MatrixRoom(room_id="!test:localhost", own_user_id=agent_user.user_id)

    # Create a 🛑 reaction from ANOTHER AGENT
    reaction_event = nio.ReactionEvent.from_dict(
        {
            "content": {
                "m.relates_to": {
                    "rel_type": "m.annotation",
                    "event_id": "$message:example.com",
                    "key": "🛑",
                },
            },
            "event_id": "$reaction:example.com",
            "sender": ids["helper"].full_id,
            "origin_server_ts": 1000000,
            "type": "m.reaction",
            "room_id": "!test:localhost",
        },
    )

    with patch("mindroom.bot.interactive.handle_reaction") as mock_handle_reaction:
        mock_handle_reaction.return_value = None  # No interactive result

        # Track a message as being generated
        task = MagicMock()  # Use MagicMock instead of AsyncMock for the task
        task.done = MagicMock(return_value=False)  # done() is a regular method, not async
        bot.stop_manager.set_current(
            message_id="$message:example.com",
            target=MessageTarget.resolve("!test:example.com", None, "$message:example.com"),
            task=task,
        )

        # Process the reaction from an agent
        await dispatch_reaction_durably(bot, room, reaction_event)

        # Should have called interactive.handle_reaction (fell through)
        mock_handle_reaction.assert_called_once()

        # Task should NOT have been cancelled (agents can't stop generation)
        task.cancel.assert_not_called()


@pytest.mark.asyncio
async def test_stop_reaction_blocked_by_reply_permissions(tmp_path: Path) -> None:
    """Disallowed senders must not trigger stop or send confirmation via 🛑 reaction."""
    config = bind_runtime_paths(
        Config(
            agents={
                "test_agent": {
                    "display_name": "Test Agent",
                    "rooms": ["!test:example.com"],
                },
            },
            authorization={
                "default_room_access": True,
                "agent_reply_permissions": {"test_agent": ["@alice:example.com"]},
            },
        ),
        orchestrator_runtime_paths(tmp_path, config_path=tmp_path / "config.yaml"),
    )
    persist_entity_accounts(config, runtime_paths_for(config))
    agent_user = _stop_test_agent_user(config)

    bot = AgentBot(
        agent_user=agent_user,
        storage_path=tmp_path,
        config=config,
        runtime_paths=runtime_paths_for(config),
        rooms=["!test:example.com"],
    )
    bot.client = AsyncMock(spec=nio.AsyncClient)
    bot.client.user_id = agent_user.user_id
    bot.logger = MagicMock()

    room = nio.MatrixRoom(room_id="!test:example.com", own_user_id=agent_user.user_id)

    # Track a message as being generated
    task = MagicMock()
    task.done = MagicMock(return_value=False)
    bot.stop_manager.set_current(
        message_id="$message:example.com",
        target=MessageTarget.resolve("!test:example.com", None, "$message:example.com"),
        task=task,
    )

    # Disallowed sender reacts with stop emoji
    reaction_event = nio.ReactionEvent.from_dict(
        {
            "content": {
                "m.relates_to": {
                    "rel_type": "m.annotation",
                    "event_id": "$message:example.com",
                    "key": "🛑",
                },
            },
            "event_id": "$reaction_bob:example.com",
            "sender": "@bob:example.com",
            "origin_server_ts": 1000000,
            "type": "m.reaction",
            "room_id": "!test:example.com",
        },
    )

    send_response = AsyncMock()
    install_send_response_mock(bot, send_response)

    with patch("mindroom.bot.is_authorized_sender", return_value=True):
        await dispatch_reaction_durably(bot, room, reaction_event)

    # Task should NOT have been cancelled — sender is disallowed
    task.cancel.assert_not_called()
    # No confirmation message should have been sent
    send_response.assert_not_called()
