"""Tests for Matrix thread tag state helpers."""

from __future__ import annotations

import asyncio
import json
import math
from typing import TYPE_CHECKING
from unittest.mock import AsyncMock, MagicMock, patch

import nio
import pytest

from mindroom.matrix.client_thread_history import RoomThreadsPageError, enumerate_room_thread_root_ids
from mindroom.matrix.conversation_cache import resolve_thread_root_event_id_for_client
from mindroom.thread_tags import (
    COERCED_TAG_MAX_LENGTH,
    THREAD_TAGS_EVENT_TYPE,
    SetThreadTagsIfEmptyResult,
    ThreadTagsError,
    ThreadTagsListing,
    ThreadTagsState,
    coerce_tag_name,
    get_thread_tags,
    list_tagged_threads,
    remove_thread_tag,
    set_thread_tag,
    set_thread_tags_if_empty,
)

if TYPE_CHECKING:
    from collections.abc import Awaitable


def _message_event_response(
    event_id: str,
    *,
    content: dict[str, object],
    room_id: str = "!room:localhost",
    event_type: str = "m.room.message",
) -> nio.RoomGetEventResponse:
    return nio.RoomGetEventResponse.from_dict(
        {
            "content": content,
            "event_id": event_id,
            "sender": "@user:localhost",
            "origin_server_ts": 1,
            "room_id": room_id,
            "type": event_type,
        },
    )


def _thread_root_event(event_id: str) -> nio.RoomMessageText:
    return nio.RoomMessageText.from_dict(
        {
            "event_id": event_id,
            "sender": "@user:localhost",
            "origin_server_ts": 1,
            "room_id": "!room:localhost",
            "type": "m.room.message",
            "content": {"body": "Root", "msgtype": "m.text"},
        },
    )


def _tag_record_content(
    *,
    set_by: str = "@user:localhost",
    set_at: str = "2026-03-21T19:02:03+00:00",
    note: str | None = None,
    data: dict[str, object] | None = None,
) -> dict[str, object]:
    content: dict[str, object] = {
        "set_by": set_by,
        "set_at": set_at,
        "data": data or {},
    }
    if note is not None:
        content["note"] = note
    return content


def _thread_tags_content(**tags: dict[str, object]) -> dict[str, object]:
    return {"tags": tags}


def _thread_tag_state_key(thread_root_id: str, tag: str) -> str:
    return json.dumps([thread_root_id, tag], separators=(",", ":"))


def _thread_tag_state_event(
    thread_root_id: str,
    tag: str,
    *,
    content: dict[str, object] | None = None,
) -> dict[str, object]:
    return {
        "type": THREAD_TAGS_EVENT_TYPE,
        "state_key": _thread_tag_state_key(thread_root_id, tag),
        "content": content if content is not None else _tag_record_content(),
    }


def _legacy_thread_tags_event(
    thread_root_id: str,
    *,
    content: dict[str, object] | None = None,
) -> dict[str, object]:
    return {
        "type": THREAD_TAGS_EVENT_TYPE,
        "state_key": thread_root_id,
        "content": content if content is not None else _thread_tags_content(resolved=_tag_record_content()),
    }


def _thread_tags_room_state_response(*events: dict[str, object]) -> nio.RoomGetStateResponse:
    return nio.RoomGetStateResponse(
        events=list(events),
        room_id="!room:localhost",
    )


def _thread_tags_room_state_from_current(
    current_events: dict[str, dict[str, object]],
) -> nio.RoomGetStateResponse:
    return _thread_tags_room_state_response(
        *[
            {
                "type": THREAD_TAGS_EVENT_TYPE,
                "state_key": state_key,
                "content": content,
            }
            for state_key, content in current_events.items()
        ],
    )


def _power_levels_response(
    *,
    users: dict[str, int],
    state_default: int = 50,
    users_default: int | None = None,
    events: dict[str, int] | None = None,
) -> nio.RoomGetStateEventResponse:
    content: dict[str, object] = {
        "users": users,
        "state_default": state_default,
    }
    if users_default is not None:
        content["users_default"] = users_default
    if events is not None:
        content["events"] = events
    return nio.RoomGetStateEventResponse(
        content=content,
        event_type="m.room.power_levels",
        state_key="",
        room_id="!room:localhost",
    )


def _thread_tags_state_response(
    thread_root_id: str,
    *,
    content: dict[str, object] | None = None,
) -> nio.RoomGetStateEventResponse:
    return nio.RoomGetStateEventResponse(
        content=content if content is not None else _thread_tags_content(resolved=_tag_record_content()),
        event_type=THREAD_TAGS_EVENT_TYPE,
        state_key=thread_root_id,
        room_id="!room:localhost",
    )


def _thread_tags_state_error(
    *,
    message: str,
    status_code: str,
) -> nio.RoomGetStateEventError:
    return nio.RoomGetStateEventError(
        message,
        status_code=status_code,
        room_id="!room:localhost",
    )


def _joined_members_response(*user_ids: str) -> nio.JoinedMembersResponse:
    return nio.JoinedMembersResponse.from_dict(
        {
            "joined": {
                user_id: {"display_name": user_id.removeprefix("@").split(":", 1)[0].title()} for user_id in user_ids
            },
        },
        room_id="!room:localhost",
    )


def _tag_writer_client(
    user_id: str,
    current_events: dict[str, dict[str, object]],
) -> AsyncMock:
    """Build one writable Matrix client backed by shared in-memory tag state."""
    client = AsyncMock()
    client.user_id = user_id
    client.room_get_state_event.return_value = _power_levels_response(users={user_id: 50})

    async def room_get_state(room_id: str) -> object:
        assert room_id == "!room:localhost"
        return _thread_tags_room_state_from_current(current_events)

    async def room_put_state(**kwargs: object) -> object:
        current_events[kwargs["state_key"]] = kwargs["content"]
        return nio.RoomPutStateResponse.from_dict(
            {"event_id": "$state"},
            room_id="!room:localhost",
        )

    client.room_get_state.side_effect = room_get_state
    client.room_put_state.side_effect = room_put_state
    return client


def _write_operation(
    client: AsyncMock,
    action: str,
) -> Awaitable[object]:
    if action == "set":
        return set_thread_tag(
            client,
            "!room:localhost",
            "$thread-root:localhost",
            "resolved",
            set_by="@alice:localhost",
        )
    return remove_thread_tag(
        client,
        "!room:localhost",
        "$thread-root:localhost",
        "resolved",
        requester_user_id="@alice:localhost",
    )


@pytest.mark.asyncio
async def test_set_thread_tag_fetches_fresh_membership_and_power_levels_even_with_synced_cache() -> None:
    """Thread-tag writes should re-check current membership and power levels from Matrix."""
    client = AsyncMock()
    client.user_id = "@mindroom_general:localhost"
    room = nio.MatrixRoom("!room:localhost", "@mindroom_general:localhost")
    room.members_synced = True
    room.users = {
        "@mindroom_general:localhost": object(),
        "@alice:localhost": object(),
    }
    room.power_levels.defaults.state_default = 50
    room.power_levels.users["@mindroom_general:localhost"] = 50
    room.power_levels.users["@alice:localhost"] = 50
    client.rooms = {"!room:localhost": room}
    client.joined_members.return_value = _joined_members_response(
        "@mindroom_general:localhost",
        "@alice:localhost",
    )
    client.room_get_state_event.return_value = _power_levels_response(
        users={
            "@mindroom_general:localhost": 50,
            "@alice:localhost": 50,
        },
    )

    current_events: dict[str, dict[str, object]] = {}

    async def room_get_state(room_id: str) -> object:
        assert room_id == "!room:localhost"
        return _thread_tags_room_state_from_current(current_events)

    async def room_put_state(**kwargs: object) -> object:
        current_events[kwargs["state_key"]] = kwargs["content"]
        return nio.RoomPutStateResponse.from_dict(
            {"event_id": "$state"},
            room_id="!room:localhost",
        )

    client.room_get_state.side_effect = room_get_state
    client.room_put_state.side_effect = room_put_state

    state = await set_thread_tag(
        client,
        "!room:localhost",
        "$thread-root:localhost",
        "resolved",
        set_by="@alice:localhost",
    )

    assert state.tags["resolved"].set_by == "@alice:localhost"
    client.joined_members.assert_awaited_once_with("!room:localhost")
    client.room_get_state_event.assert_awaited_once_with(
        room_id="!room:localhost",
        event_type="m.room.power_levels",
    )


@pytest.mark.asyncio
async def test_set_thread_tag_writes_state_and_returns_state() -> None:
    """Thread tags should be written to the expected room state key."""
    client = AsyncMock()
    client.user_id = "@mindroom_general:localhost"
    client.joined_members.return_value = _joined_members_response(
        "@mindroom_general:localhost",
        "@alice:localhost",
    )

    current_events: dict[str, dict[str, object]] = {}

    async def room_get_state_event(**kwargs: object) -> object:
        assert kwargs["event_type"] == "m.room.power_levels"
        return _power_levels_response(
            users={
                "@mindroom_general:localhost": 50,
                "@alice:localhost": 50,
            },
        )

    async def room_get_state(room_id: str) -> object:
        assert room_id == "!room:localhost"
        return _thread_tags_room_state_from_current(current_events)

    async def room_put_state(**kwargs: object) -> object:
        current_events[kwargs["state_key"]] = kwargs["content"]
        return nio.RoomPutStateResponse.from_dict(
            {"event_id": "$state"},
            room_id="!room:localhost",
        )

    client.room_get_state_event.side_effect = room_get_state_event
    client.room_get_state.side_effect = room_get_state
    client.room_put_state.side_effect = room_put_state

    state = await set_thread_tag(
        client,
        "!room:localhost",
        "$thread-root:localhost",
        "resolved",
        set_by="@alice:localhost",
        note="Fixed in abc123",
    )

    client.room_put_state.assert_awaited_once()
    _, kwargs = client.room_put_state.await_args
    assert kwargs["room_id"] == "!room:localhost"
    assert kwargs["event_type"] == THREAD_TAGS_EVENT_TYPE
    assert kwargs["state_key"] == _thread_tag_state_key("$thread-root:localhost", "resolved")
    assert kwargs["content"]["set_by"] == "@alice:localhost"
    assert kwargs["content"]["note"] == "Fixed in abc123"
    assert kwargs["content"]["data"] == {}
    assert state.thread_root_id == "$thread-root:localhost"
    assert list(state.tags) == ["resolved"]
    assert state.tags["resolved"].set_by == "@alice:localhost"
    assert state.tags["resolved"].note == "Fixed in abc123"


@pytest.mark.asyncio
async def test_manual_tag_started_first_prevents_automatic_tag_batch() -> None:
    """A manual write holding the shared lock should win before the automatic empty check."""
    current_events: dict[str, dict[str, object]] = {}
    manual_client = _tag_writer_client("@manual:localhost", current_events)
    automatic_client = _tag_writer_client("@automatic:localhost", current_events)
    manual_put_entered = asyncio.Event()
    release_manual_put = asyncio.Event()
    automatic_started = asyncio.Event()
    manual_task: asyncio.Task[ThreadTagsState] | None = None
    automatic_task: asyncio.Task[SetThreadTagsIfEmptyResult] | None = None

    async def blocked_manual_put(**kwargs: object) -> object:
        manual_put_entered.set()
        await release_manual_put.wait()
        current_events[kwargs["state_key"]] = kwargs["content"]
        return nio.RoomPutStateResponse.from_dict(
            {"event_id": "$manual"},
            room_id="!room:localhost",
        )

    async def run_automatic_batch() -> SetThreadTagsIfEmptyResult:
        automatic_started.set()
        return await set_thread_tags_if_empty(
            automatic_client,
            "!room:localhost",
            "$manual-first-root:localhost",
            ["auto-one", "auto-two"],
            set_by="@automatic:localhost",
        )

    manual_client.room_put_state.side_effect = blocked_manual_put
    try:
        manual_task = asyncio.create_task(
            set_thread_tag(
                manual_client,
                "!room:localhost",
                "$manual-first-root:localhost",
                "manual",
                set_by="@manual:localhost",
            ),
        )
        await asyncio.wait_for(manual_put_entered.wait(), timeout=1)

        automatic_task = asyncio.create_task(run_automatic_batch())
        await asyncio.wait_for(automatic_started.wait(), timeout=1)
        automatic_client.room_get_state.assert_not_awaited()
        automatic_client.room_put_state.assert_not_awaited()

        release_manual_put.set()
        await asyncio.gather(manual_task, automatic_task)
    finally:
        release_manual_put.set()
        pending_tasks = [task for task in (manual_task, automatic_task) if task is not None]
        if pending_tasks:
            await asyncio.gather(*pending_tasks, return_exceptions=True)

    result = automatic_task.result()
    assert result == SetThreadTagsIfEmptyResult(
        had_existing_tags=True,
        applied_tags=(),
        failed_tags=(),
    )
    assert set(current_events) == {_thread_tag_state_key("$manual-first-root:localhost", "manual")}


