tests/messaging/test_handler.py

import asyncio
from unittest.mock import AsyncMock, MagicMock, call, patch

import pytest

from free_claude_code.messaging.command_context import ReplyClearResult, StopOutcome
from free_claude_code.messaging.models import MessageScope
from free_claude_code.messaging.platforms.ports import MessagingStartupNotice
from free_claude_code.messaging.session import SessionStore
from free_claude_code.messaging.trees import (
    CancellationReason,
    CancellationResult,
    CancellationUiOwner,
    FailureResult,
    MessageReferenceKind,
    MessageState,
    MessageSubtreeRemovalResult,
    NodeClaim,
    NodeUiTarget,
    QueueEntry,
    ReplyTarget,
    TreeIdentity,
    TreeQueueManager,
    TreeSnapshot,
)
from free_claude_code.messaging.trees.transitions import CancellationEffect
from free_claude_code.messaging.voice import VoiceCancellationResult
from free_claude_code.messaging.workflow import MessagingWorkflow

_SCOPE = MessageScope(platform="telegram", chat_id="chat_1")
_OTHER_SCOPE = MessageScope(platform="telegram", chat_id="other_chat")


def _stop_outcome(
    cancelled_count: int,
    *,
    scopes: frozenset[MessageScope] = frozenset({_SCOPE}),
    fallback_required: bool = False,
) -> StopOutcome:
    return StopOutcome(cancelled_count, scopes, fallback_required)


def _startup_notice(chat_id: str = _SCOPE.chat_id) -> MessagingStartupNotice:
    return MessagingStartupNotice(chat_id=chat_id, transport_label="Bot API")


async def _event_stream(events):
    for event in events:
        await asyncio.sleep(0)
        yield event


def _claim(
    node_id: str = "node_1",
    *,
    prompt: str = "hello",
    parent_session_id: str | None = None,
) -> NodeClaim:
    return NodeClaim(
        identity=TreeIdentity(scope=_SCOPE, root_id="root_1"),
        claim_id="claim_1",
        node=NodeUiTarget(
            scope=_SCOPE,
            node_id=node_id,
            status_message_id="status_1",
        ),
        prompt=prompt,
        parent_session_id=parent_session_id,
    )


def _snapshot(root_id: str = "root_1") -> TreeSnapshot:
    return TreeSnapshot(scope=_SCOPE, root_id=root_id, nodes={})


def _session(events) -> MagicMock:
    session = MagicMock()
    session.start_task.return_value = _event_stream(events)
    return session


async def _wait_for_idle(workflow: MessagingWorkflow) -> None:
    for _ in range(200):
        if workflow.tree_queue.task_count() == 0:
            await asyncio.sleep(0)
            return
        await asyncio.sleep(0.01)
    raise AssertionError("messaging workflow did not become idle")


@pytest.fixture
def handler(mock_platform, mock_cli_manager, mock_session_store):
    default_session = _session([{"type": "exit", "code": 0}])
    mock_cli_manager.get_or_create_session.return_value = (
        default_session,
        "session_1",
        False,
    )
    return MessagingWorkflow(
        mock_platform,
        mock_cli_manager,
        mock_session_store,
        platform_name="telegram",
        voice_cancellation=mock_platform,
    )


@pytest.mark.asyncio
async def test_handle_message_turn_trace_always_includes_full_message_text(
    mock_platform,
    mock_cli_manager,
    mock_session_store,
    incoming_message_factory,
):
    text = "user-message-content-visible-in-trace"
    workflow = MessagingWorkflow(
        mock_platform,
        mock_cli_manager,
        mock_session_store,
    )
    incoming = incoming_message_factory(text=text)
    with (
        patch.object(workflow.turn_intake, "handle_message", new_callable=AsyncMock),
        patch("free_claude_code.messaging.workflow.trace_event") as trace_mock,
    ):
        await workflow.handle_message(incoming)

    assert trace_mock.call_args.kwargs["event"] == "turn.received"
    assert trace_mock.call_args.kwargs["message_text"] == text


@pytest.mark.asyncio
async def test_user_prompt_and_status_are_recorded_for_global_clear(
    handler,
    mock_session_store,
    incoming_message_factory,
) -> None:
    incoming = incoming_message_factory(text="keep me", message_id="user-prompt")

    await handler.handle_message(incoming)
    await _wait_for_idle(handler)

    mock_session_store.record_message_id.assert_has_calls(
        [
            call(
                incoming.platform,
                incoming.chat_id,
                "user-prompt",
                "in",
                "prompt",
            ),
            call(
                incoming.platform,
                incoming.chat_id,
                "msg_123",
                "out",
                "status",
            ),
        ]
    )


@pytest.mark.parametrize(
    ("target", "expected"),
    [
        (None, "Launching"),
        (
            ReplyTarget(
                node_id="parent",
                reference_id="parent",
                reference_kind=MessageReferenceKind.PROMPT,
                queue_position=None,
            ),
            "Continuing",
        ),
        (
            ReplyTarget(
                node_id="parent",
                reference_id="status-parent",
                reference_kind=MessageReferenceKind.STATUS,
                queue_position=3,
            ),
            "position 3",
        ),
    ],
)
def test_initial_status_uses_immutable_reply_advice(handler, target, expected):
    assert expected in handler.turn_intake._get_initial_status(target)


@pytest.mark.asyncio
async def test_global_stop_success_uses_existing_status_without_confirmation(
    handler, mock_platform, incoming_message_factory
):
    incoming = incoming_message_factory(text="/stop")
    handler.stop_all_tasks = AsyncMock(return_value=_stop_outcome(5))

    await handler.handle_message(incoming)

    handler.stop_all_tasks.assert_awaited_once()
    mock_platform.queue_send_message.assert_not_awaited()


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "status_feedback_scopes",
    (
        frozenset({_OTHER_SCOPE}),
        frozenset({_SCOPE, _OTHER_SCOPE}),
    ),
    ids=("remote-only", "mixed-local-and-remote"),
)
async def test_global_stop_reports_cross_chat_work(
    handler,
    mock_platform,
    incoming_message_factory,
    status_feedback_scopes: frozenset[MessageScope],
) -> None:
    incoming = incoming_message_factory(text="/stop")
    handler.stop_all_tasks = AsyncMock(
        return_value=_stop_outcome(2, scopes=status_feedback_scopes)
    )

    await handler.handle_message(incoming)

    assert (
        "Cancelled 2 pending or active requests"
        in mock_platform.queue_send_message.call_args.args[1]
    )


@pytest.mark.asyncio
async def test_global_stop_noop_reports_nothing_to_stop(
    handler, mock_platform, incoming_message_factory
) -> None:
    incoming = incoming_message_factory(text="/stop")
    handler.stop_all_tasks = AsyncMock(
        return_value=_stop_outcome(0, scopes=frozenset())
    )

    await handler.handle_message(incoming)

    mock_platform.queue_send_message.assert_awaited_once_with(
        incoming.chat_id,
        "⏹ *Stopped\\.* Nothing to stop\\.",
        fire_and_forget=False,
        message_thread_id=None,
    )


@pytest.mark.asyncio
async def test_global_stop_without_status_target_sends_fallback_confirmation(
    handler, mock_platform, incoming_message_factory
) -> None:
    incoming = incoming_message_factory(text="/stop")
    handler.stop_all_tasks = AsyncMock(
        return_value=_stop_outcome(
            1,
            scopes=frozenset(),
            fallback_required=True,
        )
    )

    await handler.handle_message(incoming)

    assert (
        "Cancelled 1 pending or active request"
        in mock_platform.queue_send_message.call_args.args[1]
    )


@pytest.mark.asyncio
async def test_failed_global_stop_never_reports_success(
    handler, mock_platform, incoming_message_factory
) -> None:
    incoming = incoming_message_factory(text="/stop")
    handler.stop_all_tasks = AsyncMock(
        side_effect=RuntimeError("Failed to stop 1 managed Claude session.")
    )

    with pytest.raises(RuntimeError, match="Failed to stop"):
        await handler.handle_message(incoming)

    mock_platform.queue_send_message.assert_not_awaited()


@pytest.mark.asyncio
async def test_reply_stop_resolves_and_stops_only_target(
    handler, mock_platform, mock_cli_manager, incoming_message_factory
):
    handler.stop_reply = AsyncMock(return_value=_stop_outcome(1))
    handler.stop_all_tasks = AsyncMock(return_value=_stop_outcome(999))
    incoming = incoming_message_factory(
        text="/stop",
        message_id="stop_msg",
        reply_to_message_id="status_root",
    )

    await handler.handle_message(incoming)

    handler.stop_reply.assert_awaited_once_with(incoming.scope, "status_root")
    handler.stop_all_tasks.assert_not_awaited()
    mock_cli_manager.stop_all.assert_not_awaited()
    mock_platform.queue_send_message.assert_not_awaited()


@pytest.mark.asyncio
async def test_reply_stop_unknown_does_not_stop_all(
    handler, mock_platform, mock_cli_manager, incoming_message_factory
):
    handler.stop_reply = AsyncMock(return_value=_stop_outcome(0, scopes=frozenset()))
    handler.stop_all_tasks = AsyncMock(return_value=_stop_outcome(5))
    incoming = incoming_message_factory(
        text="/stop",
        message_id="stop_msg",
        reply_to_message_id="unknown_msg",
    )

    await handler.handle_message(incoming)

    handler.stop_reply.assert_awaited_once_with(incoming.scope, "unknown_msg")
    handler.stop_all_tasks.assert_not_awaited()
    mock_cli_manager.stop_all.assert_not_awaited()
    assert (
        "Nothing to stop for that message"
        in mock_platform.queue_send_message.call_args.args[1]
    )