@pytest.mark.asyncio
async def test_manual_remove_started_first_prevents_automatic_tag_batch() -> None:
    """A completed concurrent removal should preserve the manual choice to leave a thread untagged."""
    thread_root_id = "$manual-remove-first-root:localhost"
    manual_state_key = _thread_tag_state_key(thread_root_id, "manual")
    current_events = {
        manual_state_key: _tag_record_content(set_by="@manual:localhost"),
    }
    manual_client = _tag_writer_client("@manual:localhost", current_events)
    automatic_client = _tag_writer_client("@automatic:localhost", current_events)
    manual_remove_entered = asyncio.Event()
    release_manual_remove = asyncio.Event()
    automatic_started = asyncio.Event()
    manual_task: asyncio.Task[ThreadTagsState] | None = None
    automatic_task: asyncio.Task[SetThreadTagsIfEmptyResult] | None = None

    async def blocked_manual_remove(**kwargs: object) -> object:
        manual_remove_entered.set()
        await release_manual_remove.wait()
        current_events[kwargs["state_key"]] = kwargs["content"]
        return nio.RoomPutStateResponse.from_dict(
            {"event_id": "$manual-remove"},
            room_id="!room:localhost",
        )

    async def run_automatic_batch() -> SetThreadTagsIfEmptyResult:
        automatic_started.set()
        return await set_thread_tags_if_empty(
            automatic_client,
            "!room:localhost",
            thread_root_id,
            ["auto"],
            set_by="@automatic:localhost",
        )

    manual_client.room_put_state.side_effect = blocked_manual_remove
    try:
        manual_task = asyncio.create_task(
            remove_thread_tag(
                manual_client,
                "!room:localhost",
                thread_root_id,
                "manual",
                requester_user_id="@manual:localhost",
            ),
        )
        await asyncio.wait_for(manual_remove_entered.wait(), timeout=1)

        automatic_task = asyncio.create_task(run_automatic_batch())
        await asyncio.wait_for(automatic_started.wait(), timeout=1)
        automatic_client.room_get_state.assert_not_awaited()
        automatic_client.room_put_state.assert_not_awaited()

        release_manual_remove.set()
        await asyncio.gather(manual_task, automatic_task)
    finally:
        release_manual_remove.set()
        pending_tasks = [task for task in (manual_task, automatic_task) if task is not None]
        if pending_tasks:
            await asyncio.gather(*pending_tasks, return_exceptions=True)

    assert manual_task.result().tags == {}
    assert automatic_task.result() == SetThreadTagsIfEmptyResult(
        had_existing_tags=False,
        applied_tags=(),
        failed_tags=(),
        skipped_due_to_prior_mutation=True,
    )
    automatic_client.room_get_state.assert_awaited_once_with("!room:localhost")
    automatic_client.room_put_state.assert_not_awaited()
    assert current_events == {manual_state_key: {}}


@pytest.mark.asyncio
async def test_completed_manual_removal_survives_the_model_inference_gap() -> None:
    """A durable tombstone should prevent auto tags even when no mutation call remains active."""
    thread_root_id = "$removed-during-inference-root:localhost"
    manual_state_key = _thread_tag_state_key(thread_root_id, "manual")
    current_events = {
        manual_state_key: _tag_record_content(set_by="@manual:localhost"),
    }
    manual_client = _tag_writer_client("@manual:localhost", current_events)
    automatic_client = _tag_writer_client("@automatic:localhost", current_events)

    removed_state = await remove_thread_tag(
        manual_client,
        "!room:localhost",
        thread_root_id,
        "manual",
        requester_user_id="@manual:localhost",
    )
    result = await set_thread_tags_if_empty(
        automatic_client,
        "!room:localhost",
        thread_root_id,
        ["auto"],
        set_by="@automatic:localhost",
    )

    assert removed_state.tags == {}
    assert result == SetThreadTagsIfEmptyResult(
        had_existing_tags=False,
        applied_tags=(),
        failed_tags=(),
        skipped_due_to_prior_mutation=True,
    )
    automatic_client.room_put_state.assert_not_awaited()
    assert current_events == {manual_state_key: {}}


@pytest.mark.asyncio
async def test_automatic_batch_serializes_manual_write_until_all_tags_finish() -> None:
    """A manual write must not interleave with an automatic read-and-batch transaction."""
    current_events: dict[str, dict[str, object]] = {}
    automatic_client = _tag_writer_client("@automatic:localhost", current_events)
    manual_client = _tag_writer_client("@manual:localhost", current_events)
    automatic_put_entered = asyncio.Event()
    release_automatic_put = asyncio.Event()
    manual_started = asyncio.Event()
    write_order: list[str] = []
    automatic_task: asyncio.Task[SetThreadTagsIfEmptyResult] | None = None
    manual_task: asyncio.Task[ThreadTagsState] | None = None

    async def automatic_put(**kwargs: object) -> object:
        state_key = kwargs["state_key"]
        assert isinstance(state_key, str)
        tag = json.loads(state_key)[1]
        if tag == "auto-one":
            automatic_put_entered.set()
            await release_automatic_put.wait()
        write_order.append(tag)
        current_events[state_key] = kwargs["content"]
        return nio.RoomPutStateResponse.from_dict(
            {"event_id": f"${tag}"},
            room_id="!room:localhost",
        )

    async def manual_put(**kwargs: object) -> object:
        state_key = kwargs["state_key"]
        assert isinstance(state_key, str)
        tag = json.loads(state_key)[1]
        write_order.append(tag)
        current_events[state_key] = kwargs["content"]
        return nio.RoomPutStateResponse.from_dict(
            {"event_id": f"${tag}"},
            room_id="!room:localhost",
        )

    async def run_manual_write() -> ThreadTagsState:
        manual_started.set()
        return await set_thread_tag(
            manual_client,
            "!room:localhost",
            "$automatic-first-root:localhost",
            "manual",
            set_by="@manual:localhost",
        )

    automatic_client.room_put_state.side_effect = automatic_put
    manual_client.room_put_state.side_effect = manual_put
    try:
        automatic_task = asyncio.create_task(
            set_thread_tags_if_empty(
                automatic_client,
                "!room:localhost",
                "$automatic-first-root:localhost",
                ["auto-one", "auto-two"],
                set_by="@automatic:localhost",
            ),
        )
        await asyncio.wait_for(automatic_put_entered.wait(), timeout=1)

        manual_task = asyncio.create_task(run_manual_write())
        await asyncio.wait_for(manual_started.wait(), timeout=1)
        manual_client.room_get_state_event.assert_not_awaited()
        manual_client.room_put_state.assert_not_awaited()

        release_automatic_put.set()
        await asyncio.gather(automatic_task, manual_task)
    finally:
        release_automatic_put.set()
        pending_tasks = [task for task in (automatic_task, manual_task) if task is not None]
        if pending_tasks:
            await asyncio.gather(*pending_tasks, return_exceptions=True)

    result = automatic_task.result()
    assert result == SetThreadTagsIfEmptyResult(
        had_existing_tags=False,
        applied_tags=("auto-one", "auto-two"),
        failed_tags=(),
    )
    assert write_order == ["auto-one", "auto-two", "manual"]


@pytest.mark.asyncio
async def test_cancelled_automatic_batch_releases_thread_tag_mutation_lock() -> None:
    """Cancellation during the empty-state read must not strand later manual writers."""
    current_events: dict[str, dict[str, object]] = {}
    automatic_client = _tag_writer_client("@automatic:localhost", current_events)
    manual_client = _tag_writer_client("@manual:localhost", current_events)
    automatic_read_entered = asyncio.Event()
    hold_automatic_read = asyncio.Event()

    async def blocked_automatic_read(_room_id: str) -> object:
        automatic_read_entered.set()
        await hold_automatic_read.wait()
        return _thread_tags_room_state_from_current(current_events)

    automatic_client.room_get_state.side_effect = blocked_automatic_read
    automatic_task = asyncio.create_task(
        set_thread_tags_if_empty(
            automatic_client,
            "!room:localhost",
            "$cancelled-auto-root:localhost",
            ["auto"],
            set_by="@automatic:localhost",
        ),
    )
    try:
        await asyncio.wait_for(automatic_read_entered.wait(), timeout=1)
        automatic_task.cancel()
        with pytest.raises(asyncio.CancelledError):
            await automatic_task

        state = await asyncio.wait_for(
            set_thread_tag(
                manual_client,
                "!room:localhost",
                "$cancelled-auto-root:localhost",
                "manual",
                set_by="@manual:localhost",
            ),
            timeout=1,
        )
    finally:
        hold_automatic_read.set()
        if not automatic_task.done():
            automatic_task.cancel()
        await asyncio.gather(automatic_task, return_exceptions=True)

    assert set(state.tags) == {"manual"}


@pytest.mark.asyncio
async def test_automatic_batch_does_not_block_mutations_for_another_thread() -> None:
    """The mutation lock should serialize one thread without blocking unrelated threads."""
    current_events: dict[str, dict[str, object]] = {}
    automatic_client = _tag_writer_client("@automatic:localhost", current_events)
    manual_client = _tag_writer_client("@manual:localhost", current_events)
    automatic_put_entered = asyncio.Event()
    release_automatic_put = asyncio.Event()

    async def blocked_automatic_put(**kwargs: object) -> object:
        automatic_put_entered.set()
        await release_automatic_put.wait()
        current_events[kwargs["state_key"]] = kwargs["content"]
        return nio.RoomPutStateResponse.from_dict(
            {"event_id": "$automatic"},
            room_id="!room:localhost",
        )

    automatic_client.room_put_state.side_effect = blocked_automatic_put
    automatic_task = asyncio.create_task(
        set_thread_tags_if_empty(
            automatic_client,
            "!room:localhost",
            "$blocked-root:localhost",
            ["auto"],
            set_by="@automatic:localhost",
        ),
    )
    try:
        await asyncio.wait_for(automatic_put_entered.wait(), timeout=1)
        manual_state = await asyncio.wait_for(
            set_thread_tag(
                manual_client,
                "!room:localhost",
                "$unrelated-root:localhost",
                "manual",
                set_by="@manual:localhost",
            ),
            timeout=1,
        )
        assert set(manual_state.tags) == {"manual"}
        assert not automatic_task.done()
    finally:
        release_automatic_put.set()
        await asyncio.gather(automatic_task, return_exceptions=True)


@pytest.mark.asyncio
async def test_automatic_batch_continues_after_one_tag_write_fails() -> None:
    """One failed automatic tag should not prevent later tags from being attempted."""
    client = AsyncMock()
    client.room_get_state.return_value = _thread_tags_room_state_response()
    write_tag = AsyncMock(side_effect=[TimeoutError("timed out"), None])

    with (
        patch("mindroom.thread_tags._set_thread_tag_under_lock", new=write_tag),
        patch("mindroom.thread_tags.logger.exception") as log_exception,
    ):
        result = await set_thread_tags_if_empty(
            client,
            "!room:localhost",
            "$partial-auto-root:localhost",
            ["first", "second"],
            set_by="@automatic:localhost",
        )

    assert [call.args[3] for call in write_tag.await_args_list] == ["first", "second"]
    log_exception.assert_called_once_with(
        "Failed to write conditional thread tag",
        room_id="!room:localhost",
        thread_root_id="$partial-auto-root:localhost",
        tag="first",
    )
    assert result == SetThreadTagsIfEmptyResult(
        had_existing_tags=False,
        applied_tags=("second",),
        failed_tags=("first",),
    )


@pytest.mark.asyncio
async def test_set_thread_tag_merges_existing_valid_tags_and_drops_malformed_siblings() -> None:
    """Read-merge-write should preserve valid tags and drop malformed siblings."""
    client = AsyncMock()
    client.user_id = "@mindroom_general:localhost"
    client.joined_members.return_value = _joined_members_response(
        "@mindroom_general:localhost",
        "@alice:localhost",
    )

    current_events: dict[str, dict[str, object]] = {
        _thread_tag_state_key("$thread-root:localhost", "blocked"): _tag_record_content(
            data={"blocked_by": ["  $other:localhost  "]},
        ),
        _thread_tag_state_key("$thread-root:localhost", "waiting"): _tag_record_content(data={"waiting_on": 42}),
        _thread_tag_state_key("$thread-root:localhost", "review"): {
            "set_by": "@user:localhost",
            "set_at": "2026-03-21T19:02:03+00:00",
            "note": 42,
            "data": {},
        },
        _thread_tag_state_key("$thread-root:localhost", "custom"): {
            "set_by": "@user:localhost",
            "set_at": "2026-03-21T19:02:03+00:00",
            "data": [],
        },
        "$thread-root:localhost": _thread_tags_content(
            blocked=_tag_record_content(data={"blocked_by": ["  $other:localhost  "]}),
            **{
                "bad tag!": _tag_record_content(),
                "waiting": _tag_record_content(data={"waiting_on": 42}),
                "review": {
                    "set_by": "@user:localhost",
                    "set_at": "2026-03-21T19:02:03+00:00",
                    "note": 42,
                    "data": {},
                },
                "custom": {
                    "set_by": "@user:localhost",
                    "set_at": "2026-03-21T19:02:03+00:00",
                    "data": [],
                },
            },
        ),
    }

    async def room_get_state_event(**kwargs: object) -> object:
        assert kwargs["event_type"] == "m.room.power_levels"
        return _power_levels_response(
            users={
                "@mindroom_general:localhost": 50,
                "@alice:localhost": 50,
            },
        )

    async def room_get_state(room_id: str) -> object:
        assert room_id == "!room:localhost"
        return _thread_tags_room_state_from_current(current_events)

    async def room_put_state(**kwargs: object) -> object:
        current_events[kwargs["state_key"]] = kwargs["content"]
        return nio.RoomPutStateResponse.from_dict(
            {"event_id": "$state"},
            room_id="!room:localhost",
        )

    client.room_get_state_event.side_effect = room_get_state_event
    client.room_get_state.side_effect = room_get_state
    client.room_put_state.side_effect = room_put_state

    state = await set_thread_tag(
        client,
        "!room:localhost",
        "$thread-root:localhost",
        "priority",
        set_by="@alice:localhost",
        data={"level": "HIGH"},
    )

    assert set(state.tags) == {"blocked", "priority"}
    assert state.tags["blocked"].data == {"blocked_by": ["$other:localhost"]}
    assert state.tags["priority"].data == {"level": "high"}
    _, kwargs = client.room_put_state.await_args
    assert kwargs["state_key"] == _thread_tag_state_key("$thread-root:localhost", "priority")
    assert kwargs["content"]["data"] == {"level": "high"}


@pytest.mark.asyncio
async def test_set_thread_tag_raises_on_write_error() -> None:
    """Write failures should surface as explicit helper errors."""
    client = AsyncMock()
    client.user_id = "@mindroom_general:localhost"
    client.room_get_state_event.side_effect = [
        _power_levels_response(
            users={
                "@mindroom_general:localhost": 50,
                "@alice:localhost": 50,
            },
        ),
        _thread_tags_state_error(message="missing", status_code="M_NOT_FOUND"),
    ]
    client.room_put_state.return_value = object()
    client.joined_members.return_value = _joined_members_response(
        "@mindroom_general:localhost",
        "@alice:localhost",
    )

    with pytest.raises(ThreadTagsError, match="Failed to write thread tags state"):
        await set_thread_tag(
            client,
            "!room:localhost",
            "$thread-root:localhost",
            "resolved",
            set_by="@alice:localhost",
        )


@pytest.mark.asyncio
async def test_set_thread_tag_retries_when_verification_detects_concurrent_overwrite() -> None:
    """A concurrently added sibling tag should survive without forcing a retry."""
    client = AsyncMock()
    client.user_id = "@mindroom_general:localhost"
    client.joined_members.return_value = _joined_members_response(
        "@mindroom_general:localhost",
        "@alice:localhost",
    )

    current_events: dict[str, dict[str, object]] = {}

    async def room_get_state_event(**kwargs: object) -> object:
        assert kwargs["event_type"] == "m.room.power_levels"
        return _power_levels_response(
            users={
                "@mindroom_general:localhost": 50,
                "@alice:localhost": 50,
            },
        )

    write_attempts = 0

    async def room_get_state(room_id: str) -> object:
        assert room_id == "!room:localhost"
        return _thread_tags_room_state_from_current(current_events)

    async def room_put_state(**kwargs: object) -> object:
        nonlocal write_attempts

        write_attempts += 1
        current_events[kwargs["state_key"]] = kwargs["content"]
        if write_attempts == 1:
            current_events[_thread_tag_state_key("$thread-root:localhost", "blocked")] = _tag_record_content(
                data={"blocked_by": ["$other:localhost"]},
            )

        return nio.RoomPutStateResponse.from_dict(
            {"event_id": f"$state-{write_attempts}"},
            room_id="!room:localhost",
        )

    client.room_get_state_event.side_effect = room_get_state_event
    client.room_get_state.side_effect = room_get_state
    client.room_put_state.side_effect = room_put_state

    state = await set_thread_tag(
        client,
        "!room:localhost",
        "$thread-root:localhost",
        "resolved",
        set_by="@alice:localhost",
    )

    assert write_attempts == 1
    assert set(state.tags) == {"blocked", "resolved"}
    assert state.tags["blocked"].data == {"blocked_by": ["$other:localhost"]}
    assert state.tags["resolved"].set_by == "@alice:localhost"
    _, final_kwargs = client.room_put_state.await_args
    assert final_kwargs["state_key"] == _thread_tag_state_key("$thread-root:localhost", "resolved")


@pytest.mark.asyncio
async def test_set_thread_tag_retries_when_verification_detects_same_tag_payload_mismatch() -> None:
    """A same-tag overwrite should retry until the full payload we wrote survives verification."""
    client = AsyncMock()
    client.user_id = "@mindroom_general:localhost"
    client.joined_members.return_value = _joined_members_response(
        "@mindroom_general:localhost",
        "@alice:localhost",
    )

    current_events: dict[str, dict[str, object]] = {}

    async def room_get_state_event(**kwargs: object) -> object:
        assert kwargs["event_type"] == "m.room.power_levels"
        return _power_levels_response(
            users={
                "@mindroom_general:localhost": 50,
                "@alice:localhost": 50,
            },
        )

    write_attempts = 0

    async def room_get_state(room_id: str) -> object:
        assert room_id == "!room:localhost"
        return _thread_tags_room_state_from_current(current_events)

    async def room_put_state(**kwargs: object) -> object:
        nonlocal write_attempts

        write_attempts += 1
        current_events[kwargs["state_key"]] = kwargs["content"]
        if write_attempts == 1:
            current_events[_thread_tag_state_key("$thread-root:localhost", "resolved")] = _tag_record_content(
                set_by="@bob:localhost",
                note="from bob",
                data={"source": "bob"},
            )

        return nio.RoomPutStateResponse.from_dict(
            {"event_id": f"$state-{write_attempts}"},
            room_id="!room:localhost",
        )

    client.room_get_state_event.side_effect = room_get_state_event
    client.room_get_state.side_effect = room_get_state
    client.room_put_state.side_effect = room_put_state

    state = await set_thread_tag(
        client,
        "!room:localhost",
        "$thread-root:localhost",
        "resolved",
        set_by="@alice:localhost",
        note="from alice",
        data={"source": "alice"},
    )

    assert write_attempts == 2
    assert list(state.tags) == ["resolved"]
    assert state.tags["resolved"].set_by == "@alice:localhost"
    assert state.tags["resolved"].note == "from alice"
    assert state.tags["resolved"].data == {"source": "alice"}
    _, final_kwargs = client.room_put_state.await_args
    assert final_kwargs["content"]["set_by"] == "@alice:localhost"
    assert final_kwargs["content"]["note"] == "from alice"
    assert final_kwargs["content"]["data"] == {"source": "alice"}


@pytest.mark.asyncio
async def test_set_thread_tag_retries_when_verification_detects_lost_sibling_tag() -> None:
    """A new-format write should keep a legacy sibling tag without a merge retry."""
    client = AsyncMock()
    client.user_id = "@mindroom_general:localhost"
    client.joined_members.return_value = _joined_members_response(
        "@mindroom_general:localhost",
        "@alice:localhost",
    )

    current_events: dict[str, dict[str, object]] = {
        "$thread-root:localhost": _thread_tags_content(
            blocked=_tag_record_content(note="original sibling"),
        ),
    }

    async def room_get_state_event(**kwargs: object) -> object:
        assert kwargs["event_type"] == "m.room.power_levels"
        return _power_levels_response(
            users={
                "@mindroom_general:localhost": 50,
                "@alice:localhost": 50,
            },
        )

    async def room_get_state(room_id: str) -> object:
        assert room_id == "!room:localhost"
        return _thread_tags_room_state_from_current(current_events)

    async def room_put_state(**kwargs: object) -> object:
        current_events[kwargs["state_key"]] = kwargs["content"]
        return nio.RoomPutStateResponse.from_dict(
            {"event_id": "$state"},
            room_id="!room:localhost",
        )

    client.room_get_state_event.side_effect = room_get_state_event
    client.room_get_state.side_effect = room_get_state
    client.room_put_state.side_effect = room_put_state

    state = await set_thread_tag(
        client,
        "!room:localhost",
        "$thread-root:localhost",
        "resolved",
        set_by="@alice:localhost",
    )

    assert set(state.tags) == {"blocked", "resolved"}
    assert state.tags["blocked"].note == "original sibling"


@pytest.mark.asyncio
async def test_remove_thread_tag_writes_updated_state() -> None:
    """Removing one tag should leave the remaining tag state intact."""
    client = AsyncMock()
    client.user_id = "@mindroom_general:localhost"
    client.room_get_state_event.return_value = _power_levels_response(
        users={
            "@mindroom_general:localhost": 50,
            "@alice:localhost": 50,
        },
    )
    client.room_get_state.side_effect = [
        _thread_tags_room_state_response(
            _thread_tag_state_event("$thread-root:localhost", "resolved"),
            _thread_tag_state_event(
                "$thread-root:localhost",
                "blocked",
                content=_tag_record_content(data={"blocked_by": ["$other:localhost"]}),
            ),
        ),
        _thread_tags_room_state_response(
            _thread_tag_state_event("$thread-root:localhost", "resolved"),
        ),
    ]
    client.room_put_state.return_value = nio.RoomPutStateResponse.from_dict(
        {"event_id": "$state"},
        room_id="!room:localhost",
    )
    client.joined_members.return_value = _joined_members_response(
        "@mindroom_general:localhost",
        "@alice:localhost",
    )

    state = await remove_thread_tag(
        client,
        "!room:localhost",
        "$thread-root:localhost",
        "blocked",
        requester_user_id="@alice:localhost",
    )

    assert list(state.tags) == ["resolved"]
    client.room_put_state.assert_awaited_once()
    _, kwargs = client.room_put_state.await_args
    assert kwargs["room_id"] == "!room:localhost"
    assert kwargs["event_type"] == THREAD_TAGS_EVENT_TYPE
    assert kwargs["state_key"] == _thread_tag_state_key("$thread-root:localhost", "blocked")
    assert kwargs["content"] == {}