@pytest.mark.asyncio
async def test_reply_stop_cancels_voice_when_no_tree_was_admitted(
    handler,
    mock_platform,
    incoming_message_factory,
) -> None:
    mock_platform.cancel_pending_voice.return_value = VoiceCancellationResult(
        scope=_SCOPE,
        voice_message_id="voice",
        status_message_id="voice_status",
        delete_message_ids=frozenset({"voice", "voice_status"}),
    )
    incoming = incoming_message_factory(
        text="/stop",
        message_id="stop_msg",
        reply_to_message_id="voice",
    )

    await handler.handle_message(incoming)
    await asyncio.sleep(0)

    mock_platform.queue_send_message.assert_not_awaited()
    assert mock_platform.queue_edit_message.call_args.args[1] == "voice_status"
    assert "Stopped" in mock_platform.queue_edit_message.call_args.args[2]


@pytest.mark.asyncio
async def test_reply_stop_without_voice_status_sends_fallback_confirmation(
    handler,
    mock_platform,
    incoming_message_factory,
) -> None:
    mock_platform.cancel_pending_voice.return_value = VoiceCancellationResult(
        scope=_SCOPE,
        voice_message_id="voice",
        status_message_id=None,
        delete_message_ids=frozenset({"voice"}),
    )
    incoming = incoming_message_factory(
        text="/stop",
        message_id="stop_msg",
        reply_to_message_id="voice",
    )

    await handler.handle_message(incoming)

    assert "Cancelled 1 request" in mock_platform.queue_send_message.call_args.args[1]
    mock_platform.queue_edit_message.assert_not_awaited()


@pytest.mark.asyncio
async def test_reply_stop_resolves_tree_after_joining_voice_handoff(
    handler,
    mock_platform,
    incoming_message_factory,
) -> None:
    events: list[str] = []

    async def cancel_voice(*_args) -> VoiceCancellationResult:
        events.append("voice")
        return VoiceCancellationResult(
            scope=_SCOPE,
            voice_message_id="voice",
            status_message_id="voice_status",
            delete_message_ids=frozenset({"voice", "voice_status"}),
        )

    async def resolve_tree(*_args) -> str:
        events.append("tree")
        return "voice"

    mock_platform.cancel_pending_voice.side_effect = cancel_voice
    cancellation = CancellationResult(
        effects=(
            CancellationEffect(
                node=NodeUiTarget(
                    scope=_SCOPE,
                    node_id="voice",
                    status_message_id="voice_status",
                ),
                ui_owner=CancellationUiOwner.WORKFLOW,
            ),
        )
    )
    incoming = incoming_message_factory(
        text="/stop",
        message_id="stop_msg",
        reply_to_message_id="voice",
    )

    with (
        patch.object(
            handler.tree_queue,
            "resolve_node_id",
            side_effect=resolve_tree,
        ),
        patch.object(
            handler.tree_queue,
            "cancel_node",
            new_callable=AsyncMock,
            return_value=cancellation,
        ) as cancel_node,
    ):
        await handler.handle_message(incoming)

    cancel_node.assert_awaited_once_with(
        incoming.scope,
        "voice",
        reason=CancellationReason.STOP,
    )
    assert events == ["voice", "tree"]
    mock_platform.queue_send_message.assert_not_awaited()


@pytest.mark.asyncio
async def test_reply_stop_port_failure_prevents_tree_transition(
    handler,
    mock_platform,
) -> None:
    mock_platform.cancel_pending_voice.side_effect = RuntimeError("voice port failed")
    with (
        patch.object(
            handler.tree_queue,
            "resolve_node_id",
            new_callable=AsyncMock,
        ) as resolve,
        pytest.raises(RuntimeError, match="voice port failed"),
    ):
        await handler.stop_reply(_SCOPE, "voice")

    resolve.assert_not_awaited()
    mock_platform.queue_edit_message.assert_not_awaited()


@pytest.mark.asyncio
async def test_reply_stop_renders_voice_before_resolve_failure(
    handler,
    mock_platform,
) -> None:
    mock_platform.cancel_pending_voice.return_value = VoiceCancellationResult(
        scope=_SCOPE,
        voice_message_id="voice",
        status_message_id="voice_status",
        delete_message_ids=frozenset({"voice", "voice_status"}),
    )
    with (
        patch.object(
            handler.tree_queue,
            "resolve_node_id",
            AsyncMock(side_effect=RuntimeError("resolve failed")),
        ),
        pytest.raises(RuntimeError, match="resolve failed"),
    ):
        await handler.stop_reply(_SCOPE, "voice")
    await asyncio.sleep(0)

    mock_platform.queue_edit_message.assert_awaited_once()
    assert mock_platform.queue_edit_message.call_args.args[1] == "voice_status"


@pytest.mark.asyncio
async def test_cancelled_reply_stop_finishes_after_voice_join_and_lock_wait(
    handler,
    mock_platform,
) -> None:
    voice_returned = asyncio.Event()

    async def cancel_voice(*_args) -> VoiceCancellationResult:
        voice_returned.set()
        return VoiceCancellationResult(
            scope=_SCOPE,
            voice_message_id="voice",
            status_message_id="voice_status",
            delete_message_ids=frozenset({"voice", "voice_status"}),
        )

    mock_platform.cancel_pending_voice.side_effect = cancel_voice
    resolve = AsyncMock(return_value=None)
    await handler._state_lock.acquire()
    try:
        with patch.object(handler.tree_queue, "resolve_node_id", resolve):
            stop_task = asyncio.create_task(handler.stop_reply(_SCOPE, "voice"))
            await voice_returned.wait()
            await asyncio.sleep(0)
            stop_task.cancel()
            stop_task.cancel()
            await asyncio.sleep(0)

            assert not stop_task.done()
            handler._state_lock.release()
            with pytest.raises(asyncio.CancelledError):
                await stop_task
    finally:
        if handler._state_lock.locked():
            handler._state_lock.release()
    await asyncio.sleep(0)

    assert stop_task.cancelling() == 2
    resolve.assert_awaited_once_with(_SCOPE, "voice")
    mock_platform.queue_edit_message.assert_awaited_once()
    assert mock_platform.queue_edit_message.call_args.args[1] == "voice_status"


@pytest.mark.asyncio
async def test_cancelled_reply_stop_preserves_later_operation_failure(
    handler,
    mock_platform,
) -> None:
    voice_returned = asyncio.Event()
    failure = RuntimeError("resolve failed after cancellation")

    async def cancel_voice(*_args) -> None:
        voice_returned.set()

    mock_platform.cancel_pending_voice.side_effect = cancel_voice
    resolve = AsyncMock(side_effect=failure)
    await handler._state_lock.acquire()
    try:
        with patch.object(handler.tree_queue, "resolve_node_id", resolve):
            stop_task = asyncio.create_task(handler.stop_reply(_SCOPE, "voice"))
            await voice_returned.wait()
            await asyncio.sleep(0)
            stop_task.cancel()
            await asyncio.sleep(0)

            assert not stop_task.done()
            handler._state_lock.release()
            with pytest.raises(RuntimeError) as raised:
                await stop_task
    finally:
        if handler._state_lock.locked():
            handler._state_lock.release()

    assert raised.value is failure
    assert stop_task.cancelling() == 1
    resolve.assert_awaited_once_with(_SCOPE, "voice")


@pytest.mark.asyncio
async def test_stats_command_reports_cli_and_tree_counts(
    handler, mock_platform, mock_cli_manager, incoming_message_factory
):
    mock_cli_manager.get_stats.return_value = {"active_sessions": 2}

    await handler.handle_message(incoming_message_factory(text="/stats"))

    text = mock_platform.queue_send_message.call_args.args[1]
    assert "Active CLI: 2" in text
    assert "Message Trees: 0" in text
    assert mock_platform.queue_send_message.call_args.kwargs["fire_and_forget"] is False


@pytest.mark.asyncio
async def test_status_echo_is_filtered(
    handler, mock_platform, incoming_message_factory
):
    await handler.handle_message(incoming_message_factory(text="⏳ Thinking..."))

    mock_platform.queue_send_message.assert_not_awaited()


@pytest.mark.asyncio
async def test_new_turn_uses_public_admission_and_persists_exact_snapshot(
    handler,
    mock_platform,
    mock_session_store,
    incoming_message_factory,
):
    incoming = incoming_message_factory(text="hello", message_id="node_1")
    mock_platform.queue_send_message.return_value = "status_123"

    await handler.handle_message(incoming)
    await _wait_for_idle(handler)

    assert "Launching" in mock_platform.queue_send_message.call_args.args[1]
    assert mock_session_store.save_tree_snapshot.call_count >= 2
    view = await handler.tree_queue.get_node(incoming.scope, "node_1")
    assert view is not None
    assert view.state is MessageState.COMPLETED