@pytest.mark.asyncio
async def test_remove_thread_tag_writes_empty_state_for_last_tag() -> None:
    """Removing the last tag should write an empty content payload."""
    client = AsyncMock()
    client.user_id = "@mindroom_general:localhost"
    client.room_get_state_event.return_value = _power_levels_response(
        users={
            "@mindroom_general:localhost": 50,
            "@alice:localhost": 50,
        },
    )
    client.room_get_state.side_effect = [
        _thread_tags_room_state_response(
            _thread_tag_state_event("$thread-root:localhost", "resolved"),
        ),
        _thread_tags_room_state_response(),
    ]
    client.room_put_state.return_value = nio.RoomPutStateResponse.from_dict(
        {"event_id": "$state"},
        room_id="!room:localhost",
    )
    client.joined_members.return_value = _joined_members_response(
        "@mindroom_general:localhost",
        "@alice:localhost",
    )

    state = await remove_thread_tag(
        client,
        "!room:localhost",
        "$thread-root:localhost",
        "resolved",
        requester_user_id="@alice:localhost",
    )

    assert state.tags == {}
    client.room_put_state.assert_awaited_once_with(
        room_id="!room:localhost",
        event_type=THREAD_TAGS_EVENT_TYPE,
        content={},
        state_key=_thread_tag_state_key("$thread-root:localhost", "resolved"),
    )


@pytest.mark.asyncio
async def test_remove_thread_tag_rejects_missing_existing_state() -> None:
    """Removing should fail instead of creating a tombstone for a missing state event."""
    client = AsyncMock()
    client.user_id = "@mindroom_general:localhost"
    client.room_get_state_event.return_value = _power_levels_response(
        users={
            "@mindroom_general:localhost": 50,
            "@alice:localhost": 50,
        },
    )
    client.room_get_state.return_value = _thread_tags_room_state_response()
    client.joined_members.return_value = _joined_members_response(
        "@mindroom_general:localhost",
        "@alice:localhost",
    )

    with pytest.raises(ThreadTagsError, match="No thread tags state exists"):
        await remove_thread_tag(
            client,
            "!room:localhost",
            "$thread-root:localhost",
            "resolved",
            requester_user_id="@alice:localhost",
        )

    client.room_put_state.assert_not_awaited()


@pytest.mark.asyncio
async def test_remove_thread_tag_retries_when_verification_detects_concurrent_restore() -> None:
    """A concurrent stale write that restores the removed tag should trigger one retry."""
    client = AsyncMock()
    client.user_id = "@mindroom_general:localhost"
    client.joined_members.return_value = _joined_members_response(
        "@mindroom_general:localhost",
        "@alice:localhost",
    )

    current_events: dict[str, dict[str, object]] = {
        _thread_tag_state_key("$thread-root:localhost", "resolved"): _tag_record_content(),
    }

    async def room_get_state_event(**kwargs: object) -> object:
        assert kwargs["event_type"] == "m.room.power_levels"
        return _power_levels_response(
            users={
                "@mindroom_general:localhost": 50,
                "@alice:localhost": 50,
            },
        )

    write_attempts = 0

    async def room_get_state(room_id: str) -> object:
        assert room_id == "!room:localhost"
        return _thread_tags_room_state_from_current(current_events)

    async def room_put_state(**kwargs: object) -> object:
        nonlocal write_attempts

        write_attempts += 1
        current_events[kwargs["state_key"]] = kwargs["content"]
        if write_attempts == 1:
            current_events[_thread_tag_state_key("$thread-root:localhost", "resolved")] = _tag_record_content()
            current_events[_thread_tag_state_key("$thread-root:localhost", "blocked")] = _tag_record_content(
                data={"blocked_by": ["$other:localhost"]},
            )

        return nio.RoomPutStateResponse.from_dict(
            {"event_id": f"$state-{write_attempts}"},
            room_id="!room:localhost",
        )

    client.room_get_state_event.side_effect = room_get_state_event
    client.room_get_state.side_effect = room_get_state
    client.room_put_state.side_effect = room_put_state

    state = await remove_thread_tag(
        client,
        "!room:localhost",
        "$thread-root:localhost",
        "resolved",
        requester_user_id="@alice:localhost",
    )

    assert write_attempts == 2
    assert list(state.tags) == ["blocked"]
    assert state.tags["blocked"].data == {"blocked_by": ["$other:localhost"]}
    _, final_kwargs = client.room_put_state.await_args
    assert final_kwargs["content"] == {}


@pytest.mark.asyncio
async def test_remove_thread_tag_retries_when_verification_detects_sibling_payload_change() -> None:
    """A concurrently added sibling tag should survive one remove write without a retry."""
    client = AsyncMock()
    client.user_id = "@mindroom_general:localhost"
    client.room_get_state_event.return_value = _power_levels_response(
        users={
            "@mindroom_general:localhost": 50,
            "@alice:localhost": 50,
        },
    )
    client.joined_members.return_value = _joined_members_response(
        "@mindroom_general:localhost",
        "@alice:localhost",
    )

    current_events: dict[str, dict[str, object]] = {
        _thread_tag_state_key("$thread-root:localhost", "resolved"): _tag_record_content(),
    }
    write_attempts = 0

    async def room_get_state(room_id: str) -> object:
        assert room_id == "!room:localhost"
        return _thread_tags_room_state_from_current(current_events)

    async def room_put_state(**kwargs: object) -> object:
        nonlocal write_attempts

        write_attempts += 1
        current_events[kwargs["state_key"]] = kwargs["content"]
        current_events[_thread_tag_state_key("$thread-root:localhost", "blocked")] = _tag_record_content(
            note="added concurrently",
            data={"blocked_by": ["$other:localhost"]},
        )
        return nio.RoomPutStateResponse.from_dict(
            {"event_id": "$state"},
            room_id="!room:localhost",
        )

    client.room_get_state.side_effect = room_get_state
    client.room_put_state.side_effect = room_put_state

    state = await remove_thread_tag(
        client,
        "!room:localhost",
        "$thread-root:localhost",
        "resolved",
        requester_user_id="@alice:localhost",
    )

    assert list(state.tags) == ["blocked"]
    assert state.tags["blocked"].note == "added concurrently"
    assert state.tags["blocked"].data == {"blocked_by": ["$other:localhost"]}
    assert write_attempts == 1
    client.room_put_state.assert_awaited_once()


@pytest.mark.asyncio
async def test_remove_thread_tag_accepts_empty_state_after_concurrent_last_sibling_remove() -> None:
    """A post-write empty reread should be accepted when another actor removed the last sibling."""
    client = AsyncMock()
    client.user_id = "@mindroom_general:localhost"
    client.room_get_state_event.return_value = _power_levels_response(
        users={
            "@mindroom_general:localhost": 50,
            "@alice:localhost": 50,
        },
    )
    client.room_get_state.side_effect = [
        _thread_tags_room_state_response(
            _thread_tag_state_event("$thread-root:localhost", "resolved"),
            _thread_tag_state_event(
                "$thread-root:localhost",
                "blocked",
                content=_tag_record_content(data={"blocked_by": ["$other:localhost"]}),
            ),
        ),
        _thread_tags_room_state_response(),
    ]
    client.room_put_state.return_value = nio.RoomPutStateResponse.from_dict(
        {"event_id": "$state"},
        room_id="!room:localhost",
    )
    client.joined_members.return_value = _joined_members_response(
        "@mindroom_general:localhost",
        "@alice:localhost",
    )

    state = await remove_thread_tag(
        client,
        "!room:localhost",
        "$thread-root:localhost",
        "resolved",
        requester_user_id="@alice:localhost",
    )

    assert state.tags == {}
    client.room_put_state.assert_awaited_once()


@pytest.mark.asyncio
async def test_remove_thread_tag_rejects_missing_tag() -> None:
    """Removing an absent tag should be an explicit error."""
    client = AsyncMock()
    client.user_id = "@mindroom_general:localhost"
    client.room_get_state_event.return_value = _power_levels_response(
        users={
            "@mindroom_general:localhost": 50,
            "@alice:localhost": 50,
        },
    )
    client.room_get_state.return_value = _thread_tags_room_state_response(
        _thread_tag_state_event("$thread-root:localhost", "resolved"),
    )
    client.joined_members.return_value = _joined_members_response(
        "@mindroom_general:localhost",
        "@alice:localhost",
    )

    with pytest.raises(ThreadTagsError, match="is not set"):
        await remove_thread_tag(
            client,
            "!room:localhost",
            "$thread-root:localhost",
            "blocked",
            requester_user_id="@alice:localhost",
        )

    client.room_put_state.assert_not_awaited()


@pytest.mark.asyncio
async def test_get_thread_tags_parses_valid_state() -> None:
    """A valid room-state payload should return parsed tags for the thread."""
    client = AsyncMock()
    client.room_get_state.return_value = _thread_tags_room_state_response(
        _thread_tag_state_event(
            "$thread-root:localhost",
            "resolved",
            content=_tag_record_content(note="done"),
        ),
        _thread_tag_state_event(
            "$thread-root:localhost",
            "blocked",
            content=_tag_record_content(data={"blocked_by": ["$other:localhost"]}),
        ),
    )

    state = await get_thread_tags(
        client,
        "!room:localhost",
        "$thread-root:localhost",
    )

    assert state is not None
    assert state.thread_root_id == "$thread-root:localhost"
    assert set(state.tags) == {"resolved", "blocked"}
    assert state.tags["resolved"].note == "done"
    assert state.tags["blocked"].data == {"blocked_by": ["$other:localhost"]}


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "response",
    [
        _thread_tags_room_state_response(),
        _thread_tags_room_state_response(
            {
                "type": THREAD_TAGS_EVENT_TYPE,
                "state_key": _thread_tag_state_key("$thread-root:localhost", "resolved"),
                "content": {},
            },
        ),
        _thread_tags_room_state_response(
            _legacy_thread_tags_event("$thread-root:localhost", content={}),
        ),
        _thread_tags_room_state_response(
            _legacy_thread_tags_event("$thread-root:localhost", content={"tags": {}}),
        ),
        _thread_tags_room_state_response(
            _legacy_thread_tags_event("$thread-root:localhost", content={"tags": "invalid"}),
        ),
        _thread_tags_room_state_response(
            _thread_tag_state_event(
                "$thread-root:localhost",
                "resolved",
                content={"set_by": "@user:localhost", "set_at": "bad", "data": {}},
            ),
        ),
    ],
)
async def test_get_thread_tags_returns_none_for_missing_empty_and_malformed(response: object) -> None:
    """Missing, empty, or fully malformed state should be treated as untagged."""
    client = AsyncMock()
    client.room_get_state.return_value = response

    state = await get_thread_tags(
        client,
        "!room:localhost",
        "$thread-root:localhost",
    )

    assert state is None


@pytest.mark.asyncio
async def test_get_thread_tags_drops_malformed_tags_and_preserves_valid_siblings() -> None:
    """Malformed tags should be ignored without discarding valid siblings."""
    client = AsyncMock()
    client.room_get_state.return_value = _thread_tags_room_state_response(
        _thread_tag_state_event("$thread-root:localhost", "resolved"),
        _thread_tag_state_event(
            "$thread-root:localhost",
            "blocked",
            content=_tag_record_content(data={"blocked_by": "$not-a-list"}),
        ),
        _thread_tag_state_event(
            "$thread-root:localhost",
            "due",
            content=_tag_record_content(data={"deadline": "not-a-date"}),
        ),
        _thread_tag_state_event(
            "$thread-root:localhost",
            "review",
            content={
                "set_by": "@user:localhost",
                "set_at": "2026-03-21T19:02:03+00:00",
                "note": 42,
                "data": {},
            },
        ),
        _thread_tag_state_event(
            "$thread-root:localhost",
            "custom",
            content={
                "set_by": "@user:localhost",
                "set_at": "2026-03-21T19:02:03+00:00",
                "data": [],
            },
        ),
    )

    state = await get_thread_tags(
        client,
        "!room:localhost",
        "$thread-root:localhost",
    )

    assert state is not None
    assert list(state.tags) == ["resolved"]


@pytest.mark.asyncio
async def test_get_thread_tags_ignores_malformed_per_tag_overlay_and_keeps_legacy_tag() -> None:
    """A malformed per-tag overlay must not hide a valid legacy tag during migration."""
    client = AsyncMock()
    client.room_get_state.return_value = _thread_tags_room_state_response(
        _legacy_thread_tags_event(
            "$thread-root:localhost",
            content=_thread_tags_content(
                resolved=_tag_record_content(note="legacy tag"),
            ),
        ),
        _thread_tag_state_event(
            "$thread-root:localhost",
            "resolved",
            content={"set_by": "@user:localhost", "set_at": "bad", "data": {}},
        ),
    )

    state = await get_thread_tags(
        client,
        "!room:localhost",
        "$thread-root:localhost",
    )

    assert state is not None
    assert list(state.tags) == ["resolved"]
    assert state.tags["resolved"].note == "legacy tag"