@pytest.mark.asyncio
async def test_duplicate_delivery_removes_its_provisional_status(
    handler,
    mock_platform,
    mock_session_store,
    incoming_message_factory,
):
    incoming = incoming_message_factory(text="hello", message_id="duplicate")
    mock_platform.queue_send_message.side_effect = ["status-first", "status-rejected"]

    await handler.handle_message(incoming)
    await _wait_for_idle(handler)
    await handler.handle_message(incoming)

    mock_platform.queue_delete_messages.assert_awaited_once_with(
        incoming.chat_id,
        ["status-rejected"],
        fire_and_forget=False,
    )
    mock_session_store.forget_tracked_message_ids.assert_called_once_with(
        incoming.platform,
        incoming.chat_id,
        {"status-rejected"},
    )


@pytest.mark.asyncio
async def test_pre_sent_status_is_edited_in_place(
    handler, mock_platform, incoming_message_factory
):
    incoming = incoming_message_factory(
        text="hello",
        message_id="node_1",
        status_message_id="existing_status",
    )

    await handler.handle_message(incoming)
    await _wait_for_idle(handler)

    first_edit = mock_platform.queue_edit_message.call_args_list[0]
    assert first_edit.args[1] == "existing_status"
    assert "Launching" in first_edit.args[2]
    assert first_edit.kwargs["fire_and_forget"] is False
    mock_platform.queue_send_message.assert_not_awaited()


@pytest.mark.asyncio
async def test_busy_reply_is_rendered_with_atomic_queue_position(
    handler, mock_platform, mock_cli_manager, incoming_message_factory
):
    started = asyncio.Event()

    async def blocking_start(*args, **kwargs):
        started.set()
        await asyncio.sleep(60)
        if False:
            yield {}

    session = MagicMock()
    session.start_task = blocking_start
    mock_cli_manager.get_or_create_session.return_value = (
        session,
        "session_1",
        False,
    )
    mock_platform.queue_send_message.side_effect = ["status_root", "status_child"]

    root = incoming_message_factory(text="root", message_id="root")
    await handler.handle_message(root)
    await started.wait()
    child = incoming_message_factory(
        text="child",
        message_id="child",
        reply_to_message_id="status_root",
    )
    await handler.handle_message(child)

    assert "position 1" in mock_platform.queue_send_message.call_args.args[1]
    queued_edit = mock_platform.queue_edit_message.call_args_list[-1]
    assert queued_edit.args[1] == "status_child"
    assert "position 1" in queued_edit.args[2]

    await handler.stop_all_tasks()


@pytest.mark.asyncio
async def test_queue_position_callback_consumes_immutable_entries(
    handler, mock_platform
):
    queue = (
        QueueEntry(
            node=NodeUiTarget(
                scope=_SCOPE,
                node_id="child_1",
                status_message_id="status_1",
            ),
            position=1,
        ),
        QueueEntry(
            node=NodeUiTarget(
                scope=_SCOPE,
                node_id="child_2",
                status_message_id="status_2",
            ),
            position=2,
        ),
    )

    await handler.turn_intake.update_queue_positions(queue)
    await asyncio.sleep(0)

    calls = mock_platform.queue_edit_message.call_args_list
    assert [call.args[1] for call in calls] == ["status_1", "status_2"]
    assert "position 1" in calls[0].args[2]
    assert "position 2" in calls[1].args[2]


@pytest.mark.asyncio
async def test_claim_started_callback_renders_processing(handler, mock_platform):
    claim = _claim()

    await handler.turn_intake.mark_node_processing(claim)
    await asyncio.sleep(0)

    args, kwargs = mock_platform.queue_edit_message.call_args
    assert args[0:2] == ("chat_1", "status_1")
    assert "Processing" in args[2]
    assert kwargs["parse_mode"] == "MarkdownV2"


@pytest.mark.asyncio
async def test_stop_all_applies_immutable_ui_ownership_and_snapshots(
    handler, mock_cli_manager, mock_platform, mock_session_store
):
    workflow_owned = CancellationEffect(
        node=NodeUiTarget(
            scope=_SCOPE,
            node_id="queued",
            status_message_id="status_queued",
        ),
        ui_owner=CancellationUiOwner.WORKFLOW,
    )
    runner_owned = CancellationEffect(
        node=NodeUiTarget(
            scope=_SCOPE,
            node_id="active",
            status_message_id="status_active",
        ),
        ui_owner=CancellationUiOwner.RUNNER,
    )
    snapshot = _snapshot()
    result = CancellationResult(
        effects=(workflow_owned, runner_owned),
        snapshots=(snapshot,),
    )
    with patch.object(
        handler.tree_queue,
        "cancel_all",
        AsyncMock(return_value=result),
    ) as cancel_all:
        outcome = await handler.stop_all_tasks()
    await asyncio.sleep(0)

    assert outcome == _stop_outcome(2)
    cancel_all.assert_awaited_once_with(reason=CancellationReason.STOP)
    mock_cli_manager.stop_all.assert_awaited_once()
    assert mock_platform.fire_and_forget.call_count == 1
    assert mock_platform.queue_edit_message.call_args.args[1] == "status_queued"
    mock_session_store.save_tree_snapshot.assert_called_once_with(snapshot)


@pytest.mark.asyncio
async def test_stop_all_joins_voices_before_tree_transaction_and_deduplicates(
    handler,
    mock_cli_manager,
    mock_platform,
) -> None:
    events: list[str] = []
    voice = VoiceCancellationResult(
        scope=_SCOPE,
        voice_message_id="voice",
        status_message_id=None,
        delete_message_ids=frozenset({"voice"}),
    )
    tree_result = CancellationResult(
        effects=(
            CancellationEffect(
                node=NodeUiTarget(
                    scope=_SCOPE,
                    node_id="voice",
                    status_message_id="voice_status",
                ),
                ui_owner=CancellationUiOwner.WORKFLOW,
            ),
        )
    )

    async def cancel_voices() -> tuple[VoiceCancellationResult, ...]:
        assert not handler._state_lock.locked()
        events.append("voices")
        return (voice,)

    async def cancel_trees(*, reason: CancellationReason) -> CancellationResult:
        assert reason is CancellationReason.STOP
        assert handler._state_lock.locked()
        events.append("trees")
        return tree_result

    mock_platform.cancel_all_pending_voices.side_effect = cancel_voices
    with patch.object(handler.tree_queue, "cancel_all", side_effect=cancel_trees):
        outcome = await handler.stop_all_tasks()
    await asyncio.sleep(0)

    assert outcome == _stop_outcome(1)
    assert events == ["voices", "trees"]
    mock_cli_manager.stop_all.assert_awaited_once()
    mock_platform.queue_edit_message.assert_awaited_once()
    assert mock_platform.queue_edit_message.call_args.args[1] == "voice_status"


@pytest.mark.asyncio
async def test_stop_all_voice_join_failure_prevents_false_transition(
    handler,
    mock_cli_manager,
    mock_platform,
) -> None:
    mock_platform.cancel_all_pending_voices.side_effect = RuntimeError(
        "voice handoff cleanup failed"
    )

    with (
        patch.object(
            handler.tree_queue, "cancel_all", new_callable=AsyncMock
        ) as cancel,
        pytest.raises(RuntimeError, match="voice handoff cleanup failed"),
    ):
        await handler.stop_all_tasks()

    cancel.assert_not_awaited()
    mock_cli_manager.stop_all.assert_not_awaited()


@pytest.mark.asyncio
async def test_stop_all_persists_committed_transition_before_cli_shutdown(
    handler,
    mock_cli_manager,
    mock_session_store,
):
    shutdown_started = asyncio.Event()
    release_shutdown = asyncio.Event()
    snapshot = _snapshot()
    result = CancellationResult(snapshots=(snapshot,))

    async def block_shutdown() -> None:
        shutdown_started.set()
        await release_shutdown.wait()

    mock_cli_manager.stop_all.side_effect = block_shutdown
    with patch.object(
        handler.tree_queue,
        "cancel_all",
        AsyncMock(return_value=result),
    ):
        stop_task = asyncio.create_task(handler.stop_all_tasks())
        await shutdown_started.wait()

        mock_session_store.save_tree_snapshot.assert_called_once_with(snapshot)
        stop_task.cancel()
        stop_task.cancel()
        await asyncio.sleep(0)
        assert not stop_task.done()
        release_shutdown.set()
        with pytest.raises(asyncio.CancelledError):
            await stop_task
        assert stop_task.cancelling() == 2


@pytest.mark.asyncio
async def test_terminal_close_waits_past_interactive_drain_timeout(
    monkeypatch,
    mock_platform,
    mock_cli_manager,
    mock_session_store,
    incoming_message_factory,
) -> None:
    monkeypatch.setattr(
        "free_claude_code.messaging.trees.manager.CANCEL_TASK_DRAIN_TIMEOUT_S",
        0.01,
    )
    runner_started = asyncio.Event()
    cancellation_seen = asyncio.Event()
    release_cleanup = asyncio.Event()
    cli_stop_seen = asyncio.Event()

    async def cancellation_delayed_runner(_claim: NodeClaim) -> None:
        runner_started.set()
        try:
            await asyncio.Event().wait()
        except asyncio.CancelledError:
            cancellation_seen.set()
            await release_cleanup.wait()
            raise

    async def stop_cli() -> None:
        cli_stop_seen.set()

    mock_platform.queue_send_message.return_value = "status_1"
    mock_cli_manager.stop_all.side_effect = stop_cli
    workflow = MessagingWorkflow(
        mock_platform,
        mock_cli_manager,
        mock_session_store,
        platform_name="telegram",
    )
    workflow._tree_queue = TreeQueueManager(cancellation_delayed_runner)
    await workflow.handle_message(
        incoming_message_factory(text="work", message_id="work_1")
    )
    await runner_started.wait()

    close_task = asyncio.create_task(workflow.close())
    try:
        await asyncio.wait_for(cancellation_seen.wait(), timeout=1)
        await asyncio.wait_for(cli_stop_seen.wait(), timeout=1)

        mock_cli_manager.stop_all.assert_awaited_once()
        assert close_task.done() is False
        assert workflow.tree_queue.task_count() == 1
        mock_session_store.flush_pending_save.assert_not_called()

        release_cleanup.set()
        await asyncio.wait_for(close_task, timeout=1)

        assert workflow.tree_queue.task_count() == 0
        mock_session_store.flush_pending_save.assert_called_once()
    finally:
        release_cleanup.set()
        if not close_task.done():
            await asyncio.wait_for(close_task, timeout=1)


@pytest.mark.asyncio
async def test_node_runner_success_uses_claim_and_semantic_completion(
    handler, mock_cli_manager, mock_platform, mock_session_store
):
    claim = _claim(prompt="say hello")
    session = _session(
        [
            {
                "type": "assistant",
                "message": {
                    "content": [
                        {"type": "thinking", "thinking": "Let me think"},
                        {"type": "text", "text": "Hello world"},
                    ]
                },
            },
            {"type": "exit", "code": 0},
        ]
    )
    mock_cli_manager.get_or_create_session.return_value = (
        session,
        "session_1",
        False,
    )
    snapshot = _snapshot()
    with patch.object(
        handler.tree_queue,
        "complete_claim",
        AsyncMock(return_value=snapshot),
    ) as complete_claim:
        await handler.node_runner.process_node(claim)

    complete_claim.assert_awaited_once_with(claim, "session_1")
    mock_session_store.save_tree_snapshot.assert_called_once_with(snapshot)
    rendered = mock_platform.queue_edit_message.call_args_list[-1].args[2]
    assert "✅ *Complete*" in rendered
    assert "Hello world" in rendered
    mock_cli_manager.get_or_create_session.assert_awaited_once_with(session_id=None)
    assert session.start_task.call_args.args == ("say hello",)
    assert session.start_task.call_args.kwargs == {
        "session_id": None,
        "fork_session": False,
    }


@pytest.mark.asyncio
async def test_node_runner_uses_claim_parent_session_for_fork(
    handler, mock_cli_manager
):
    claim = _claim(parent_session_id="parent_session")
    session = _session([{"type": "exit", "code": 0}])
    mock_cli_manager.get_or_create_session.return_value = (
        session,
        "child_session",
        False,
    )

    await handler.node_runner.process_node(claim)

    mock_cli_manager.get_or_create_session.assert_awaited_once_with(
        session_id="parent_session"
    )
    assert session.start_task.call_args.kwargs == {
        "session_id": "parent_session",
        "fork_session": True,
    }


@pytest.mark.asyncio
async def test_session_info_records_real_session_through_manager(
    handler, mock_cli_manager, mock_session_store
):
    claim = _claim()
    session = _session(
        [
            {"type": "session_info", "session_id": "real_session"},
            {"type": "exit", "code": 0},
        ]
    )
    mock_cli_manager.get_or_create_session.return_value = (
        session,
        "temporary_session",
        True,
    )
    record_snapshot = _snapshot("record")
    complete_snapshot = _snapshot("complete")
    with (
        patch.object(
            handler.tree_queue,
            "record_session",
            AsyncMock(return_value=record_snapshot),
        ) as record_session,
        patch.object(
            handler.tree_queue,
            "complete_claim",
            AsyncMock(return_value=complete_snapshot),
        ) as complete_claim,
    ):
        await handler.node_runner.process_node(claim)

    mock_cli_manager.register_real_session_id.assert_awaited_once_with(
        "temporary_session", "real_session"
    )
    record_session.assert_awaited_once_with(claim, "real_session")
    complete_claim.assert_awaited_once_with(claim, "real_session")
    assert mock_session_store.save_tree_snapshot.call_args_list == [
        ((record_snapshot,), {}),
        ((complete_snapshot,), {}),
    ]


@pytest.mark.asyncio
async def test_session_info_rejects_unregistered_real_session_without_recording_it(
    handler,
    mock_cli_manager,
) -> None:
    from free_claude_code.messaging.node_event_pipeline import (
        handle_session_info_event,
    )

    claim = _claim()
    record_session = AsyncMock()
    mock_cli_manager.register_real_session_id.return_value = False

    with pytest.raises(
        RuntimeError,
        match=r"^Managed Claude session registration failed\.$",
    ) as raised:
        await handle_session_info_event(
            {"type": "session_info", "session_id": "real_session"},
            claim,
            None,
            "temporary_session",
            cli_manager=mock_cli_manager,
            record_session=record_session,
        )

    assert "real_session" not in str(raised.value)
    assert "temporary_session" not in str(raised.value)
    record_session.assert_not_awaited()


@pytest.mark.asyncio
async def test_session_limit_failure_uses_non_propagating_claim_failure(
    handler, mock_cli_manager, mock_platform
):
    claim = _claim()
    mock_cli_manager.get_or_create_session.side_effect = RuntimeError("session limit")
    result = FailureResult(
        affected=(claim.node,),
        queue_update=None,
        snapshot=_snapshot(),
    )
    with patch.object(
        handler.tree_queue,
        "fail_claim",
        AsyncMock(return_value=result),
    ) as fail_claim:
        await handler.node_runner.process_node(claim)

    fail_claim.assert_awaited_once_with(claim, propagate=False)
    rendered = mock_platform.queue_edit_message.call_args_list[-1].args[2]
    assert "Session limit reached" in rendered


@pytest.mark.asyncio
async def test_non_exit_error_defers_child_failure_until_stream_ends(
    handler, mock_cli_manager, mock_platform, mock_session_store
):
    claim = _claim()
    session = _session([{"type": "error", "error": {"message": "CLI crashed"}}])
    mock_cli_manager.get_or_create_session.return_value = (
        session,
        "session_1",
        False,
    )
    child = NodeUiTarget(
        scope=_SCOPE,
        node_id="child",
        status_message_id="status_child",
    )
    snapshot = _snapshot()
    result = FailureResult(
        affected=(claim.node, child),
        queue_update=None,
        snapshot=snapshot,
    )
    with patch.object(
        handler.tree_queue,
        "fail_claim",
        AsyncMock(return_value=result),
    ) as fail_claim:
        await handler.node_runner.process_node(claim)
    await asyncio.sleep(0)

    assert fail_claim.await_args_list == [
        call(claim, propagate=False),
        call(claim, propagate=True),
    ]
    assert mock_session_store.save_tree_snapshot.call_args_list == [
        call(snapshot),
        call(snapshot),
    ]
    rendered = "\n".join(
        call.args[2] for call in mock_platform.queue_edit_message.call_args_list
    )
    assert "❌ *Error*" in rendered
    assert "CLI crashed" in rendered
    assert "Parent task failed" in rendered


@pytest.mark.asyncio
async def test_provider_error_exit_does_not_mask_or_complete(
    handler, mock_cli_manager, mock_platform
):
    claim = _claim()
    provider_error = "API Error: Request rejected (429)\nProvider rate limit reached."
    session = _session(
        [
            {"type": "error", "error": {"message": provider_error}},
            {"type": "exit", "code": 1},
        ]
    )
    mock_cli_manager.get_or_create_session.return_value = (
        session,
        "session_1",
        False,
    )
    failure = FailureResult(
        affected=(claim.node,),
        queue_update=None,
        snapshot=_snapshot(),
    )
    with (
        patch.object(
            handler.tree_queue,
            "fail_claim",
            AsyncMock(return_value=failure),
        ) as fail_claim,
        patch.object(
            handler.tree_queue,
            "complete_claim",
            AsyncMock(),
        ) as complete_claim,
    ):
        await handler.node_runner.process_node(claim)

    assert fail_claim.await_args_list == [
        call(claim, propagate=False),
        call(claim, propagate=True),
    ]
    complete_claim.assert_not_awaited()
    rendered = mock_platform.queue_edit_message.call_args_list[-1].args[2]
    assert "API Error: Request rejected" in rendered
    assert "Process exited with code" not in rendered
    assert "✅ *Complete*" not in rendered


@pytest.mark.asyncio
async def test_success_exit_still_renders_complete_after_non_exit_error(
    handler, mock_cli_manager, mock_platform
):
    claim = _claim()
    session = _session(
        [
            {"type": "error", "error": {"message": "recoverable warning"}},
            {"type": "exit", "code": 0},
        ]
    )
    mock_cli_manager.get_or_create_session.return_value = (
        session,
        "session_1",
        False,
    )
    failure = FailureResult(
        affected=(claim.node,),
        queue_update=None,
        snapshot=_snapshot(),
    )
    with (
        patch.object(
            handler.tree_queue,
            "fail_claim",
            AsyncMock(return_value=failure),
        ) as fail_claim,
        patch.object(
            handler.tree_queue,
            "complete_claim",
            AsyncMock(return_value=_snapshot()),
        ) as complete_claim,
    ):
        await handler.node_runner.process_node(claim)

    fail_claim.assert_awaited_once_with(
        claim,
        propagate=False,
    )
    complete_claim.assert_awaited_once_with(claim, "session_1")
    assert (
        "✅ *Complete*" in mock_platform.queue_edit_message.call_args_list[-1].args[2]
    )