@pytest.mark.asyncio
async def test_list_tagged_threads_filters_non_matching_events_and_supports_tag_filter() -> None:
    """Room-wide listing should keep only valid tag state and support tag filtering."""
    client = AsyncMock()
    client.room_get_state.return_value = _thread_tags_room_state_response(
        _thread_tag_state_event("$thread-one:localhost", "resolved"),
        _thread_tag_state_event(
            "$thread-two:localhost",
            "blocked",
            content=_tag_record_content(data={"blocked_by": ["$other:localhost"]}),
        ),
        _thread_tag_state_event(
            "$thread-three:localhost",
            "blocked",
            content=_tag_record_content(data={"blocked_by": "$bad"}),
        ),
        {
            "type": THREAD_TAGS_EVENT_TYPE,
            "state_key": _thread_tag_state_key("$thread-four:localhost", "resolved"),
            "content": {},
        },
        _legacy_thread_tags_event(
            "$thread-six:localhost",
            content={
                "tags": {
                    "custom": {
                        "set_by": "@user:localhost",
                        "set_at": "2026-03-21T19:02:03+00:00",
                        "data": [],
                    },
                },
            },
        ),
        _legacy_thread_tags_event(
            "$thread-seven:localhost",
            content={
                "tags": {
                    "review": {
                        "set_by": "@user:localhost",
                        "set_at": "2026-03-21T19:02:03+00:00",
                        "note": 42,
                        "data": {},
                    },
                },
            },
        ),
        {
            "type": "com.mindroom.other",
            "state_key": "$thread-five:localhost",
            "content": _thread_tags_content(resolved=_tag_record_content()),
        },
    )

    all_threads = await list_tagged_threads(client, "!room:localhost")
    resolved_threads = await list_tagged_threads(client, "!room:localhost", tag="resolved")

    assert all_threads.include_untagged is False
    assert all_threads.truncated is False
    assert set(all_threads.tag_state) == {"$thread-one:localhost", "$thread-two:localhost"}
    assert set(resolved_threads.tag_state) == {"$thread-one:localhost"}


@pytest.mark.asyncio
async def test_list_tagged_threads_default_does_not_enumerate_room_threads() -> None:
    """Default room-wide listing should not call the /threads enumeration helper."""
    client = AsyncMock()
    client.room_get_state.return_value = _thread_tags_room_state_response(
        _thread_tag_state_event("$thread-one:localhost", "resolved"),
    )

    with patch("mindroom.thread_tags.enumerate_room_thread_root_ids", new=AsyncMock()) as mock_enumerate:
        listing = await list_tagged_threads(client, "!room:localhost", include_untagged=False)

    assert list(listing.tag_state) == ["$thread-one:localhost"]
    assert listing.include_untagged is False
    assert listing.truncated is False
    mock_enumerate.assert_not_awaited()


@pytest.mark.asyncio
async def test_list_tagged_threads_include_untagged_unions_tag_state_with_room_threads() -> None:
    """Including untagged threads should preserve /threads order and synthesize empty tags."""
    client = AsyncMock()
    client.room_get_state.return_value = _thread_tags_room_state_response(
        _thread_tag_state_event("$thread-two:localhost", "resolved"),
        _thread_tag_state_event("$thread-four:localhost", "blocked"),
    )

    with patch(
        "mindroom.thread_tags.enumerate_room_thread_root_ids",
        new=AsyncMock(
            return_value=(
                [
                    "$thread-one:localhost",
                    "$thread-two:localhost",
                    "$thread-three:localhost",
                    "$thread-four:localhost",
                    "$thread-five:localhost",
                ],
                False,
            ),
        ),
    ):
        listing = await list_tagged_threads(client, "!room:localhost", include_untagged=True)

    assert isinstance(listing, ThreadTagsListing)
    assert list(listing.tag_state) == [
        "$thread-one:localhost",
        "$thread-two:localhost",
        "$thread-three:localhost",
        "$thread-four:localhost",
        "$thread-five:localhost",
    ]
    assert listing.tag_state["$thread-one:localhost"].tags == {}
    assert listing.tag_state["$thread-three:localhost"].tags == {}
    assert listing.tag_state["$thread-five:localhost"].tags == {}
    assert set(listing.tag_state["$thread-two:localhost"].tags) == {"resolved"}
    assert set(listing.tag_state["$thread-four:localhost"].tags) == {"blocked"}
    assert listing.truncated is False


@pytest.mark.asyncio
async def test_list_tagged_threads_include_untagged_appends_tagged_state_only_threads() -> None:
    """Tagged state for roots absent from /threads should be preserved at the end."""
    client = AsyncMock()
    client.room_get_state.return_value = _thread_tags_room_state_response(
        _thread_tag_state_event("$thread-two:localhost", "resolved"),
        _thread_tag_state_event("$thread-x:localhost", "blocked"),
    )

    with patch(
        "mindroom.thread_tags.enumerate_room_thread_root_ids",
        new=AsyncMock(return_value=(["$thread-one:localhost", "$thread-two:localhost"], False)),
    ):
        listing = await list_tagged_threads(client, "!room:localhost", include_untagged=True)

    assert isinstance(listing, ThreadTagsListing)
    assert list(listing.tag_state) == ["$thread-one:localhost", "$thread-two:localhost", "$thread-x:localhost"]
    assert listing.tag_state["$thread-one:localhost"].tags == {}
    assert set(listing.tag_state["$thread-x:localhost"].tags) == {"blocked"}


@pytest.mark.asyncio
async def test_list_tagged_threads_include_untagged_filters_tagged_state_only_threads() -> None:
    """Tagged-state-only roots should obey filters while keeping trailing order."""
    client = AsyncMock()
    client.room_get_state.return_value = _thread_tags_room_state_response(
        _thread_tag_state_event("$thread-live:localhost", "blocked"),
        _thread_tag_state_event("$thread-orphan:localhost", "resolved"),
    )

    with patch(
        "mindroom.thread_tags.enumerate_room_thread_root_ids",
        new=AsyncMock(return_value=(["$thread-live:localhost"], False)),
    ):
        unresolved_listing = await list_tagged_threads(
            client,
            "!room:localhost",
            exclude_tag="resolved",
            include_untagged=True,
        )
        other_listing = await list_tagged_threads(
            client,
            "!room:localhost",
            exclude_tag="other",
            include_untagged=True,
        )

    assert list(unresolved_listing.tag_state) == ["$thread-live:localhost"]
    assert list(other_listing.tag_state) == ["$thread-live:localhost", "$thread-orphan:localhost"]


@pytest.mark.asyncio
async def test_enumerate_room_thread_root_ids_fetches_all_pages() -> None:
    """Room thread enumeration should follow pagination to completion."""
    client = AsyncMock()

    with patch(
        "mindroom.matrix.client_thread_history.get_room_threads_page",
        new=AsyncMock(
            side_effect=[
                ([_thread_root_event("$thread-one:localhost")], "A"),
                ([_thread_root_event("$thread-two:localhost")], "B"),
                ([_thread_root_event("$thread-three:localhost")], None),
            ],
        ),
    ) as mock_get_page:
        thread_root_ids, truncated = await enumerate_room_thread_root_ids(client, "!room:localhost")

    assert thread_root_ids == ["$thread-one:localhost", "$thread-two:localhost", "$thread-three:localhost"]
    assert truncated is False
    assert mock_get_page.await_args_list[0].kwargs["page_token"] is None
    assert mock_get_page.await_args_list[1].kwargs["page_token"] == "A"  # noqa: S105
    assert mock_get_page.await_args_list[2].kwargs["page_token"] == "B"  # noqa: S105


@pytest.mark.asyncio
async def test_enumerate_room_thread_root_ids_skips_blank_event_ids() -> None:
    """Blank thread-root event IDs should not leak into room-wide listings."""
    client = AsyncMock()

    with patch(
        "mindroom.matrix.client_thread_history.get_room_threads_page",
        new=AsyncMock(
            return_value=(
                [_thread_root_event(""), _thread_root_event("$root:localhost")],
                None,
            ),
        ),
    ):
        thread_root_ids, truncated = await enumerate_room_thread_root_ids(client, "!room:localhost")

    assert thread_root_ids == ["$root:localhost"]
    assert truncated is False


@pytest.mark.asyncio
async def test_enumerate_room_thread_root_ids_truncates_empty_page_with_next_token() -> None:
    """Room thread enumeration should stop on ambiguous empty continuation pages."""
    client = AsyncMock()

    with patch(
        "mindroom.matrix.client_thread_history.get_room_threads_page",
        new=AsyncMock(
            side_effect=[
                ([], "next_token_1"),
                ([_thread_root_event("$root_A"), _thread_root_event("$root_B")], None),
            ],
        ),
    ) as mock_get_page:
        thread_root_ids, truncated = await enumerate_room_thread_root_ids(client, "!room:localhost")

    assert thread_root_ids == []
    assert truncated is True
    assert mock_get_page.await_args_list[0].kwargs["page_token"] is None
    assert mock_get_page.await_count == 1


@pytest.mark.asyncio
async def test_enumerate_room_thread_root_ids_does_not_follow_consecutive_empty_pages() -> None:
    """Empty continuation pages should not burn requests until the hard page cap."""
    client = AsyncMock()

    with patch(
        "mindroom.matrix.client_thread_history.get_room_threads_page",
        new=AsyncMock(
            side_effect=[
                ([], "empty-1"),
                ([], "empty-2"),
                ([], "empty-3"),
                ([_thread_root_event("$root_A"), _thread_root_event("$root_B")], None),
            ],
        ),
    ) as mock_get_page:
        thread_root_ids, truncated = await enumerate_room_thread_root_ids(client, "!room:localhost")

    assert thread_root_ids == []
    assert truncated is True
    assert mock_get_page.await_args_list[0].kwargs["page_token"] is None
    assert mock_get_page.await_count == 1


@pytest.mark.asyncio
async def test_enumerate_room_thread_root_ids_truncates_fresh_empty_token_loop() -> None:
    """Room thread enumeration should bound fresh-token empty-page loops."""
    client = AsyncMock()
    page_count = 0

    async def get_empty_page(*_args: object, **_kwargs: object) -> tuple[list[object], str]:
        nonlocal page_count
        page_count += 1
        assert page_count <= 110
        return [], f"token-{page_count}"

    with patch("mindroom.matrix.client_thread_history.get_room_threads_page", new=get_empty_page):
        thread_root_ids, truncated = await enumerate_room_thread_root_ids(client, "!room:localhost")

    assert thread_root_ids == []
    assert truncated is True
    assert page_count == 1


@pytest.mark.asyncio
async def test_enumerate_room_thread_root_ids_exact_cap_on_final_page_is_not_truncated() -> None:
    """Exact cap completion should not report truncation without a continuation token."""
    client = AsyncMock()

    with patch(
        "mindroom.matrix.client_thread_history.get_room_threads_page",
        new=AsyncMock(
            side_effect=[
                ([_thread_root_event("$thread-one:localhost"), _thread_root_event("$thread-two:localhost")], "A"),
                ([_thread_root_event("$thread-three:localhost"), _thread_root_event("$thread-four:localhost")], None),
            ],
        ),
    ):
        thread_root_ids, truncated = await enumerate_room_thread_root_ids(
            client,
            "!room:localhost",
            max_thread_roots=4,
            page_size=2,
        )

    assert thread_root_ids == [
        "$thread-one:localhost",
        "$thread-two:localhost",
        "$thread-three:localhost",
        "$thread-four:localhost",
    ]
    assert truncated is False


@pytest.mark.asyncio
async def test_enumerate_room_thread_root_ids_truncates_at_cap() -> None:
    """Room thread enumeration should stop at the hard root cap."""
    client = AsyncMock()
    pages = []
    for page_index in range(25):
        roots = [_thread_root_event(f"$thread-{page_index * 100 + root_index}:localhost") for root_index in range(100)]
        pages.append((roots, f"token-{page_index}"))

    with patch(
        "mindroom.matrix.client_thread_history.get_room_threads_page",
        new=AsyncMock(side_effect=pages),
    ) as mock_get_page:
        thread_root_ids, truncated = await enumerate_room_thread_root_ids(client, "!room:localhost")

    assert len(thread_root_ids) == 2000
    assert thread_root_ids[0] == "$thread-0:localhost"
    assert thread_root_ids[-1] == "$thread-1999:localhost"
    assert truncated is True
    assert mock_get_page.await_count == 20