@pytest.mark.asyncio
async def test_unexpected_runner_exception_uses_detailed_task_failed_ui(
    handler, mock_cli_manager, mock_platform
):
    claim = _claim()

    async def failing_start(*args, **kwargs):
        raise ValueError("runner exploded")
        if False:
            yield {}

    session = MagicMock()
    session.start_task = failing_start
    mock_cli_manager.get_or_create_session.return_value = (
        session,
        "session_1",
        False,
    )
    failure = FailureResult(
        affected=(claim.node,),
        queue_update=None,
        snapshot=_snapshot(),
    )
    with patch.object(
        handler.tree_queue,
        "fail_claim",
        AsyncMock(return_value=failure),
    ) as fail_claim:
        await handler.node_runner.process_node(claim)

    fail_claim.assert_awaited_once_with(
        claim,
        propagate=True,
    )
    rendered = mock_platform.queue_edit_message.call_args_list[-1].args[2]
    assert "Task Failed" in rendered
    assert "runner exploded" in rendered


@pytest.mark.asyncio
async def test_stop_cancellation_preserves_partial_transcript(
    handler, mock_cli_manager, mock_platform
):
    claim = _claim(prompt="work")
    started = asyncio.Event()

    async def start_task(*args, **kwargs):
        yield {
            "type": "assistant",
            "message": {"content": [{"type": "text", "text": "partial answer"}]},
        }
        started.set()
        await asyncio.sleep(60)

    session = MagicMock()
    session.start_task = start_task
    mock_cli_manager.get_or_create_session.return_value = (
        session,
        "session_1",
        False,
    )
    failure = FailureResult(
        affected=(claim.node,),
        queue_update=None,
        snapshot=_snapshot(),
    )
    with patch.object(
        handler.tree_queue,
        "fail_claim",
        AsyncMock(return_value=failure),
    ) as fail_claim:
        task = asyncio.create_task(handler.node_runner.process_node(claim))
        await started.wait()
        task.cancel(CancellationReason.STOP)
        await task

    fail_claim.assert_awaited_once_with(
        claim,
        propagate=False,
    )
    rendered = mock_platform.queue_edit_message.call_args_list[-1].args[2]
    assert "partial answer" in rendered
    assert "⏹ *Stopped\\.*" in rendered
    assert rendered.index("partial answer") < rendered.index("⏹ *Stopped\\.*")


@pytest.mark.asyncio
async def test_global_clear_command_deletes_returned_ids(
    handler, mock_platform, incoming_message_factory
):
    handler.clear_chat = AsyncMock(return_value=frozenset({"100", "101"}))
    incoming = incoming_message_factory(
        text="/clear",
        chat_id="chat_1",
        message_id="150",
    )

    await handler.handle_message(incoming)

    handler.clear_chat.assert_awaited_once_with("telegram", "chat_1")
    mock_platform.queue_delete_messages.assert_awaited_once_with(
        "chat_1",
        ["150", "101", "100"],
        fire_and_forget=False,
    )
    mock_platform.queue_send_message.assert_not_awaited()
    handler.session_store.record_message_id.assert_not_called()


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "failure",
    (RuntimeError("clear failed"), asyncio.CancelledError()),
    ids=("failure", "cancellation"),
)
async def test_interrupted_global_clear_records_its_deferred_command_id(
    handler,
    mock_platform,
    mock_session_store,
    incoming_message_factory,
    failure: BaseException,
) -> None:
    handler.clear_chat = AsyncMock(side_effect=failure)
    incoming = incoming_message_factory(
        text="/clear",
        chat_id="chat_1",
        message_id="150",
    )

    with pytest.raises(type(failure)):
        await handler.handle_message(incoming)

    mock_session_store.record_message_id.assert_called_once_with(
        incoming.platform,
        incoming.chat_id,
        incoming.message_id,
        "in",
        "command",
    )
    mock_platform.queue_delete_messages.assert_not_awaited()


@pytest.mark.asyncio
async def test_startup_notice_is_published_and_recorded_for_clear(
    handler,
    mock_platform,
    mock_session_store,
) -> None:
    mock_platform.queue_send_message.return_value = "startup_1"

    await handler.publish_startup_notice(_startup_notice())

    mock_platform.queue_send_message.assert_awaited_once_with(
        _SCOPE.chat_id,
        "🚀 *Claude Code Proxy is online\\!* \\(Bot API\\)",
        parse_mode="MarkdownV2",
        fire_and_forget=False,
    )
    mock_session_store.record_message_id.assert_called_once_with(
        _SCOPE.platform,
        _SCOPE.chat_id,
        "startup_1",
        "out",
        "startup",
    )
    mock_platform.queue_delete_messages.assert_not_awaited()


@pytest.mark.asyncio
async def test_startup_notice_failure_is_nonfatal_and_records_nothing(
    handler,
    mock_platform,
    mock_session_store,
) -> None:
    mock_platform.queue_send_message.side_effect = RuntimeError("unavailable")

    await handler.publish_startup_notice(_startup_notice())

    mock_session_store.record_message_id.assert_not_called()
    mock_platform.queue_delete_messages.assert_not_awaited()


@pytest.mark.asyncio
async def test_startup_notice_without_delivery_receipt_records_nothing(
    handler,
    mock_platform,
    mock_session_store,
) -> None:
    mock_platform.queue_send_message.return_value = None

    await handler.publish_startup_notice(_startup_notice())

    mock_session_store.record_message_id.assert_not_called()
    mock_platform.queue_delete_messages.assert_not_awaited()


@pytest.mark.asyncio
async def test_startup_notice_record_failure_is_compensated_outside_state_lock(
    handler,
    mock_platform,
    mock_session_store,
) -> None:
    mock_platform.queue_send_message.return_value = "startup_1"
    mock_session_store.record_message_id.side_effect = OSError("store unavailable")

    async def delete_notice(*args, **kwargs) -> None:
        assert not handler._state_lock.locked()

    mock_platform.queue_delete_messages.side_effect = delete_notice

    await handler.publish_startup_notice(_startup_notice())

    mock_session_store.record_message_id.assert_called_once_with(
        _SCOPE.platform,
        _SCOPE.chat_id,
        "startup_1",
        "out",
        "startup",
    )
    mock_platform.queue_delete_messages.assert_awaited_once_with(
        _SCOPE.chat_id,
        ["startup_1"],
        fire_and_forget=False,
    )
    mock_session_store.forget_tracked_message_ids.assert_called_once_with(
        _SCOPE.platform,
        _SCOPE.chat_id,
        {"startup_1"},
    )


@pytest.mark.asyncio
async def test_failed_startup_notice_compensation_restores_clear_ownership(
    handler,
    mock_platform,
    mock_session_store,
) -> None:
    mock_platform.queue_send_message.return_value = "startup_1"
    mock_session_store.record_message_id.side_effect = (
        OSError("store unavailable"),
        None,
    )
    mock_platform.queue_delete_messages.side_effect = RuntimeError("delete unavailable")

    await handler.publish_startup_notice(_startup_notice())

    assert mock_session_store.record_message_id.call_count == 2
    assert mock_session_store.record_message_id.call_args_list == [
        call(
            _SCOPE.platform,
            _SCOPE.chat_id,
            "startup_1",
            "out",
            "startup",
        ),
        call(
            _SCOPE.platform,
            _SCOPE.chat_id,
            "startup_1",
            "out",
            "startup",
        ),
    ]
    mock_session_store.forget_tracked_message_ids.assert_not_called()


@pytest.mark.asyncio
async def test_startup_notice_publication_propagates_cancellation(
    handler,
    mock_platform,
    mock_session_store,
) -> None:
    send_started = asyncio.Event()
    send_cancelled = asyncio.Event()

    async def send_notice(*args, **kwargs) -> None:
        send_started.set()
        try:
            await asyncio.Event().wait()
        finally:
            send_cancelled.set()

    mock_platform.queue_send_message.side_effect = send_notice
    task = asyncio.create_task(handler.publish_startup_notice(_startup_notice()))
    await send_started.wait()

    task.cancel()

    with pytest.raises(asyncio.CancelledError):
        await task
    await send_cancelled.wait()
    mock_session_store.record_message_id.assert_not_called()
    mock_platform.queue_delete_messages.assert_not_awaited()


@pytest.mark.asyncio
async def test_startup_notice_cancellation_after_receipt_finishes_compensation(
    handler,
    mock_platform,
    mock_session_store,
) -> None:
    send_started = asyncio.Event()
    release_send = asyncio.Event()
    finalizer_started = asyncio.Event()

    async def send_notice(*args, **kwargs) -> str:
        send_started.set()
        await release_send.wait()
        return "startup_1"

    original_finalize = handler._finalize_startup_notice

    async def finalize_notice(*args, **kwargs) -> None:
        finalizer_started.set()
        await original_finalize(*args, **kwargs)

    mock_platform.queue_send_message.side_effect = send_notice
    with patch.object(
        handler,
        "_finalize_startup_notice",
        side_effect=finalize_notice,
    ):
        task = asyncio.create_task(handler.publish_startup_notice(_startup_notice()))
        await send_started.wait()
        await handler._state_lock.acquire()
        try:
            release_send.set()
            await finalizer_started.wait()
        finally:
            handler._state_lock.release()
        task.cancel()

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

    mock_session_store.record_message_id.assert_not_called()
    mock_platform.queue_delete_messages.assert_awaited_once_with(
        _SCOPE.chat_id,
        ["startup_1"],
        fire_and_forget=False,
    )
    mock_session_store.forget_tracked_message_ids.assert_called_once_with(
        _SCOPE.platform,
        _SCOPE.chat_id,
        {"startup_1"},
    )


@pytest.mark.asyncio
async def test_concurrent_global_clear_does_not_wait_for_startup_delivery(
    mock_platform,
    mock_cli_manager,
    tmp_path,
) -> None:
    store = SessionStore(storage_path=str(tmp_path / "sessions.json"))
    workflow = MessagingWorkflow(
        mock_platform,
        mock_cli_manager,
        store,
        platform_name="telegram",
        voice_cancellation=mock_platform,
    )
    send_started = asyncio.Event()
    release_send = asyncio.Event()

    async def send_notice(*args, **kwargs) -> str:
        send_started.set()
        await release_send.wait()
        return "startup_1"

    mock_platform.queue_send_message.side_effect = send_notice
    publish_task = asyncio.create_task(
        workflow.publish_startup_notice(_startup_notice())
    )
    await send_started.wait()
    clear_task = asyncio.create_task(
        workflow.clear_chat(_SCOPE.platform, _SCOPE.chat_id)
    )
    try:
        done, _pending = await asyncio.wait({clear_task}, timeout=1)
        assert clear_task in done
        assert clear_task.result() == frozenset()
    finally:
        release_send.set()
        await asyncio.gather(publish_task, clear_task, return_exceptions=True)

    assert store.get_tracked_message_ids_for_chat(_SCOPE.platform, _SCOPE.chat_id) == []
    mock_platform.queue_delete_messages.assert_awaited_once_with(
        _SCOPE.chat_id,
        ["startup_1"],
        fire_and_forget=False,
    )


@pytest.mark.asyncio
async def test_global_stop_does_not_wait_for_or_invalidate_startup_delivery(
    handler,
    mock_platform,
    mock_session_store,
) -> None:
    send_started = asyncio.Event()
    release_send = asyncio.Event()

    async def send_notice(*args, **kwargs) -> str:
        send_started.set()
        await release_send.wait()
        return "startup_1"

    mock_platform.queue_send_message.side_effect = send_notice
    publish_task = asyncio.create_task(
        handler.publish_startup_notice(_startup_notice())
    )
    await send_started.wait()
    stop_task = asyncio.create_task(handler.stop_all_tasks())
    try:
        done, _pending = await asyncio.wait({stop_task}, timeout=1)
        assert stop_task in done
        stop_task.result()
    finally:
        release_send.set()
        await asyncio.gather(publish_task, stop_task, return_exceptions=True)

    mock_session_store.record_message_id.assert_called_once_with(
        _SCOPE.platform,
        _SCOPE.chat_id,
        "startup_1",
        "out",
        "startup",
    )
    mock_platform.queue_delete_messages.assert_not_awaited()


@pytest.mark.asyncio
async def test_global_clear_precedes_concurrent_startup_notice_publication(
    mock_platform,
    mock_cli_manager,
    tmp_path,
) -> None:
    store = SessionStore(storage_path=str(tmp_path / "sessions.json"))
    workflow = MessagingWorkflow(
        mock_platform,
        mock_cli_manager,
        store,
        platform_name="telegram",
        voice_cancellation=mock_platform,
    )
    clear_started = asyncio.Event()
    release_clear = asyncio.Event()

    async def clear_trees(
        scope: MessageScope, *, reason: CancellationReason
    ) -> CancellationResult:
        assert reason is CancellationReason.CLEAR
        clear_started.set()
        await release_clear.wait()
        return CancellationResult()

    with patch.object(workflow.tree_queue, "clear_scope", side_effect=clear_trees):
        clear_task = asyncio.create_task(
            workflow.clear_chat(_SCOPE.platform, _SCOPE.chat_id)
        )
        await clear_started.wait()
        publish_task = asyncio.create_task(
            workflow.publish_startup_notice(_startup_notice())
        )
        await asyncio.sleep(0)
        mock_platform.queue_send_message.assert_not_awaited()

        release_clear.set()

        assert await clear_task == frozenset()
        await publish_task

    assert store.get_tracked_message_ids_for_chat(_SCOPE.platform, _SCOPE.chat_id) == [
        "msg_123"
    ]
    mock_platform.queue_delete_messages.assert_not_awaited()


@pytest.mark.asyncio
async def test_clear_command_cannot_evict_startup_notice_at_managed_message_cap(
    mock_platform,
    mock_cli_manager,
    incoming_message_factory,
    tmp_path,
) -> None:
    store = SessionStore(
        storage_path=str(tmp_path / "sessions.json"),
        managed_message_cap=1,
    )
    workflow = MessagingWorkflow(
        mock_platform,
        mock_cli_manager,
        store,
        platform_name="telegram",
        voice_cancellation=mock_platform,
    )
    store.record_message_id(
        _SCOPE.platform,
        _SCOPE.chat_id,
        "100",
        "out",
        "startup",
    )
    incoming = incoming_message_factory(
        text="/clear",
        chat_id=_SCOPE.chat_id,
        message_id="101",
    )

    await workflow.handle_message(incoming)

    mock_platform.queue_delete_messages.assert_awaited_once_with(
        _SCOPE.chat_id,
        ["101", "100"],
        fire_and_forget=False,
    )


@pytest.mark.asyncio
async def test_clear_chat_is_fully_scoped_to_the_invoking_chat(
    handler, mock_cli_manager, mock_session_store, incoming_message_factory
):
    root_1 = incoming_message_factory(
        text="one",
        chat_id="chat_1",
        message_id="100",
    )
    root_2 = incoming_message_factory(
        text="two",
        chat_id="chat_2",
        message_id="200",
    )
    await handler.tree_queue.admit(root_1, "101")
    await handler.tree_queue.admit(root_2, "201")
    await _wait_for_idle(handler)
    mock_session_store.get_tracked_message_ids_for_chat.return_value = ["42"]
    mock_session_store.reset_mock()
    mock_session_store.get_tracked_message_ids_for_chat.return_value = ["42"]

    message_ids = await handler.clear_chat("telegram", "chat_1")

    assert message_ids == frozenset({"42", "100", "101"})
    assert "200" not in message_ids
    assert handler.get_tree_count() == 1
    assert await handler.tree_queue.get_node(root_2.scope, "200") is not None
    mock_cli_manager.stop_all.assert_not_awaited()
    mock_session_store.clear_scope.assert_called_once_with(_SCOPE)


@pytest.mark.asyncio
async def test_global_clear_cancels_and_deletes_only_current_chat_voices(
    handler,
    mock_platform,
    mock_session_store,
) -> None:
    events: list[str] = []
    current = VoiceCancellationResult(
        scope=_SCOPE,
        voice_message_id="voice",
        status_message_id="voice_status",
        delete_message_ids=frozenset({"voice", "voice_status"}),
    )

    async def cancel_voices(scope: MessageScope) -> tuple[VoiceCancellationResult, ...]:
        assert not handler._state_lock.locked()
        assert scope == _SCOPE
        events.append("voices")
        return (current,)

    async def clear_trees(
        scope: MessageScope, *, reason: CancellationReason
    ) -> CancellationResult:
        assert reason is CancellationReason.CLEAR
        assert handler._state_lock.locked()
        events.append("trees")
        return CancellationResult()

    mock_platform.cancel_pending_voices_in_scope.side_effect = cancel_voices
    mock_session_store.get_tracked_message_ids_for_chat.return_value = ["stored"]
    with patch.object(handler.tree_queue, "clear_scope", side_effect=clear_trees):
        message_ids = await handler.clear_chat("telegram", "chat_1")
    await asyncio.sleep(0)

    assert events == ["voices", "trees"]
    assert message_ids == frozenset({"stored", "voice", "voice_status"})
    assert "other_voice" not in message_ids
    assert "other_status" not in message_ids
    mock_platform.queue_edit_message.assert_not_awaited()


@pytest.mark.asyncio
async def test_global_clear_persists_after_tree_detach(
    handler,
    mock_cli_manager,
    mock_session_store,
) -> None:
    events: list[str] = []

    async def clear_trees(
        scope: MessageScope, *, reason: CancellationReason
    ) -> CancellationResult:
        assert scope == _SCOPE
        assert reason is CancellationReason.CLEAR
        events.append("trees.clear_scope")
        return CancellationResult()

    mock_session_store.clear_scope.side_effect = lambda scope: events.append(
        f"store.clear_scope:{scope.chat_id}"
    )

    with patch.object(handler.tree_queue, "clear_scope", side_effect=clear_trees):
        result = await handler.clear_chat("telegram", "chat_1")

    assert result == frozenset()
    assert events == ["trees.clear_scope", "store.clear_scope:chat_1"]
    mock_cli_manager.stop_all.assert_not_awaited()