@pytest.mark.asyncio
async def test_enumerate_room_thread_root_ids_repeated_token_guard_truncates() -> None:
    """Room thread enumeration should stop when the homeserver repeats a cursor."""
    client = AsyncMock()

    with patch(
        "mindroom.matrix.client_thread_history.get_room_threads_page",
        new=AsyncMock(
            side_effect=[
                ([_thread_root_event("$thread-one:localhost")], "A"),
                ([_thread_root_event("$thread-two:localhost")], "A"),
            ],
        ),
    ):
        thread_root_ids, truncated = await enumerate_room_thread_root_ids(client, "!room:localhost")

    assert thread_root_ids == ["$thread-one:localhost", "$thread-two:localhost"]
    assert truncated is True


@pytest.mark.asyncio
async def test_enumerate_room_thread_root_ids_zero_new_roots_guard_truncates() -> None:
    """Room thread enumeration should stop when a non-empty page only repeats roots."""
    client = AsyncMock()

    with patch(
        "mindroom.matrix.client_thread_history.get_room_threads_page",
        new=AsyncMock(
            side_effect=[
                ([_thread_root_event("$thread-one:localhost")], "A"),
                ([_thread_root_event("$thread-one:localhost")], "B"),
            ],
        ),
    ):
        thread_root_ids, truncated = await enumerate_room_thread_root_ids(client, "!room:localhost")

    assert thread_root_ids == ["$thread-one:localhost"]
    assert truncated is True


@pytest.mark.asyncio
async def test_enumerate_room_thread_root_ids_duplicate_final_page_completes() -> None:
    """Duplicate roots on the final page should honor completed pagination."""
    client = AsyncMock()

    with patch(
        "mindroom.matrix.client_thread_history.get_room_threads_page",
        new=AsyncMock(
            side_effect=[
                ([_thread_root_event("$thread-one:localhost"), _thread_root_event("$thread-two:localhost")], "A"),
                ([_thread_root_event("$thread-one:localhost"), _thread_root_event("$thread-two:localhost")], None),
            ],
        ),
    ):
        thread_root_ids, truncated = await enumerate_room_thread_root_ids(client, "!room:localhost")

    assert thread_root_ids == ["$thread-one:localhost", "$thread-two:localhost"]
    assert truncated is False


@pytest.mark.asyncio
async def test_list_tagged_threads_include_untagged_propagates_room_threads_page_error() -> None:
    """Room-wide include-untagged listing should preserve /threads error details."""
    client = AsyncMock()
    client.room_get_state.return_value = _thread_tags_room_state_response()
    error = RoomThreadsPageError(response="rate limited", errcode="M_LIMIT_EXCEEDED", retry_after_ms=250)

    with (
        patch(
            "mindroom.matrix.client_thread_history.get_room_threads_page",
            new=AsyncMock(side_effect=error),
        ),
        pytest.raises(RoomThreadsPageError) as exc_info,
    ):
        await list_tagged_threads(client, "!room:localhost", include_untagged=True)

    assert exc_info.value.response == "rate limited"
    assert exc_info.value.errcode == "M_LIMIT_EXCEEDED"
    assert exc_info.value.retry_after_ms == 250


@pytest.mark.asyncio
async def test_list_tagged_threads_include_untagged_filter_semantics_on_synthesized_entries() -> None:
    """Tag and include_tag should exclude untagged entries while exclude_tag keeps them."""
    client = AsyncMock()
    client.room_get_state.return_value = _thread_tags_room_state_response(
        _thread_tag_state_event("$resolved-thread:localhost", "resolved"),
    )

    mock_enumerate = AsyncMock(
        return_value=(["$untagged-thread:localhost", "$resolved-thread:localhost"], True),
    )
    with patch(
        "mindroom.thread_tags.enumerate_room_thread_root_ids",
        new=mock_enumerate,
    ):
        unresolved_listing = await list_tagged_threads(
            client,
            "!room:localhost",
            exclude_tag="resolved",
            include_untagged=True,
        )
        include_listing = await list_tagged_threads(
            client,
            "!room:localhost",
            include_tag="resolved",
            include_untagged=True,
        )
        tag_listing = await list_tagged_threads(
            client,
            "!room:localhost",
            tag="resolved",
            include_untagged=True,
        )

    assert isinstance(unresolved_listing, ThreadTagsListing)
    assert isinstance(include_listing, ThreadTagsListing)
    assert isinstance(tag_listing, ThreadTagsListing)
    assert list(unresolved_listing.tag_state) == ["$untagged-thread:localhost"]
    assert list(include_listing.tag_state) == ["$resolved-thread:localhost"]
    assert list(tag_listing.tag_state) == ["$resolved-thread:localhost"]
    assert unresolved_listing.truncated is True
    assert include_listing.truncated is False
    assert tag_listing.truncated is False
    assert mock_enumerate.await_count == 1


@pytest.mark.asyncio
async def test_list_tagged_threads_raises_for_room_state_fetch_errors() -> None:
    """Room-wide listing should surface Matrix failures instead of faking an empty result."""
    client = AsyncMock()
    client.room_get_state.return_value = object()

    with pytest.raises(ThreadTagsError, match="Failed to fetch room state for thread tags"):
        await list_tagged_threads(client, "!room:localhost")


@pytest.mark.asyncio
@pytest.mark.parametrize("tag", ["", " ", "Bad Tag", "needs_underscore", "a" * 51])
async def test_set_thread_tag_rejects_invalid_tag_names(tag: str) -> None:
    """Tag names should follow the stable slug format before any Matrix I/O."""
    client = AsyncMock()

    with pytest.raises(ThreadTagsError, match="tag"):
        await set_thread_tag(
            client,
            "!room:localhost",
            "$thread-root:localhost",
            tag,
            set_by="@alice:localhost",
        )

    client.room_get_state_event.assert_not_awaited()


@pytest.mark.parametrize(
    ("raw_tag", "expected"),
    [
        ("bug", "bug"),
        ("  Follow Up  ", "follow-up"),
        ("Feature\tRequest", "feature-request"),
        ("needs_underscore", "needs-underscore"),
        ("Beta -- Release!", "beta-release"),
        ("-leading-and-trailing-", "leading-and-trailing"),
        ("x" * (COERCED_TAG_MAX_LENGTH + 10), "x" * COERCED_TAG_MAX_LENGTH),
    ],
)
def test_coerce_tag_name_normalizes_free_form_input(raw_tag: str, expected: str) -> None:
    """Coercion lowercases, hyphenates separators, and truncates to the cap."""
    assert coerce_tag_name(raw_tag) == expected


@pytest.mark.parametrize("raw_tag", ["", "   ", "!!!", "---", None, 42])
def test_coerce_tag_name_returns_none_when_nothing_survives(raw_tag: object) -> None:
    """Input with no valid characters coerces to None instead of raising."""
    assert coerce_tag_name(raw_tag) is None


def test_coerce_tag_name_strips_hyphen_left_by_truncation() -> None:
    """A truncation cut landing on a separator must not leave a trailing hyphen."""
    raw_tag = "x" * (COERCED_TAG_MAX_LENGTH - 1) + " tail"

    assert coerce_tag_name(raw_tag) == "x" * (COERCED_TAG_MAX_LENGTH - 1)


@pytest.mark.asyncio
async def test_set_thread_tag_normalizes_supported_predefined_payloads() -> None:
    """Predefined tag schemas should normalize accepted structured data."""
    client = AsyncMock()
    client.user_id = "@mindroom_general:localhost"
    client.joined_members.return_value = _joined_members_response(
        "@mindroom_general:localhost",
        "@alice:localhost",
    )

    current_events: dict[str, dict[str, object]] = {}

    async def room_get_state_event(**kwargs: object) -> object:
        assert kwargs["event_type"] == "m.room.power_levels"
        return _power_levels_response(
            users={
                "@mindroom_general:localhost": 50,
                "@alice:localhost": 50,
            },
        )

    async def room_get_state(room_id: str) -> object:
        assert room_id == "!room:localhost"
        return _thread_tags_room_state_from_current(current_events)

    async def room_put_state(**kwargs: object) -> object:
        current_events[kwargs["state_key"]] = kwargs["content"]
        return nio.RoomPutStateResponse.from_dict(
            {"event_id": "$state"},
            room_id="!room:localhost",
        )

    client.room_get_state_event.side_effect = room_get_state_event
    client.room_get_state.side_effect = room_get_state
    client.room_put_state.side_effect = room_put_state

    state = await set_thread_tag(
        client,
        "!room:localhost",
        "$thread-root:localhost",
        "due",
        set_by="@alice:localhost",
        data={"deadline": "2026-03-26T22:00:00Z"},
    )

    assert state.tags["due"].data == {"deadline": "2026-03-26T22:00:00+00:00"}


@pytest.mark.asyncio
@pytest.mark.parametrize(
    ("tag", "data", "message"),
    [
        ("blocked", {"blocked_by": "$other:localhost"}, "blocked.data.blocked_by"),
        ("waiting", {"waiting_on": 42}, "waiting.data.waiting_on"),
        ("priority", {"level": "urgent"}, "priority.data.level"),
        ("due", {"deadline": "not-a-date"}, "due.data.deadline"),
    ],
)
async def test_set_thread_tag_rejects_invalid_predefined_payloads(
    tag: str,
    data: dict[str, object],
    message: str,
) -> None:
    """Predefined tag data should fail fast on invalid shapes."""
    client = AsyncMock()

    with pytest.raises(ThreadTagsError, match=message):
        await set_thread_tag(
            client,
            "!room:localhost",
            "$thread-root:localhost",
            tag,
            set_by="@alice:localhost",
            data=data,
        )

    client.room_get_state_event.assert_not_awaited()


@pytest.mark.asyncio
async def test_set_thread_tag_rejects_non_json_compatible_custom_data() -> None:
    """Custom tag data should fail fast when nested values are not JSON-compatible."""
    client = AsyncMock()

    with pytest.raises(ThreadTagsError, match="JSON-compatible"):
        await set_thread_tag(
            client,
            "!room:localhost",
            "$thread-root:localhost",
            "custom",
            set_by="@alice:localhost",
            data={"bad": object()},
        )

    client.room_get_state_event.assert_not_awaited()


@pytest.mark.asyncio
@pytest.mark.parametrize("value", [math.nan, math.inf, -math.inf])
async def test_set_thread_tag_rejects_non_finite_float_custom_data(value: float) -> None:
    """Custom tag data should reject NaN and infinity before any Matrix I/O."""
    client = AsyncMock()

    with pytest.raises(ThreadTagsError, match="finite JSON-compatible numbers"):
        await set_thread_tag(
            client,
            "!room:localhost",
            "$thread-root:localhost",
            "custom",
            set_by="@alice:localhost",
            data={"bad": value},
        )

    client.room_get_state_event.assert_not_awaited()


@pytest.mark.asyncio
@pytest.mark.parametrize("action", ["set", "remove"])
async def test_thread_tags_write_rejects_insufficient_power_level(action: str) -> None:
    """State writes should fail before sending when the Matrix account lacks power."""
    client = AsyncMock()
    client.user_id = "@mindroom_general:localhost"
    client.room_get_state_event.return_value = _power_levels_response(
        users={
            "@mindroom_general:localhost": 0,
            "@alice:localhost": 50,
        },
    )

    with pytest.raises(ThreadTagsError, match="Insufficient Matrix power level"):
        await _write_operation(client, action)

    client.room_put_state.assert_not_awaited()


@pytest.mark.asyncio
@pytest.mark.parametrize("action", ["set", "remove"])
async def test_thread_tags_write_rejects_low_power_requester(action: str) -> None:
    """State writes should fail when the requester lacks the required room power."""
    client = AsyncMock()
    client.user_id = "@mindroom_general:localhost"
    client.room_get_state_event.return_value = _power_levels_response(
        users={
            "@mindroom_general:localhost": 50,
            "@alice:localhost": 0,
        },
    )
    client.joined_members.return_value = _joined_members_response(
        "@mindroom_general:localhost",
        "@alice:localhost",
    )

    with pytest.raises(ThreadTagsError, match="the requester"):
        await _write_operation(client, action)

    client.room_put_state.assert_not_awaited()


@pytest.mark.asyncio
@pytest.mark.parametrize("action", ["set", "remove"])
async def test_thread_tags_write_rejects_non_member_requester_even_with_pl0_event(action: str) -> None:
    """Cross-room writes should reject requesters who are not joined to the target room."""
    client = AsyncMock()
    client.user_id = "@mindroom_general:localhost"
    client.room_get_state_event.return_value = _power_levels_response(
        users={
            "@mindroom_general:localhost": 100,
        },
        users_default=0,
        events={THREAD_TAGS_EVENT_TYPE: 0},
    )
    client.joined_members.return_value = nio.JoinedMembersResponse.from_dict(
        {
            "joined": {
                "@mindroom_general:localhost": {"display_name": "MindRoom"},
            },
        },
        room_id="!room:localhost",
    )

    with pytest.raises(ThreadTagsError, match="not joined to the target room"):
        await _write_operation(client, action)

    client.joined_members.assert_awaited_once_with("!room:localhost")
    client.room_put_state.assert_not_awaited()


@pytest.mark.asyncio
@pytest.mark.parametrize("action", ["set", "remove"])
async def test_thread_tags_write_rejects_missing_client_user_id(action: str) -> None:
    """State writes should fail closed when the Matrix client identity is unavailable."""
    client = AsyncMock()
    client.user_id = None

    with pytest.raises(ThreadTagsError, match=r"client\.user_id must be a non-empty string"):
        await _write_operation(client, action)

    client.room_get_state_event.assert_not_awaited()
    client.room_put_state.assert_not_awaited()


@pytest.mark.asyncio
@pytest.mark.parametrize("action", ["set", "remove"])
async def test_thread_tags_write_rejects_when_power_levels_fetch_fails(action: str) -> None:
    """State writes should fail closed when room power levels cannot be loaded."""
    client = AsyncMock()
    client.user_id = "@mindroom_general:localhost"
    client.room_get_state_event.return_value = object()

    with pytest.raises(ThreadTagsError, match="Failed to fetch Matrix power levels"):
        await _write_operation(client, action)

    client.room_put_state.assert_not_awaited()


@pytest.mark.asyncio
async def test_get_thread_tags_raises_for_non_missing_state_fetch_error() -> None:
    """State read failures should not be reported as missing tags."""
    client = AsyncMock()
    client.room_get_state.return_value = object()

    with pytest.raises(ThreadTagsError, match="Failed to fetch room state for thread tags"):
        await get_thread_tags(
            client,
            "!room:localhost",
            "$thread-root:localhost",
        )


@pytest.mark.asyncio
async def test_resolve_thread_root_event_id_for_client_returns_root_for_root_event() -> None:
    """A proven thread root should normalize to itself."""
    client = AsyncMock()
    client.room_get_event.return_value = _message_event_response(
        "$thread-root:localhost",
        content={"body": "Root", "msgtype": "m.text"},
    )
    thread_reply = nio.RoomMessageText.from_dict(
        {
            "event_id": "$thread-reply:localhost",
            "sender": "@user:localhost",
            "origin_server_ts": 2,
            "room_id": "!room:localhost",
            "type": "m.room.message",
            "content": {
                "body": "Reply",
                "msgtype": "m.text",
                "m.relates_to": {
                    "rel_type": "m.thread",
                    "event_id": "$thread-root:localhost",
                },
            },
        },
    )
    thread_root = nio.RoomMessageText.from_dict(
        {
            "event_id": "$thread-root:localhost",
            "sender": "@user:localhost",
            "origin_server_ts": 1,
            "room_id": "!room:localhost",
            "type": "m.room.message",
            "content": {
                "body": "Root",
                "msgtype": "m.text",
            },
        },
    )
    room_messages_response = MagicMock(spec=nio.RoomMessagesResponse)
    room_messages_response.chunk = [thread_reply, thread_root]
    room_messages_response.end = None
    client.room_messages = AsyncMock(return_value=room_messages_response)

    normalized = await resolve_thread_root_event_id_for_client(
        client,
        "!room:localhost",
        "$thread-root:localhost",
    )

    assert normalized == "$thread-root:localhost"
    client.room_messages.assert_awaited_once()


@pytest.mark.asyncio
async def test_resolve_thread_root_event_id_for_client_returns_none_for_unproven_root_event() -> None:
    """Unproven room-level roots should not normalize as canonical thread IDs."""
    client = AsyncMock()
    client.room_get_event.return_value = _message_event_response(
        "$thread-root:localhost",
        content={"body": "Root", "msgtype": "m.text"},
    )
    room_messages_response = MagicMock(spec=nio.RoomMessagesResponse)
    room_messages_response.chunk = [
        nio.RoomMessageText.from_dict(
            {
                "event_id": "$thread-root:localhost",
                "sender": "@user:localhost",
                "origin_server_ts": 1,
                "room_id": "!room:localhost",
                "type": "m.room.message",
                "content": {
                    "body": "Root",
                    "msgtype": "m.text",
                },
            },
        ),
    ]
    room_messages_response.end = None
    client.room_messages = AsyncMock(return_value=room_messages_response)

    normalized = await resolve_thread_root_event_id_for_client(
        client,
        "!room:localhost",
        "$thread-root:localhost",
    )

    assert normalized is None


@pytest.mark.asyncio
async def test_resolve_thread_root_event_id_for_client_returns_thread_root_for_thread_reply() -> None:
    """Native thread replies should normalize to their thread root."""
    client = AsyncMock()
    client.room_get_event = AsyncMock(
        side_effect=[
            _message_event_response(
                "$thread-reply:localhost",
                content={
                    "body": "Reply",
                    "msgtype": "m.text",
                    "m.relates_to": {
                        "rel_type": "m.thread",
                        "event_id": "$thread-root:localhost",
                    },
                },
            ),
        ],
    )

    normalized = await resolve_thread_root_event_id_for_client(
        client,
        "!room:localhost",
        "$thread-reply:localhost",
    )

    assert normalized == "$thread-root:localhost"
    client.room_get_event.assert_awaited_once_with("!room:localhost", "$thread-reply:localhost")


@pytest.mark.asyncio
@pytest.mark.parametrize("event_id", ["", "   "])
async def test_resolve_thread_root_event_id_for_client_returns_none_for_blank_input(event_id: str) -> None:
    """Blank event IDs should be rejected before any event lookup."""
    client = AsyncMock()

    normalized = await resolve_thread_root_event_id_for_client(
        client,
        "!room:localhost",
        event_id,
    )

    assert normalized is None
    client.room_get_event.assert_not_awaited()


@pytest.mark.asyncio
async def test_resolve_thread_root_event_id_for_client_returns_thread_root_for_plain_reply_to_thread_reply() -> None:
    """Plain replies to explicit thread messages should normalize to the existing thread root."""
    client = AsyncMock()
    client.room_get_event = AsyncMock(
        side_effect=[
            _message_event_response(
                "$plain-reply:localhost",
                content={
                    "body": "Bridge reply",
                    "msgtype": "m.text",
                    "m.relates_to": {"m.in_reply_to": {"event_id": "$thread-reply:localhost"}},
                },
            ),
            _message_event_response(
                "$thread-reply:localhost",
                content={
                    "body": "Reply",
                    "msgtype": "m.text",
                    "m.relates_to": {
                        "rel_type": "m.thread",
                        "event_id": "$thread-root:localhost",
                    },
                },
            ),
        ],
    )

    normalized = await resolve_thread_root_event_id_for_client(
        client,
        "!room:localhost",
        "$plain-reply:localhost",
    )

    assert normalized == "$thread-root:localhost"
    assert client.room_get_event.await_args_list[0].args == ("!room:localhost", "$plain-reply:localhost")
    assert client.room_get_event.await_args_list[1].args == ("!room:localhost", "$thread-reply:localhost")


@pytest.mark.asyncio
async def test_resolve_thread_root_event_id_for_client_returns_none_for_missing_plain_reply_target() -> None:
    """Plain replies to missing/redacted targets should remain best-effort room-level."""
    client = AsyncMock()
    client.room_get_event = AsyncMock(
        side_effect=[
            _message_event_response(
                "$plain-reply:localhost",
                content={
                    "body": "Bridge reply",
                    "msgtype": "m.text",
                    "m.relates_to": {"m.in_reply_to": {"event_id": "$redacted-root:localhost"}},
                },
            ),
            nio.RoomGetEventError("missing", status_code="M_NOT_FOUND"),
        ],
    )

    normalized = await resolve_thread_root_event_id_for_client(
        client,
        "!room:localhost",
        "$plain-reply:localhost",
    )

    assert normalized is None
    assert client.room_get_event.await_args_list[0].args == ("!room:localhost", "$plain-reply:localhost")
    assert client.room_get_event.await_args_list[1].args == ("!room:localhost", "$redacted-root:localhost")


@pytest.mark.asyncio
async def test_resolve_thread_root_event_id_for_client_walks_transitively_to_threaded_ancestor() -> None:
    """Plain replies should normalize transitively when the reply chain eventually reaches a threaded ancestor."""
    client = AsyncMock()
    client.room_get_event = AsyncMock(
        side_effect=[
            _message_event_response(
                "$plain-reply-2:localhost",
                content={
                    "body": "Second bridge reply",
                    "msgtype": "m.text",
                    "m.relates_to": {"m.in_reply_to": {"event_id": "$plain-reply-1:localhost"}},
                },
            ),
            _message_event_response(
                "$plain-reply-1:localhost",
                content={
                    "body": "First bridge reply",
                    "msgtype": "m.text",
                    "m.relates_to": {"m.in_reply_to": {"event_id": "$thread-reply:localhost"}},
                },
            ),
            _message_event_response(
                "$thread-reply:localhost",
                content={
                    "body": "Thread reply",
                    "msgtype": "m.text",
                    "m.relates_to": {"rel_type": "m.thread", "event_id": "$thread-root:localhost"},
                },
            ),
        ],
    )

    normalized = await resolve_thread_root_event_id_for_client(
        client,
        "!room:localhost",
        "$plain-reply-2:localhost",
    )

    assert normalized == "$thread-root:localhost"
    assert client.room_get_event.await_args_list[0].args == ("!room:localhost", "$plain-reply-2:localhost")
    assert client.room_get_event.await_args_list[1].args == ("!room:localhost", "$plain-reply-1:localhost")
    assert client.room_get_event.await_args_list[2].args == ("!room:localhost", "$thread-reply:localhost")
    client.room_messages.assert_not_awaited()


@pytest.mark.asyncio
async def test_resolve_thread_root_event_id_for_client_returns_thread_root_for_plain_reply_to_thread_root() -> None:
    """Plain replies to the actual thread root should normalize back to that root."""
    client = AsyncMock()
    client.room_get_event = AsyncMock(
        side_effect=[
            _message_event_response(
                "$plain-reply:localhost",
                content={
                    "body": "Bridge reply",
                    "msgtype": "m.text",
                    "m.relates_to": {"m.in_reply_to": {"event_id": "$thread-root:localhost"}},
                },
            ),
            _message_event_response(
                "$thread-root:localhost",
                content={
                    "body": "Root",
                    "msgtype": "m.text",
                },
            ),
        ],
    )
    thread_reply = nio.RoomMessageText.from_dict(
        {
            "event_id": "$thread-reply:localhost",
            "sender": "@user:localhost",
            "origin_server_ts": 2,
            "room_id": "!room:localhost",
            "type": "m.room.message",
            "content": {
                "body": "Reply",
                "msgtype": "m.text",
                "m.relates_to": {
                    "rel_type": "m.thread",
                    "event_id": "$thread-root:localhost",
                },
            },
        },
    )
    thread_root = nio.RoomMessageText.from_dict(
        {
            "event_id": "$thread-root:localhost",
            "sender": "@user:localhost",
            "origin_server_ts": 1,
            "room_id": "!room:localhost",
            "type": "m.room.message",
            "content": {
                "body": "Root",
                "msgtype": "m.text",
            },
        },
    )
    room_messages_response = MagicMock(spec=nio.RoomMessagesResponse)
    room_messages_response.chunk = [thread_reply, thread_root]
    room_messages_response.end = None
    client.room_messages = AsyncMock(return_value=room_messages_response)

    normalized = await resolve_thread_root_event_id_for_client(
        client,
        "!room:localhost",
        "$plain-reply:localhost",
    )

    assert normalized == "$thread-root:localhost"
    client.room_messages.assert_awaited_once()