@pytest.mark.asyncio
async def test_global_clear_finishes_cleanup_before_final_persistence_error_escapes(
    handler,
    mock_cli_manager,
    mock_session_store,
) -> None:
    events: list[str] = []

    async def clear_trees(
        scope: MessageScope, *, reason: CancellationReason
    ) -> CancellationResult:
        assert scope == _SCOPE
        assert reason is CancellationReason.CLEAR
        events.append("trees.clear_scope")
        return CancellationResult()

    def fail_final_clear(scope: MessageScope) -> None:
        assert scope == _SCOPE
        events.append("store.clear_scope")
        raise OSError("final clear failure")

    mock_session_store.clear_scope.side_effect = fail_final_clear

    with (
        patch.object(handler.tree_queue, "clear_scope", side_effect=clear_trees),
        pytest.raises(OSError, match="final clear failure"),
    ):
        await handler.clear_chat("telegram", "chat_1")

    assert events == ["trees.clear_scope", "store.clear_scope"]
    mock_cli_manager.stop_all.assert_not_awaited()


@pytest.mark.asyncio
async def test_global_clear_preserves_tree_and_persistence_failures(
    handler,
    mock_cli_manager,
    mock_session_store,
) -> None:
    events: list[str] = []
    persistence_error = OSError("final clear failure")
    tree_error = RuntimeError("tree clear failure")

    async def fail_tree_clear(
        scope: MessageScope, *, reason: CancellationReason
    ) -> CancellationResult:
        assert scope == _SCOPE
        assert reason is CancellationReason.CLEAR
        events.append("trees.clear_scope")
        raise tree_error

    def fail_final_clear(scope: MessageScope) -> None:
        assert scope == _SCOPE
        events.append("store.clear_scope")
        raise persistence_error

    mock_session_store.clear_scope.side_effect = fail_final_clear

    with (
        patch.object(handler.tree_queue, "clear_scope", side_effect=fail_tree_clear),
        pytest.raises(ExceptionGroup) as raised,
    ):
        await handler.clear_chat("telegram", "chat_1")

    assert raised.value.exceptions == (tree_error, persistence_error)
    assert events == ["trees.clear_scope", "store.clear_scope"]
    mock_cli_manager.stop_all.assert_not_awaited()


@pytest.mark.asyncio
async def test_committed_global_clear_attempts_remaining_steps_after_tree_failure(
    handler,
    mock_cli_manager,
    mock_session_store,
) -> None:
    events: list[str] = []
    tree_error = RuntimeError("tree clear failure")

    async def fail_tree_clear(
        scope: MessageScope, *, reason: CancellationReason
    ) -> CancellationResult:
        assert scope == _SCOPE
        assert reason is CancellationReason.CLEAR
        events.append("trees.clear_scope")
        raise tree_error

    mock_session_store.clear_scope.side_effect = lambda scope: events.append(
        f"store.clear_scope:{scope.chat_id}"
    )

    with (
        patch.object(handler.tree_queue, "clear_scope", side_effect=fail_tree_clear),
        pytest.raises(RuntimeError, match="tree clear failure") as raised,
    ):
        await handler.clear_chat("telegram", "chat_1")

    assert raised.value is tree_error
    assert events == [
        "trees.clear_scope",
        "store.clear_scope:chat_1",
    ]
    mock_cli_manager.stop_all.assert_not_awaited()


@pytest.mark.asyncio
async def test_cancelled_global_clear_finishes_owned_transaction_before_propagating(
    handler,
    mock_cli_manager,
    mock_session_store,
    incoming_message_factory,
):
    root = incoming_message_factory(text="work", message_id="100")
    await handler.tree_queue.admit(root, "101")
    await _wait_for_idle(handler)
    id_read_started = asyncio.Event()
    release_id_read = asyncio.Event()

    async def block_id_read(platform: str, chat_id: str) -> set[str]:
        id_read_started.set()
        await release_id_read.wait()
        return {"100", "101"}

    mock_session_store.reset_mock()
    with patch.object(
        handler.tree_queue,
        "get_message_ids_for_chat",
        new=block_id_read,
    ):
        clear_task = asyncio.create_task(handler.clear_chat("telegram", root.chat_id))
        await id_read_started.wait()
        clear_task.cancel()
        await asyncio.sleep(0)
        assert not clear_task.done()
        release_id_read.set()
        with pytest.raises(asyncio.CancelledError):
            await clear_task

    assert handler._clear_generations[_SCOPE] == 1
    assert handler.get_tree_count() == 0
    mock_session_store.clear_scope.assert_called_once_with(_SCOPE)
    mock_cli_manager.stop_all.assert_not_awaited()


@pytest.mark.asyncio
async def test_clear_with_mention_uses_same_global_command(
    handler, mock_platform, incoming_message_factory
):
    handler.clear_chat = AsyncMock(return_value=frozenset())
    incoming = incoming_message_factory(
        text="/clear@MyBot",
        chat_id="chat_1",
        message_id="10",
    )

    await handler.handle_message(incoming)

    handler.clear_chat.assert_awaited_once_with("telegram", "chat_1")
    mock_platform.queue_delete_messages.assert_awaited_once_with(
        "chat_1",
        ["10"],
        fire_and_forget=False,
    )


@pytest.mark.asyncio
async def test_clear_continues_after_platform_delete_failure(
    handler, mock_platform, incoming_message_factory
):
    handler.clear_chat = AsyncMock(return_value=frozenset({"41", "42"}))
    mock_platform.queue_delete_messages.side_effect = RuntimeError(
        "platform rejected delete"
    )

    await handler.handle_message(
        incoming_message_factory(text="/clear", message_id="150")
    )

    handler.clear_chat.assert_awaited_once()
    mock_platform.queue_delete_messages.assert_awaited_once()


@pytest.mark.asyncio
async def test_reply_clear_status_preserves_prompt_and_persists_remaining_tree(
    handler,
    mock_platform,
    mock_session_store,
    incoming_message_factory,
):
    root = incoming_message_factory(text="root", message_id="100")
    child = incoming_message_factory(
        text="child",
        message_id="102",
        reply_to_message_id="100",
    )
    await handler.tree_queue.admit(root, "101")
    await _wait_for_idle(handler)
    await handler.tree_queue.admit(child, "103", parent_reference_id="100")
    await _wait_for_idle(handler)
    mock_session_store.reset_mock()
    deleted_ids: list[str] = []

    async def capture_delete(chat_id, message_ids, fire_and_forget=True):
        deleted_ids.extend(message_ids)

    mock_platform.queue_delete_messages.side_effect = capture_delete
    await handler.handle_message(
        incoming_message_factory(
            text="/clear",
            message_id="150",
            reply_to_message_id="103",
        )
    )

    assert set(deleted_ids) == {"103", "150"}
    assert "102" not in deleted_ids
    assert "100" not in deleted_ids
    assert "101" not in deleted_ids
    cleared_prompt = await handler.tree_queue.get_node(root.scope, "102")
    assert cleared_prompt is not None
    assert cleared_prompt.state is MessageState.ERROR
    assert cleared_prompt.session_id is None
    assert await handler.tree_queue.get_node(root.scope, "100") is not None
    mock_session_store.save_tree_snapshot.assert_called_once()
    mock_session_store.record_message_id.assert_called_once_with(
        "telegram",
        "chat_1",
        "150",
        "in",
        "command",
    )
    mock_session_store.forget_tracked_message_ids.assert_called_once_with(
        "telegram",
        "chat_1",
        {"103", "150"},
    )


@pytest.mark.asyncio
async def test_reply_clear_unknown_reports_nothing_to_clear(
    handler, mock_platform, mock_session_store, incoming_message_factory
):
    incoming = incoming_message_factory(
        text="/clear",
        message_id="150",
        reply_to_message_id="999",
    )

    await handler.handle_message(incoming)

    assert "Nothing to clear" in mock_platform.queue_send_message.call_args.args[1]
    mock_session_store.record_message_id.assert_any_call(
        incoming.platform,
        incoming.chat_id,
        incoming.message_id,
        "in",
        "command",
    )
    mock_session_store.clear_scope.assert_not_called()


@pytest.mark.asyncio
async def test_reply_clear_root_removes_tree_snapshot(
    handler,
    mock_platform,
    mock_session_store,
    incoming_message_factory,
):
    root = incoming_message_factory(text="root", message_id="100")
    await handler.tree_queue.admit(root, "101")
    await _wait_for_idle(handler)
    mock_session_store.reset_mock()
    deleted_ids: list[str] = []

    async def capture_delete(chat_id, message_ids, fire_and_forget=True):
        deleted_ids.extend(message_ids)

    mock_platform.queue_delete_messages.side_effect = capture_delete
    await handler.handle_message(
        incoming_message_factory(
            text="/clear",
            message_id="150",
            reply_to_message_id="100",
        )
    )

    assert set(deleted_ids) == {"100", "101", "150"}
    mock_session_store.remove_tree_snapshot.assert_called_once_with(
        TreeIdentity(scope=root.scope, root_id="100")
    )
    assert handler.get_tree_count() == 0