@pytest.mark.asyncio
async def test_resolve_thread_root_event_id_for_client_returns_none_for_plain_reply_to_plain_root_message() -> None:
    """Plain replies to unrelated room roots should not normalize as threads."""
    client = AsyncMock()
    client.room_get_event = AsyncMock(
        side_effect=[
            _message_event_response(
                "$plain-reply:localhost",
                content={
                    "body": "Plain reply",
                    "msgtype": "m.text",
                    "m.relates_to": {"m.in_reply_to": {"event_id": "$plain-root:localhost"}},
                },
            ),
            _message_event_response(
                "$plain-root:localhost",
                content={
                    "body": "Root",
                    "msgtype": "m.text",
                },
            ),
        ],
    )
    plain_root = nio.RoomMessageText.from_dict(
        {
            "event_id": "$plain-root:localhost",
            "sender": "@user:localhost",
            "origin_server_ts": 1,
            "room_id": "!room:localhost",
            "type": "m.room.message",
            "content": {
                "body": "Root",
                "msgtype": "m.text",
            },
        },
    )
    room_messages_response = MagicMock(spec=nio.RoomMessagesResponse)
    room_messages_response.chunk = [plain_root]
    room_messages_response.end = None
    client.room_messages = AsyncMock(return_value=room_messages_response)

    normalized = await resolve_thread_root_event_id_for_client(
        client,
        "!room:localhost",
        "$plain-reply:localhost",
    )

    assert normalized is None
    assert client.room_get_event.await_args_list[0].args == ("!room:localhost", "$plain-reply:localhost")
    assert client.room_get_event.await_args_list[1].args == ("!room:localhost", "$plain-root:localhost")
    client.room_messages.assert_awaited_once()


@pytest.mark.asyncio
async def test_resolve_thread_root_event_id_for_client_returns_none_for_plain_reply() -> None:
    """Plain replies should no longer be promoted into synthetic thread roots."""
    client = AsyncMock()
    client.room_get_event = AsyncMock(
        side_effect=[
            _message_event_response(
                "$reply-two:localhost",
                content={
                    "body": "Reply two",
                    "msgtype": "m.text",
                    "m.relates_to": {"m.in_reply_to": {"event_id": "$reply-one:localhost"}},
                },
            ),
            _message_event_response(
                "$reply-one:localhost",
                content={"body": "Reply one", "msgtype": "m.text"},
            ),
        ],
    )
    room_messages_response = MagicMock(spec=nio.RoomMessagesResponse)
    room_messages_response.chunk = [
        nio.RoomMessageText.from_dict(
            {
                "event_id": "$reply-one:localhost",
                "sender": "@user:localhost",
                "origin_server_ts": 1,
                "room_id": "!room:localhost",
                "type": "m.room.message",
                "content": {
                    "body": "Reply one",
                    "msgtype": "m.text",
                },
            },
        ),
    ]
    room_messages_response.end = None
    client.room_messages = AsyncMock(return_value=room_messages_response)

    normalized = await resolve_thread_root_event_id_for_client(
        client,
        "!room:localhost",
        "$reply-two:localhost",
    )

    assert normalized is None
    assert client.room_get_event.await_args_list[0].args == ("!room:localhost", "$reply-two:localhost")
    assert client.room_get_event.await_args_list[1].args == ("!room:localhost", "$reply-one:localhost")
    client.room_messages.assert_awaited_once()


@pytest.mark.asyncio
async def test_resolve_thread_root_event_id_for_client_returns_none_when_lookup_fails() -> None:
    """Missing or unreadable events should not guess a thread root."""
    client = AsyncMock()
    client.room_get_event = AsyncMock(return_value=object())

    normalized = await resolve_thread_root_event_id_for_client(
        client,
        "!room:localhost",
        "$reply-one:localhost",
    )
    assert normalized is None
    client.room_get_event.assert_awaited_once_with("!room:localhost", "$reply-one:localhost")


@pytest.mark.asyncio
async def test_resolve_thread_root_event_id_for_client_uses_cache_when_event_lookup_misses() -> None:
    """Cache-backed normalization should still work when the homeserver lookup misses."""
    client = AsyncMock()
    client.room_get_event = AsyncMock(return_value=object())
    conversation_cache = MagicMock()
    conversation_cache.get_thread_id_for_event = AsyncMock(return_value="$thread-root:localhost")

    normalized = await resolve_thread_root_event_id_for_client(
        client,
        "!room:localhost",
        "$fresh-local-reply:localhost",
        conversation_cache=conversation_cache,
    )

    assert normalized == "$thread-root:localhost"
    conversation_cache.get_thread_id_for_event.assert_awaited_once_with(
        "!room:localhost",
        "$fresh-local-reply:localhost",
    )


@pytest.mark.asyncio
async def test_resolve_thread_root_event_id_for_client_resolves_thread_edit_via_original_event() -> None:
    """Thread edits should normalize directly from explicit thread metadata."""
    client = AsyncMock()
    client.room_get_event = AsyncMock(
        return_value=_message_event_response(
            "$edit:localhost",
            content={
                "body": "* edited",
                "msgtype": "m.text",
                "m.new_content": {
                    "body": "edited",
                    "msgtype": "m.text",
                    "m.relates_to": {
                        "rel_type": "m.thread",
                        "event_id": "$thread-root:localhost",
                    },
                },
                "m.relates_to": {
                    "rel_type": "m.replace",
                    "event_id": "$thread-reply:localhost",
                },
            },
        ),
    )

    normalized = await resolve_thread_root_event_id_for_client(
        client,
        "!room:localhost",
        "$edit:localhost",
    )

    assert normalized == "$thread-root:localhost"
    client.room_get_event.assert_awaited_once_with("!room:localhost", "$edit:localhost")


@pytest.mark.asyncio
async def test_resolve_thread_root_event_id_for_client_resolves_thread_reply_edit_via_original_event_without_nested_thread_metadata() -> (
    None
):
    """Edits of thread replies should fall back through the original event when nested thread metadata is absent."""
    client = AsyncMock()
    client.room_get_event = AsyncMock(
        side_effect=[
            _message_event_response(
                "$edit:localhost",
                content={
                    "body": "* edited",
                    "msgtype": "m.text",
                    "m.new_content": {
                        "body": "edited",
                        "msgtype": "m.text",
                    },
                    "m.relates_to": {
                        "rel_type": "m.replace",
                        "event_id": "$thread-reply:localhost",
                    },
                },
            ),
            _message_event_response(
                "$thread-reply:localhost",
                content={
                    "body": "Reply",
                    "msgtype": "m.text",
                    "m.relates_to": {
                        "rel_type": "m.thread",
                        "event_id": "$thread-root:localhost",
                    },
                },
            ),
        ],
    )

    normalized = await resolve_thread_root_event_id_for_client(
        client,
        "!room:localhost",
        "$edit:localhost",
    )

    assert normalized == "$thread-root:localhost"
    assert client.room_get_event.await_args_list[0].args == ("!room:localhost", "$edit:localhost")
    assert client.room_get_event.await_args_list[1].args == ("!room:localhost", "$thread-reply:localhost")


@pytest.mark.asyncio
async def test_resolve_thread_root_event_id_for_client_resolves_edit_of_promoted_plain_reply_via_original_reply_target() -> (
    None
):
    """Edits of promoted plain replies should reuse the same transitive thread inheritance."""
    client = AsyncMock()
    client.room_get_event = AsyncMock(
        side_effect=[
            _message_event_response(
                "$edit:localhost",
                content={
                    "body": "* edited",
                    "msgtype": "m.text",
                    "m.new_content": {
                        "body": "edited",
                        "msgtype": "m.text",
                    },
                    "m.relates_to": {
                        "rel_type": "m.replace",
                        "event_id": "$plain-reply:localhost",
                    },
                },
            ),
            _message_event_response(
                "$plain-reply:localhost",
                content={
                    "body": "Bridge reply",
                    "msgtype": "m.text",
                    "m.relates_to": {"m.in_reply_to": {"event_id": "$thread-reply:localhost"}},
                },
            ),
            _message_event_response(
                "$thread-reply:localhost",
                content={
                    "body": "Reply",
                    "msgtype": "m.text",
                    "m.relates_to": {
                        "rel_type": "m.thread",
                        "event_id": "$thread-root:localhost",
                    },
                },
            ),
        ],
    )

    normalized = await resolve_thread_root_event_id_for_client(
        client,
        "!room:localhost",
        "$edit:localhost",
    )

    assert normalized == "$thread-root:localhost"
    fetched_event_ids = [call.args[1] for call in client.room_get_event.await_args_list]
    assert fetched_event_ids[0] == "$edit:localhost"
    assert "$plain-reply:localhost" in fetched_event_ids


@pytest.mark.asyncio
async def test_resolve_thread_root_event_id_for_client_resolves_thread_root_edit_via_original_event() -> None:
    """Edits of thread-root messages should normalize back to the original root event."""
    client = AsyncMock()
    client.room_get_event = AsyncMock(
        side_effect=[
            _message_event_response(
                "$edit:localhost",
                content={
                    "body": "* edited root",
                    "msgtype": "m.text",
                    "m.new_content": {
                        "body": "edited root",
                        "msgtype": "m.text",
                    },
                    "m.relates_to": {
                        "rel_type": "m.replace",
                        "event_id": "$thread-root:localhost",
                    },
                },
            ),
            _message_event_response(
                "$thread-root:localhost",
                content={"body": "Root", "msgtype": "m.text"},
            ),
        ],
    )
    thread_reply = nio.RoomMessageText.from_dict(
        {
            "event_id": "$thread-reply:localhost",
            "sender": "@user:localhost",
            "origin_server_ts": 2,
            "room_id": "!room:localhost",
            "type": "m.room.message",
            "content": {
                "body": "Reply",
                "msgtype": "m.text",
                "m.relates_to": {
                    "rel_type": "m.thread",
                    "event_id": "$thread-root:localhost",
                },
            },
        },
    )
    thread_root = nio.RoomMessageText.from_dict(
        {
            "event_id": "$thread-root:localhost",
            "sender": "@user:localhost",
            "origin_server_ts": 1,
            "room_id": "!room:localhost",
            "type": "m.room.message",
            "content": {
                "body": "Root",
                "msgtype": "m.text",
            },
        },
    )
    room_messages_response = MagicMock(spec=nio.RoomMessagesResponse)
    room_messages_response.chunk = [thread_reply, thread_root]
    room_messages_response.end = None
    client.room_messages = AsyncMock(return_value=room_messages_response)

    normalized = await resolve_thread_root_event_id_for_client(
        client,
        "!room:localhost",
        "$edit:localhost",
    )

    assert normalized == "$thread-root:localhost"
    assert client.room_get_event.await_args_list[0].args == ("!room:localhost", "$edit:localhost")
    assert client.room_get_event.await_args_list[1].args == ("!room:localhost", "$thread-root:localhost")
    client.room_messages.assert_awaited_once()


@pytest.mark.asyncio
async def test_resolve_thread_root_event_id_for_client_returns_none_for_cyclic_edit_chain() -> None:
    """Cyclic edit chains should fail closed instead of recursing until crash."""
    client = AsyncMock()
    responses = {
        "$edit-a:localhost": _message_event_response(
            "$edit-a:localhost",
            content={
                "body": "* a",
                "msgtype": "m.text",
                "m.new_content": {"body": "a", "msgtype": "m.text"},
                "m.relates_to": {"rel_type": "m.replace", "event_id": "$edit-b:localhost"},
            },
        ),
        "$edit-b:localhost": _message_event_response(
            "$edit-b:localhost",
            content={
                "body": "* b",
                "msgtype": "m.text",
                "m.new_content": {"body": "b", "msgtype": "m.text"},
                "m.relates_to": {"rel_type": "m.replace", "event_id": "$edit-a:localhost"},
            },
        ),
    }
    client.room_get_event = AsyncMock(
        side_effect=lambda _room_id, event_id: responses[event_id],
    )

    normalized = await resolve_thread_root_event_id_for_client(
        client,
        "!room:localhost",
        "$edit-a:localhost",
    )

    assert normalized is None