@pytest.mark.asyncio
async def test_late_cancelled_runner_cannot_save_or_render_after_chat_clear(
    handler,
    mock_platform,
    mock_cli_manager,
    mock_session_store,
    incoming_message_factory,
):
    started = asyncio.Event()

    async def blocking_start(*args, **kwargs):
        started.set()
        await asyncio.sleep(60)
        if False:
            yield {}

    session = MagicMock()
    session.start_task = blocking_start
    mock_cli_manager.get_or_create_session.return_value = (
        session,
        "pending_1",
        True,
    )
    await handler.handle_message(
        incoming_message_factory(text="work", message_id="100")
    )
    await started.wait()
    mock_session_store.reset_mock()
    mock_platform.queue_edit_message.reset_mock()

    await handler.clear_chat("telegram", "chat_1")

    mock_session_store.save_tree_snapshot.assert_not_called()
    mock_session_store.clear_scope.assert_called_once_with(_SCOPE)
    mock_platform.queue_edit_message.assert_not_awaited()
    assert handler.get_tree_count() == 0


@pytest.mark.asyncio
async def test_global_clear_removes_snapshot_saved_during_detach_window(
    tmp_path,
    mock_platform,
    mock_cli_manager,
    incoming_message_factory,
):
    runner_started = asyncio.Event()
    release_runner = asyncio.Event()
    id_read_started = asyncio.Event()
    release_id_read = asyncio.Event()

    async def finish_after_release(*args, **kwargs):
        runner_started.set()
        await release_runner.wait()
        yield {"type": "exit", "code": 0}

    session = MagicMock()
    session.start_task = finish_after_release
    mock_cli_manager.get_or_create_session.return_value = (
        session,
        "session_1",
        False,
    )
    mock_platform.queue_send_message.return_value = "status-new"
    store_path = tmp_path / "sessions.json"
    store = SessionStore(storage_path=str(store_path))
    workflow = MessagingWorkflow(
        mock_platform,
        mock_cli_manager,
        store,
        platform_name="telegram",
    )
    incoming = incoming_message_factory(text="work", message_id="new")
    await workflow.handle_message(incoming)
    await runner_started.wait()

    get_ids = workflow.tree_queue.get_message_ids_for_chat

    async def block_id_read(platform: str, chat_id: str) -> set[str]:
        id_read_started.set()
        await release_id_read.wait()
        return await get_ids(platform, chat_id)

    try:
        with patch.object(
            workflow.tree_queue,
            "get_message_ids_for_chat",
            new=block_id_read,
        ):
            clear_task = asyncio.create_task(
                workflow.clear_chat("telegram", incoming.chat_id)
            )
            await id_read_started.wait()
            release_runner.set()
            await _wait_for_idle(workflow)

            assert not store.load_conversation_snapshot().is_empty
            store.flush_pending_save()
            assert (
                not SessionStore(storage_path=str(store_path))
                .load_conversation_snapshot()
                .is_empty
            )
            release_id_read.set()
            await clear_task

        assert store.load_conversation_snapshot().is_empty
        assert (
            SessionStore(storage_path=str(store_path))
            .load_conversation_snapshot()
            .is_empty
        )
    finally:
        release_runner.set()
        release_id_read.set()
        await workflow.close()


@pytest.mark.asyncio
async def test_global_clear_invalidates_inflight_prompt_without_waiting_for_status(
    handler,
    mock_platform,
    incoming_message_factory,
):
    status_send_started = asyncio.Event()
    release_status_send = asyncio.Event()

    async def block_status_send(*args, **kwargs):
        status_send_started.set()
        await release_status_send.wait()
        return "status-new"

    mock_platform.queue_send_message.side_effect = block_status_send
    prompt_task = asyncio.create_task(
        handler.handle_message(
            incoming_message_factory(text="new prompt", message_id="new")
        )
    )
    await status_send_started.wait()
    clear_task = asyncio.create_task(handler.clear_chat("telegram", "chat_1"))

    try:
        await asyncio.wait_for(clear_task, timeout=1)
        release_status_send.set()
        await prompt_task
        assert handler.get_tree_count() == 0
        mock_platform.queue_delete_messages.assert_awaited_once_with(
            "chat_1",
            ["status-new"],
            fire_and_forget=False,
        )
    finally:
        release_status_send.set()
        if not prompt_task.done():
            prompt_task.cancel()
        if not clear_task.done():
            clear_task.cancel()


@pytest.mark.asyncio
@pytest.mark.parametrize(
    ("status_message_id", "expected_deleted_ids"),
    [
        ("101", {"100", "101", "150"}),
        (None, {"100", "150"}),
    ],
)
async def test_reply_clear_pending_voice_cancels_and_reports(
    handler,
    mock_platform,
    incoming_message_factory,
    status_message_id,
    expected_deleted_ids,
) -> None:
    delete_message_ids = {"100"}
    if status_message_id is not None:
        delete_message_ids.add(status_message_id)
    handler.clear_reply = AsyncMock(
        return_value=ReplyClearResult(
            delete_message_ids=frozenset(delete_message_ids),
            tree_matched=False,
        )
    )
    deleted_ids: list[str] = []

    async def capture_delete(chat_id, message_ids, fire_and_forget=True):
        deleted_ids.extend(message_ids)

    mock_platform.queue_delete_messages.side_effect = capture_delete
    incoming = incoming_message_factory(
        text="/clear",
        message_id="150",
        reply_to_message_id="100",
    )

    await handler.handle_message(incoming)

    handler.clear_reply.assert_awaited_once_with(incoming.scope, "100")
    assert set(deleted_ids) == expected_deleted_ids
    mock_platform.queue_send_message.assert_not_awaited()


@pytest.mark.asyncio
async def test_reply_clear_deletes_owned_result_ids(
    handler,
    mock_platform,
    incoming_message_factory,
) -> None:
    handler.clear_reply = AsyncMock(
        return_value=ReplyClearResult(
            delete_message_ids=frozenset({"tree_status"}),
            tree_matched=True,
        )
    )
    deleted_ids: list[str] = []

    async def capture_delete(_chat_id, message_ids, fire_and_forget=True):
        deleted_ids.extend(message_ids)

    mock_platform.queue_delete_messages.side_effect = capture_delete
    incoming = incoming_message_factory(
        text="/clear",
        message_id="clear",
        reply_to_message_id="voice",
    )

    await handler.handle_message(incoming)

    handler.clear_reply.assert_awaited_once_with(incoming.scope, "voice")
    assert set(deleted_ids) == {
        "tree_status",
        "clear",
    }
    assert "voice" not in deleted_ids
    mock_platform.queue_send_message.assert_not_awaited()


@pytest.mark.asyncio
async def test_reply_clear_joins_voice_then_unions_authoritative_branch_ids(
    handler,
    mock_platform,
) -> None:
    events: list[str] = []

    async def cancel_voice(*_args) -> VoiceCancellationResult:
        events.append("voice")
        return VoiceCancellationResult(
            scope=_SCOPE,
            voice_message_id="voice",
            status_message_id="voice_status",
            delete_message_ids=frozenset({"voice", "voice_status"}),
        )

    branch = MessageSubtreeRemovalResult(
        cancellation=CancellationResult(),
        removed_tree_identity=None,
        delete_message_ids=frozenset({"tree_status"}),
        tree_matched=True,
    )
    mock_platform.cancel_pending_voice.side_effect = cancel_voice
    with patch.object(
        handler.tree_queue,
        "remove_message_subtree",
        new_callable=AsyncMock,
        return_value=branch,
    ) as remove_message_subtree:
        result = await handler.clear_reply(_SCOPE, "voice")
    await asyncio.sleep(0)

    assert result == ReplyClearResult(
        delete_message_ids=frozenset({"voice", "voice_status", "tree_status"}),
        tree_matched=True,
    )
    assert events == ["voice"]
    remove_message_subtree.assert_awaited_once_with(
        _SCOPE,
        "voice",
        reason=CancellationReason.CLEAR,
    )
    mock_platform.queue_edit_message.assert_not_awaited()


@pytest.mark.asyncio
async def test_cancelled_reply_clear_finishes_voice_only_owner_after_lock_wait(
    handler,
    mock_platform,
) -> None:
    voice_returned = asyncio.Event()

    async def cancel_voice(*_args) -> VoiceCancellationResult:
        voice_returned.set()
        return VoiceCancellationResult(
            scope=_SCOPE,
            voice_message_id="voice",
            status_message_id="voice_status",
            delete_message_ids=frozenset({"voice", "voice_status"}),
        )

    mock_platform.cancel_pending_voice.side_effect = cancel_voice
    remove_subtree = AsyncMock(
        return_value=MessageSubtreeRemovalResult(
            cancellation=CancellationResult(),
            removed_tree_identity=None,
            delete_message_ids=frozenset(),
            tree_matched=False,
        )
    )
    await handler._state_lock.acquire()
    try:
        with patch.object(
            handler.tree_queue,
            "remove_message_subtree",
            remove_subtree,
        ):
            clear_task = asyncio.create_task(handler.clear_reply(_SCOPE, "voice"))
            await voice_returned.wait()
            await asyncio.sleep(0)
            clear_task.cancel()
            await asyncio.sleep(0)

            assert not clear_task.done()
            handler._state_lock.release()
            with pytest.raises(asyncio.CancelledError):
                await clear_task
    finally:
        if handler._state_lock.locked():
            handler._state_lock.release()
    await asyncio.sleep(0)

    remove_subtree.assert_awaited_once_with(
        _SCOPE,
        "voice",
        reason=CancellationReason.CLEAR,
    )
    mock_platform.queue_edit_message.assert_not_awaited()