"""Tests for the standalone subagents toolkit and session registry helpers."""

from __future__ import annotations

import json
from dataclasses import replace
from types import SimpleNamespace
from typing import TYPE_CHECKING
from unittest.mock import AsyncMock, MagicMock, Mock

import nio
import pytest

import mindroom.tools  # noqa: F401
from mindroom.agent_descriptions import describe_agent
from mindroom.config.agent import AgentConfig
from mindroom.constants import (
    ORIGINAL_SENDER_KEY,
    ROUTER_AGENT_NAME,
    SOURCE_KIND_KEY,
    RuntimePaths,
    resolve_runtime_paths,
)
from mindroom.custom_tools import subagents as subagents_module
from mindroom.custom_tools.delegate import DelegateTools
from mindroom.custom_tools.subagents import SubAgentsTools
from mindroom.dispatch_source import TRUSTED_INTERNAL_RELAY_SOURCE_KIND
from mindroom.entity_resolution import entity_identity_registry
from mindroom.message_target import MessageTarget
from mindroom.session_ids import create_session_id, parse_session_id
from mindroom.thread_summary import THREAD_SUMMARY_MAX_LENGTH
from mindroom.tool_system.metadata import TOOL_METADATA, get_tool_by_name
from mindroom.tool_system.runtime_context import ToolRuntimeContext, tool_runtime_context
from tests.conftest import delivered_matrix_side_effect, make_event_cache_mock
from tests.identity_helpers import actual_entity_usernames, persist_entity_accounts

if TYPE_CHECKING:
    from pathlib import Path


EXPECTED_SUBAGENT_TOOL_NAMES = {
    "agents_list",
    "sessions_send",
    "sessions_spawn",
    "list_sessions",
}
TEST_SUMMARY = "test summary"
TEST_TAG = "test-tag"
TEST_SUBAGENT_PASSWORD = "pw"  # noqa: S105
EXPECTED_SUBAGENTS_DESCRIPTION = (
    "Discover, spawn, and communicate with sub-agent sessions. "
    "`agents_list` reports per-tool capability flags (delegate-aware)."
)


def _make_agent_config(
    *,
    role: str = "Handle test tasks",
    tools: list[str] | None = None,
    delegate_to: list[str] | None = None,
    rooms: list[str] | None = None,
) -> AgentConfig:
    return AgentConfig(
        display_name="TestAgent",
        role=role,
        tools=list(tools) if tools is not None else ["shell"],
        delegate_to=list(delegate_to) if delegate_to is not None else [],
        rooms=list(rooms) if rooms is not None else [],
    )


def _make_config(
    *,
    thread_mode: str = "thread",
    agents: dict[str, AgentConfig] | None = None,
) -> MagicMock:
    config = MagicMock()
    config.agents = agents or {
        "openclaw": _make_agent_config(role="Coordinate work", delegate_to=["code"]),
        "code": _make_agent_config(role="Write code"),
        "research": _make_agent_config(role="Research topics"),
    }
    config.teams = {}
    config.mindroom_user = None
    config.get_domain = MagicMock(return_value="localhost")
    config.get_entity_thread_mode = MagicMock(return_value=thread_mode)
    config.get_agent_tools = MagicMock(side_effect=lambda agent_name: config.agents[agent_name].tool_names)
    config.render_prompt = MagicMock(return_value="Delegate only to listed agents.")
    config.authorization.agent_reply_permissions = {}
    config.authorization.resolve_alias.side_effect = lambda sender_id: sender_id
    return config


def _make_context(
    tmp_path: Path,
    *,
    config: MagicMock | None = None,
    agent_name: str = "openclaw",
    room_id: str = "!room:localhost",
    thread_id: str | None = "$ctx-thread:localhost",
    requester_id: str = "@alice:localhost",
    room_agent_names: list[str] | None = None,
) -> ToolRuntimeContext:
    async def _latest_thread_event_id(
        _room_id: str,
        thread_id: str | None,
        *_args: object,
        **_kwargs: object,
    ) -> str | None:
        return thread_id

    runtime_paths = resolve_runtime_paths(config_path=tmp_path / "config.yaml", storage_path=tmp_path)
    effective_config = config or _make_config()
    _persist_subagent_accounts(effective_config, runtime_paths)
    room = _make_room(effective_config, runtime_paths, room_id, agent_name, room_agent_names)
    conversation_cache = AsyncMock()
    conversation_cache.get_latest_thread_event_id_if_needed.side_effect = _latest_thread_event_id
    conversation_cache.notify_outbound_message = Mock()
    conversation_cache.notify_outbound_redaction = Mock()
    return ToolRuntimeContext(
        agent_name=agent_name,
        target=MessageTarget.resolve(
            room_id=room_id,
            thread_id=thread_id,
            reply_to_event_id=None,
        ),
        requester_id=requester_id,
        client=MagicMock(),
        config=effective_config,
        runtime_paths=runtime_paths,
        event_cache=make_event_cache_mock(),
        conversation_cache=conversation_cache,
        room=room,
        storage_path=tmp_path,
    )


def _make_room(
    config: MagicMock,
    runtime_paths: RuntimePaths,
    room_id: str,
    own_agent: str,
    agent_names: list[str] | None = None,
) -> nio.MatrixRoom:
    entity_ids = entity_identity_registry(config, runtime_paths).current_ids
    room = nio.MatrixRoom(room_id=room_id, own_user_id=entity_ids[own_agent].full_id)
    for agent_name in agent_names or list(config.agents):
        room.add_member(entity_ids[agent_name].full_id, config.agents[agent_name].display_name, None)
    room.members_synced = True
    return room


def _persist_subagent_accounts(
    config: MagicMock,
    runtime_paths: RuntimePaths,
    *,
    usernames: dict[str, str] | None = None,
) -> None:
    persist_entity_accounts(
        config,
        runtime_paths,
        usernames=usernames or actual_entity_usernames(config),
        password=TEST_SUBAGENT_PASSWORD,
    )


def _stub_spawn_followups(
    monkeypatch: pytest.MonkeyPatch,
) -> tuple[AsyncMock, AsyncMock, MagicMock]:
    summary_mock = AsyncMock(return_value="$summary:localhost")
    tag_mock = AsyncMock(return_value=SimpleNamespace())
    update_mock = MagicMock()
    monkeypatch.setattr(subagents_module, "send_thread_summary_event", summary_mock)
    monkeypatch.setattr(subagents_module, "set_thread_tag", tag_mock)
    monkeypatch.setattr(subagents_module, "update_last_summary_count", update_mock)
    return summary_mock, tag_mock, update_mock


def test_subagents_tool_registered_and_instantiates() -> None:
    """Subagents should be present in metadata and constructible from the registry."""
    assert "subagents" in TOOL_METADATA
    assert TOOL_METADATA["subagents"].description == EXPECTED_SUBAGENTS_DESCRIPTION
    assert isinstance(get_tool_by_name("subagents", resolve_runtime_paths(), worker_target=None), SubAgentsTools)


def test_subagents_tool_name_contract() -> None:
    """Toolkit should expose the expected stable async method names."""
    tool = SubAgentsTools()
    exposed_names = {func.name for func in tool.functions.values()} | {
        func.name for func in tool.async_functions.values()
    }
    assert exposed_names == EXPECTED_SUBAGENT_TOOL_NAMES


@pytest.mark.asyncio
async def test_agents_list_requires_runtime_context() -> None:
    """agents_list should return a structured context-unavailable error outside runtime scope."""
    payload = json.loads(await SubAgentsTools().agents_list())
    assert payload["status"] == "error"
    assert payload["tool"] == "agents_list"
    assert "context" in payload["message"]


@pytest.mark.asyncio
async def test_agents_list_payload_structure(tmp_path: Path) -> None:
    """agents_list should return sorted capability rows without the caller."""
    config = _make_config()
    ctx = _make_context(tmp_path, config=config)

    with tool_runtime_context(ctx):
        payload = json.loads(await SubAgentsTools().agents_list())

    assert payload["status"] == "ok"
    assert payload["tool"] == "agents_list"
    assert payload["current_agent"] == "openclaw"
    assert [row["name"] for row in payload["agents"]] == ["code", "research"]
    assert all(set(row) == {"name", "can_delegate", "can_spawn", "description"} for row in payload["agents"])
    assert all(isinstance(row, dict) for row in payload["agents"])
    assert all(row["can_spawn"] is True for row in payload["agents"])


@pytest.mark.asyncio
async def test_agents_list_uses_current_room_responder_boundary_for_spawn(tmp_path: Path) -> None:
    """agents_list should mark current-room responders as spawnable."""
    config = _make_config()
    ctx = _make_context(tmp_path, config=config, room_agent_names=["openclaw", "code"])

    with tool_runtime_context(ctx):
        payload = json.loads(await SubAgentsTools().agents_list())

    assert payload["status"] == "ok"
    assert [row["name"] for row in payload["agents"]] == ["code"]


@pytest.mark.asyncio
async def test_agents_list_includes_delegate_only_target(tmp_path: Path) -> None:
    """agents_list should show delegate targets even when they cannot be spawned in this room."""
    config = _make_config(
        agents={
            "openclaw": _make_agent_config(role="Coordinate work", delegate_to=["research"]),
            "code": _make_agent_config(role="Write code"),
            "research": _make_agent_config(role="Research topics"),
        },
    )
    ctx = _make_context(tmp_path, config=config, room_agent_names=["openclaw", "code"])

    with tool_runtime_context(ctx):
        payload = json.loads(await SubAgentsTools().agents_list())

    assert payload["status"] == "ok"
    rows_by_name = {row["name"]: row for row in payload["agents"]}
    assert list(rows_by_name) == ["code", "research"]
    assert rows_by_name["code"]["can_delegate"] is False
    assert rows_by_name["code"]["can_spawn"] is True
    assert rows_by_name["research"]["can_delegate"] is True
    assert rows_by_name["research"]["can_spawn"] is False


@pytest.mark.asyncio
async def test_agents_list_does_not_widen_configured_rooms_to_present_unconfigured_agents(tmp_path: Path) -> None:
    """agents_list should keep configured-room scope even when other agents are present."""
    room_id = "!room:localhost"
    config = _make_config(
        agents={
            "openclaw": _make_agent_config(role="Coordinate work", delegate_to=["code"], rooms=[room_id]),
            "code": _make_agent_config(role="Write code", rooms=[room_id]),
            "research": _make_agent_config(role="Research topics"),
        },
    )
    ctx = _make_context(tmp_path, config=config, room_id=room_id, room_agent_names=["openclaw", "code", "research"])

    with tool_runtime_context(ctx):
        payload = json.loads(await SubAgentsTools().agents_list())

    assert payload["status"] == "ok"
    assert [row["name"] for row in payload["agents"]] == ["code"]


@pytest.mark.asyncio
async def test_agents_list_filters_non_running_configured_room_agents_when_live_state_is_known(
    tmp_path: Path,
) -> None:
    """agents_list spawn discovery should require configured-room and live responder availability."""
    room_id = "!room:localhost"
    config = _make_config(
        agents={
            "openclaw": _make_agent_config(role="Coordinate work", rooms=[room_id], delegate_to=[]),
            "code": _make_agent_config(role="Write code", rooms=[room_id]),
            "research": _make_agent_config(role="Research topics", rooms=[room_id]),
            "writer": _make_agent_config(role="Write prose"),
        },
    )
    ctx = _make_context(
        tmp_path,
        config=config,
        room_id=room_id,
        room_agent_names=["openclaw", "code", "research", "writer"],
    )
    ctx = replace(
        ctx,
        orchestrator=SimpleNamespace(
            agent_bots={
                "openclaw": SimpleNamespace(running=True),
                "code": SimpleNamespace(running=False),
                "research": SimpleNamespace(running=True),
                "writer": SimpleNamespace(running=True),
            },
        ),
    )

    with tool_runtime_context(ctx):
        payload = json.loads(await SubAgentsTools().agents_list())

    assert payload["status"] == "ok"
    assert [row["name"] for row in payload["agents"]] == ["research"]
    assert payload["agents"][0]["can_spawn"] is True


@pytest.mark.asyncio
async def test_agents_list_can_delegate_reflects_delegate_to(tmp_path: Path) -> None:
    """agents_list should flag only names present in the caller delegate_to allowlist."""
    config = _make_config(
        agents={
            "A": _make_agent_config(delegate_to=["B"]),
            "B": _make_agent_config(),
            "C": _make_agent_config(),
            "D": _make_agent_config(),
        },
    )
    ctx = _make_context(tmp_path, config=config, agent_name="A")

    with tool_runtime_context(ctx):
        payload = json.loads(await SubAgentsTools().agents_list())

    rows_by_name = {row["name"]: row for row in payload["agents"]}
    assert rows_by_name["B"]["can_delegate"] is True
    assert rows_by_name["C"]["can_delegate"] is False
    assert rows_by_name["D"]["can_delegate"] is False


@pytest.mark.asyncio
async def test_agents_list_empty_delegate_to(tmp_path: Path) -> None:
    """agents_list should report no delegation capability when the caller allowlist is empty."""
    config = _make_config(
        agents={
            "A": _make_agent_config(delegate_to=[]),
            "B": _make_agent_config(),
            "C": _make_agent_config(),
        },
    )
    ctx = _make_context(tmp_path, config=config, agent_name="A")

    with tool_runtime_context(ctx):
        payload = json.loads(await SubAgentsTools().agents_list())

    assert all(row["can_delegate"] is False for row in payload["agents"])


@pytest.mark.asyncio
async def test_agents_list_caller_not_in_config_returns_no_delegate(tmp_path: Path) -> None:
    """agents_list should tolerate caller identities that are not configured agents."""
    config = _make_config(
        agents={
            "A": _make_agent_config(delegate_to=["B"]),
            "B": _make_agent_config(),
            "C": _make_agent_config(),
        },
    )
    ctx = _make_context(tmp_path, config=config, agent_name=ROUTER_AGENT_NAME)

    with tool_runtime_context(ctx):
        payload = json.loads(await SubAgentsTools().agents_list())

    assert [row["name"] for row in payload["agents"]] == ["A", "B", "C"]
    assert all(row["can_delegate"] is False for row in payload["agents"])


@pytest.mark.asyncio
async def test_agents_list_description_matches_describe_agent(tmp_path: Path) -> None:
    """agents_list descriptions should use the shared agent description renderer exactly."""
    config = _make_config()
    ctx = _make_context(tmp_path, config=config)

    with tool_runtime_context(ctx):
        payload = json.loads(await SubAgentsTools().agents_list())

    for row in payload["agents"]:
        assert row["description"] == describe_agent(row["name"], config)


@pytest.mark.asyncio
async def test_agents_list_can_delegate_aligns_with_delegate_tools(tmp_path: Path) -> None:
    """agents_list can_delegate flags should match DelegateTools rejection behavior."""
    config = _make_config(
        agents={
            "A": _make_agent_config(delegate_to=["B"]),
            "B": _make_agent_config(),
            "C": _make_agent_config(),
        },
    )
    ctx = _make_context(tmp_path, config=config, agent_name="A")
    delegate_tools = DelegateTools(
        agent_name="A",
        delegate_to=["B"],
        runtime_paths=ctx.runtime_paths,
        config=config,
    )

    with tool_runtime_context(ctx):
        payload = json.loads(await SubAgentsTools().agents_list())

    for row in payload["agents"]:
        if row["can_delegate"]:
            assert row["name"] in delegate_tools._delegate_to
        else:
            assert row["name"] not in delegate_tools._delegate_to

    result = await delegate_tools.delegate_task("C", "task")
    assert "Cannot delegate to 'C'" in result
    assert "Run agents_list to inspect can_delegate flags." in result


@pytest.mark.asyncio
async def test_sessions_send_requires_runtime_context() -> None:
    """sessions_send should return a structured context-unavailable error outside runtime scope."""
    payload = json.loads(await SubAgentsTools().sessions_send(message="hello"))
    assert payload["status"] == "error"
    assert payload["tool"] == "sessions_send"
    assert "context" in payload["message"]


@pytest.mark.asyncio
async def test_sessions_send_rejects_empty_message(tmp_path: Path) -> None:
    """sessions_send should validate non-empty message content."""
    ctx = _make_context(tmp_path)

    with tool_runtime_context(ctx):
        payload = json.loads(await SubAgentsTools().sessions_send(message="   "))

    assert payload["status"] == "error"
    assert payload["tool"] == "sessions_send"
    assert "cannot be empty" in payload["message"]


@pytest.mark.asyncio
async def test_sessions_send_returns_error_when_matrix_send_fails(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """sessions_send should return an error payload when Matrix dispatch fails."""
    send_mock = AsyncMock(return_value=None)
    monkeypatch.setattr(subagents_module, "_send_matrix_text", send_mock)
    ctx = _make_context(tmp_path)

    with tool_runtime_context(ctx):
        payload = json.loads(await SubAgentsTools().sessions_send(message="hello"))

    assert payload["status"] == "error"
    assert payload["tool"] == "sessions_send"
    assert "Failed to send message" in payload["message"]


@pytest.mark.asyncio
async def test_sessions_send_relays_original_sender(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """sessions_send should preserve original requester identity for relayed events."""
    send_mock = AsyncMock(return_value="$evt")
    monkeypatch.setattr(subagents_module, "_send_matrix_text", send_mock)
    ctx = _make_context(tmp_path, requester_id="@user:localhost")

    with tool_runtime_context(ctx):
        payload = json.loads(await SubAgentsTools().sessions_send(message="hello"))

    assert payload["status"] == "ok"
    send_mock.assert_awaited_once_with(
        ctx,
        room_id=ctx.room_id,
        text="hello",
        thread_id=ctx.thread_id,
        original_sender=ctx.requester_id,
    )


@pytest.mark.asyncio
async def test_sessions_send_agent_id_uses_current_matrix_id(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """sessions_send should target the live persisted Matrix ID after account drift."""
    send_mock = AsyncMock(return_value="$evt")
    monkeypatch.setattr(subagents_module, "_send_matrix_text", send_mock)
    ctx = _make_context(tmp_path)
    _persist_subagent_accounts(ctx.config, ctx.runtime_paths, usernames={"code": "actual_code_live"})
    ctx = replace(ctx, room=_make_room(ctx.config, ctx.runtime_paths, ctx.room_id, ctx.agent_name))

    with tool_runtime_context(ctx):
        payload = json.loads(await SubAgentsTools().sessions_send(message="do it", agent_id="code"))

    assert payload["status"] == "ok"
    send_mock.assert_awaited_once_with(
        ctx,
        room_id=ctx.room_id,
        text="@actual_code_live:localhost do it",
        thread_id=ctx.thread_id,
        original_sender=ctx.requester_id,
    )


@pytest.mark.asyncio
async def test_sessions_send_rejects_unknown_agent_id(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """sessions_send should validate explicit agent_id before Matrix dispatch."""
    send_mock = AsyncMock(return_value="$evt")
    monkeypatch.setattr(subagents_module, "_send_matrix_text", send_mock)
    ctx = _make_context(tmp_path)

    with tool_runtime_context(ctx):
        payload = json.loads(await SubAgentsTools().sessions_send(message="do it", agent_id="missing"))

    assert payload["status"] == "error"
    assert payload["tool"] == "sessions_send"
    assert payload["message"] == "Unknown agent_id 'missing'. Available agents: code, openclaw, research."
    send_mock.assert_not_awaited()


@pytest.mark.asyncio
async def test_sessions_send_rejects_agent_outside_room_responder_boundary(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """sessions_send should reject explicit agents outside the room responder boundary."""
    send_mock = AsyncMock(return_value="$evt")
    monkeypatch.setattr(subagents_module, "_send_matrix_text", send_mock)
    ctx = _make_context(tmp_path, room_agent_names=["openclaw", "code"])

    with tool_runtime_context(ctx):
        payload = json.loads(await SubAgentsTools().sessions_send(message="do it", agent_id="research"))

    assert payload["status"] == "error"
    assert payload["tool"] == "sessions_send"
    assert payload["message"] == "Agent 'research' is not available in this room. Available agents: code, openclaw."
    send_mock.assert_not_awaited()


@pytest.mark.asyncio
async def test_sessions_send_rejects_agent_available_only_in_caller_room(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """sessions_send should validate explicit agents against the target configured room."""
    send_mock = AsyncMock(return_value="$evt")
    monkeypatch.setattr(subagents_module, "_send_matrix_text", send_mock)
    caller_room_id = "!caller:localhost"
    target_room_id = "!target:localhost"
    config = _make_config(
        agents={
            "openclaw": _make_agent_config(role="Coordinate work", rooms=[caller_room_id]),
            "code": _make_agent_config(role="Write code", rooms=[caller_room_id]),
            "research": _make_agent_config(role="Research topics", rooms=[target_room_id]),
        },
    )
    ctx = _make_context(
        tmp_path,
        config=config,
        room_id=caller_room_id,
        room_agent_names=["openclaw", "code"],
    )
    target_session = create_session_id(target_room_id, "$target-thread:localhost")

    with tool_runtime_context(ctx):
        payload = json.loads(
            await SubAgentsTools().sessions_send(
                message="do it",
                session_key=target_session,
                agent_id="code",
            ),
        )

    assert payload["status"] == "error"
    assert payload["tool"] == "sessions_send"
    assert payload["message"] == "Agent 'code' is not available in this room. Available agents: research."
    send_mock.assert_not_awaited()


@pytest.mark.asyncio
async def test_sessions_send_allows_agent_configured_for_target_room(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """sessions_send should allow target-room agents even when absent from the caller room."""
    send_mock = AsyncMock(return_value="$evt")
    monkeypatch.setattr(subagents_module, "_send_matrix_text", send_mock)
    caller_room_id = "!caller:localhost"
    target_room_id = "!target:localhost"
    config = _make_config(
        agents={
            "openclaw": _make_agent_config(role="Coordinate work", rooms=[caller_room_id]),
            "code": _make_agent_config(role="Write code", rooms=[caller_room_id]),
            "research": _make_agent_config(role="Research topics", rooms=[target_room_id]),
        },
    )
    ctx = _make_context(
        tmp_path,
        config=config,
        room_id=caller_room_id,
        room_agent_names=["openclaw", "code"],
    )
    target_session = create_session_id(target_room_id, "$target-thread:localhost")

    with tool_runtime_context(ctx):
        payload = json.loads(
            await SubAgentsTools().sessions_send(
                message="do it",
                session_key=target_session,
                agent_id="research",
            ),
        )

    assert payload["status"] == "ok"
    assert payload["room_id"] == target_room_id
    send_mock.assert_awaited_once_with(
        ctx,
        room_id=target_room_id,
        text="@actual_research:localhost do it",
        thread_id="$target-thread:localhost",
        original_sender=ctx.requester_id,
    )


@pytest.mark.asyncio
async def test_sessions_send_defaults_to_resolved_thread_session(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """sessions_send should keep first-turn follow-ups in the canonical reply thread."""
    send_mock = AsyncMock(return_value="$evt")
    monkeypatch.setattr(subagents_module, "_send_matrix_text", send_mock)
    base_context = _make_context(tmp_path, thread_id=None)
    resolved_thread_id = "$resolved-thread:localhost"
    ctx = replace(
        base_context,
        target=replace(
            base_context.target,
            resolved_thread_id=resolved_thread_id,
            session_id=create_session_id(base_context.room_id, resolved_thread_id),
        ),
    )

    with tool_runtime_context(ctx):
        payload = json.loads(await SubAgentsTools().sessions_send(message="hello"))

    assert payload["status"] == "ok"
    assert payload["session_key"] == create_session_id(ctx.room_id, resolved_thread_id)
    send_mock.assert_awaited_once_with(
        ctx,
        room_id=ctx.room_id,
        text="hello",
        thread_id="$resolved-thread:localhost",
        original_sender=ctx.requester_id,
    )


@pytest.mark.asyncio
async def test_send_matrix_text_uses_latest_thread_event_id_for_fallback(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Threaded subagent sends should include the latest thread event for fallback replies."""
    send_mock = AsyncMock(side_effect=delivered_matrix_side_effect("$evt"))
    monkeypatch.setattr(subagents_module, "send_message_result", send_mock)
    event_cache = MagicMock()
    ctx = replace(_make_context(tmp_path, requester_id="@user:localhost"), event_cache=event_cache)
    ctx.conversation_cache.get_latest_thread_event_id_if_needed = AsyncMock(return_value="$latest:localhost")

    await subagents_module._send_matrix_text(
        ctx,
        room_id=ctx.room_id,
        text="hello",
        thread_id=ctx.thread_id,
        original_sender=ctx.requester_id,
    )

    ctx.conversation_cache.get_latest_thread_event_id_if_needed.assert_awaited_once_with(
        ctx.room_id,
        ctx.thread_id,
        caller_label="subagent_tool_send",
    )
    content = send_mock.await_args.args[2]
    assert content["m.relates_to"]["event_id"] == ctx.thread_id
    assert content["m.relates_to"]["m.in_reply_to"]["event_id"] == "$latest:localhost"
    assert content[ORIGINAL_SENDER_KEY] == ctx.requester_id


@pytest.mark.asyncio
async def test_send_matrix_text_marks_original_sender_as_trusted_relay(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Root session messages should carry trusted human-relay metadata."""
    send_mock = AsyncMock(side_effect=delivered_matrix_side_effect("$evt"))
    monkeypatch.setattr(subagents_module, "send_message_result", send_mock)
    ctx = _make_context(tmp_path, requester_id="@user:localhost")

    await subagents_module._send_matrix_text(
        ctx,
        room_id=ctx.room_id,
        text="@actual_openclaw:localhost do work",
        thread_id=None,
        original_sender=ctx.requester_id,
    )

    content = send_mock.await_args.args[2]
    assert content[ORIGINAL_SENDER_KEY] == ctx.requester_id
    assert content[SOURCE_KIND_KEY] == TRUSTED_INTERNAL_RELAY_SOURCE_KIND


@pytest.mark.asyncio
async def test_sessions_send_rejects_room_mode_threaded_dispatch(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """sessions_send should reject threaded dispatch into room-mode target agents."""
    send_mock = AsyncMock(return_value="$evt")
    monkeypatch.setattr(subagents_module, "_send_matrix_text", send_mock)
    ctx = _make_context(tmp_path, config=_make_config(thread_mode="room"))
    target_session = create_session_id(ctx.room_id, "$worker-thread:localhost")

    with tool_runtime_context(ctx):
        payload = json.loads(
            await SubAgentsTools().sessions_send(
                message="hello",
                session_key=target_session,
                agent_id="openclaw",
            ),
        )

    assert payload["status"] == "error"
    assert payload["tool"] == "sessions_send"
    assert "thread_mode=room" in payload["message"]
    send_mock.assert_not_awaited()


@pytest.mark.asyncio
async def test_sessions_send_checks_target_room_thread_mode(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """sessions_send should evaluate thread mode using the target room context."""
    send_mock = AsyncMock(return_value="$evt")
    monkeypatch.setattr(subagents_module, "_send_matrix_text", send_mock)
    target_room_id = "!target:localhost"
    config = _make_config(
        agents={
            "openclaw": _make_agent_config(role="Coordinate work", rooms=[target_room_id]),
            "code": _make_agent_config(role="Write code"),
            "research": _make_agent_config(role="Research topics"),
        },
    )
    config.get_entity_thread_mode.side_effect = lambda _agent_name, _runtime_paths, room_id=None: (
        "room" if room_id == target_room_id else "thread"
    )
    ctx = _make_context(tmp_path, config=config)
    target_session = create_session_id(target_room_id, "$worker-thread:localhost")

    with tool_runtime_context(ctx):
        payload = json.loads(
            await SubAgentsTools().sessions_send(
                message="hello",
                session_key=target_session,
                agent_id="openclaw",
            ),
        )

    assert payload["status"] == "error"
    assert payload["tool"] == "sessions_send"
    assert "thread_mode=room" in payload["message"]
    send_mock.assert_not_awaited()
    config.get_entity_thread_mode.assert_called_with(
        "openclaw",
        ctx.runtime_paths,
        room_id=target_room_id,
    )


@pytest.mark.asyncio
async def test_sessions_send_label_resolves_to_tracked_session(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """sessions_send should resolve labels to tracked session keys in scope."""
    send_mock = AsyncMock(return_value="$evt")
    monkeypatch.setattr(subagents_module, "_send_matrix_text", send_mock)
    ctx = _make_context(tmp_path)

    session_key = create_session_id(ctx.room_id, "$target:localhost")
    subagents_module._record_session(ctx, session_key=session_key, label="work", target_agent="openclaw")

    with tool_runtime_context(ctx):
        payload = json.loads(await SubAgentsTools().sessions_send(message="hello", label="work"))

    assert payload["status"] == "ok"
    assert payload["session_key"] == session_key
    send_mock.assert_awaited_once_with(
        ctx,
        room_id=ctx.room_id,
        text="hello",
        thread_id="$target:localhost",
        original_sender=ctx.requester_id,
    )


@pytest.mark.asyncio
async def test_sessions_spawn_requires_runtime_context() -> None:
    """sessions_spawn should return a structured context-unavailable error outside runtime scope."""
    payload = json.loads(await SubAgentsTools().sessions_spawn(task="do this", summary=TEST_SUMMARY, tag=TEST_TAG))
    assert payload["status"] == "error"
    assert payload["tool"] == "sessions_spawn"
    assert "context" in payload["message"]


@pytest.mark.asyncio
async def test_sessions_spawn_rejects_empty_task(tmp_path: Path) -> None:
    """sessions_spawn should validate non-empty task content."""
    ctx = _make_context(tmp_path)

    with tool_runtime_context(ctx):
        payload = json.loads(await SubAgentsTools().sessions_spawn(task="  ", summary=TEST_SUMMARY, tag=TEST_TAG))

    assert payload["status"] == "error"
    assert payload["tool"] == "sessions_spawn"
    assert "Task cannot be empty" in payload["message"]


@pytest.mark.asyncio
async def test_sessions_spawn_returns_error_when_matrix_send_fails(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """sessions_spawn should return an error payload when Matrix dispatch fails."""
    send_mock = AsyncMock(return_value=None)
    monkeypatch.setattr(subagents_module, "_send_matrix_text", send_mock)
    ctx = _make_context(tmp_path)

    with tool_runtime_context(ctx):
        payload = json.loads(
            await SubAgentsTools().sessions_spawn(task="do thing", summary=TEST_SUMMARY, tag=TEST_TAG),
        )

    assert payload["status"] == "error"
    assert payload["tool"] == "sessions_spawn"
    assert "Failed to send spawn message" in payload["message"]


@pytest.mark.asyncio
async def test_sessions_spawn_relays_original_sender(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """sessions_spawn should preserve original requester identity for relayed events."""
    send_mock = AsyncMock(return_value="$event")
    monkeypatch.setattr(subagents_module, "_send_matrix_text", send_mock)
    _stub_spawn_followups(monkeypatch)
    ctx = _make_context(tmp_path, requester_id="@user:localhost")

    with tool_runtime_context(ctx):
        payload = json.loads(
            await SubAgentsTools().sessions_spawn(task="do thing", summary=TEST_SUMMARY, tag=TEST_TAG),
        )

    assert payload["status"] == "ok"
    assert payload["target_agent"] == "openclaw"
    assert payload["event_id"] == "$event"
    assert payload["summary"] == TEST_SUMMARY
    assert payload["tag"] == TEST_TAG
    send_mock.assert_awaited_once_with(
        ctx,
        room_id=ctx.room_id,
        text="@actual_openclaw:localhost do thing",
        thread_id=None,
        original_sender=ctx.requester_id,
    )


@pytest.mark.asyncio
async def test_sessions_spawn_uses_current_matrix_id(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """sessions_spawn should target the live persisted Matrix ID after account drift."""
    send_mock = AsyncMock(return_value="$event")
    monkeypatch.setattr(subagents_module, "_send_matrix_text", send_mock)
    _stub_spawn_followups(monkeypatch)
    ctx = _make_context(tmp_path)
    _persist_subagent_accounts(ctx.config, ctx.runtime_paths, usernames={"code": "actual_code_live"})
    ctx = replace(ctx, room=_make_room(ctx.config, ctx.runtime_paths, ctx.room_id, ctx.agent_name))

    with tool_runtime_context(ctx):
        payload = json.loads(
            await SubAgentsTools().sessions_spawn(
                task="do thing",
                summary=TEST_SUMMARY,
                tag=TEST_TAG,
                agent_id="code",
            ),
        )

    assert payload["status"] == "ok"
    assert payload["target_agent"] == "code"
    send_mock.assert_awaited_once_with(
        ctx,
        room_id=ctx.room_id,
        text="@actual_code_live:localhost do thing",
        thread_id=None,
        original_sender=ctx.requester_id,
    )


@pytest.mark.asyncio
async def test_sessions_spawn_rejects_unknown_agent_id(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """sessions_spawn should validate explicit agent_id before Matrix dispatch."""
    send_mock = AsyncMock(return_value="$event")
    monkeypatch.setattr(subagents_module, "_send_matrix_text", send_mock)
    ctx = _make_context(tmp_path)

    with tool_runtime_context(ctx):
        payload = json.loads(
            await SubAgentsTools().sessions_spawn(
                task="do thing",
                summary=TEST_SUMMARY,
                tag=TEST_TAG,
                agent_id="missing",
            ),
        )

    assert payload["status"] == "error"
    assert payload["tool"] == "sessions_spawn"
    assert payload["message"] == "Unknown agent_id 'missing'. Available agents: code, openclaw, research."
    send_mock.assert_not_awaited()


@pytest.mark.asyncio
async def test_sessions_spawn_rejects_agent_outside_configured_room_boundary(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """sessions_spawn should reject explicit agents outside the configured room boundary."""
    send_mock = AsyncMock(return_value="$event")
    monkeypatch.setattr(subagents_module, "_send_matrix_text", send_mock)
    _stub_spawn_followups(monkeypatch)
    room_id = "!room:localhost"
    config = _make_config(
        agents={
            "openclaw": _make_agent_config(role="Coordinate work", delegate_to=["code"], rooms=[room_id]),
            "code": _make_agent_config(role="Write code", rooms=[room_id]),
            "research": _make_agent_config(role="Research topics"),
        },
    )
    ctx = _make_context(tmp_path, config=config, room_id=room_id, room_agent_names=["openclaw", "code", "research"])

    with tool_runtime_context(ctx):
        payload = json.loads(
            await SubAgentsTools().sessions_spawn(
                task="do thing",
                summary=TEST_SUMMARY,
                tag=TEST_TAG,
                agent_id="research",
            ),
        )

    assert payload["status"] == "error"
    assert payload["tool"] == "sessions_spawn"
    assert payload["message"] == "Agent 'research' is not available in this room. Available agents: code, openclaw."
    send_mock.assert_not_awaited()


@pytest.mark.asyncio
async def test_sessions_spawn_rejects_room_mode_target_agent(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """sessions_spawn should reject isolated sessions for room-mode target agents."""
    send_mock = AsyncMock(return_value="$event")
    monkeypatch.setattr(subagents_module, "_send_matrix_text", send_mock)
    ctx = _make_context(tmp_path, config=_make_config(thread_mode="room"))

    with tool_runtime_context(ctx):
        payload = json.loads(
            await SubAgentsTools().sessions_spawn(task="do thing", summary=TEST_SUMMARY, tag=TEST_TAG),
        )

    assert payload["status"] == "error"
    assert payload["tool"] == "sessions_spawn"
    assert "thread_mode=room" in payload["message"]
    send_mock.assert_not_awaited()


@pytest.mark.asyncio
async def test_sessions_spawn_rejects_blank_summary(tmp_path: Path) -> None:
    """sessions_spawn should reject blank summaries before spawning."""
    ctx = _make_context(tmp_path)

    with tool_runtime_context(ctx):
        payload = json.loads(await SubAgentsTools().sessions_spawn(task="do thing", summary="   ", tag=TEST_TAG))

    assert payload["status"] == "error"
    assert payload["tool"] == "sessions_spawn"
    assert "summary must be a non-empty string" in payload["message"]


@pytest.mark.asyncio
async def test_sessions_spawn_rejects_overlong_summary(tmp_path: Path) -> None:
    """sessions_spawn should reject summaries longer than the shared thread-summary limit."""
    ctx = _make_context(tmp_path)

    with tool_runtime_context(ctx):
        payload = json.loads(
            await SubAgentsTools().sessions_spawn(
                task="do thing",
                summary="x" * (THREAD_SUMMARY_MAX_LENGTH + 1),
                tag=TEST_TAG,
            ),
        )

    assert payload["status"] == "error"
    assert payload["tool"] == "sessions_spawn"
    assert f"{THREAD_SUMMARY_MAX_LENGTH} characters or fewer" in payload["message"]


@pytest.mark.asyncio
async def test_sessions_spawn_strips_markdown_from_summary(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """sessions_spawn should normalize markdown before writing the summary event."""
    send_mock = AsyncMock(return_value="$event")
    tag_mock = AsyncMock(return_value=SimpleNamespace())
    update_mock = MagicMock()
    monkeypatch.setattr(subagents_module, "_send_matrix_text", send_mock)
    monkeypatch.setattr(subagents_module, "set_thread_tag", tag_mock)
    monkeypatch.setattr(subagents_module, "update_last_summary_count", update_mock)
    ctx = _make_context(tmp_path)
    ctx.client.room_send = AsyncMock(
        return_value=nio.RoomSendResponse(event_id="$summary:localhost", room_id=ctx.room_id),
    )

    with tool_runtime_context(ctx):
        payload = json.loads(
            await SubAgentsTools().sessions_spawn(
                task="do thing",
                summary="# **Fix** [ISSUE-116](http://example.com)\n> `deploy` ~~done~~",
                tag=TEST_TAG,
            ),
        )

    assert payload["status"] == "ok"
    assert payload["summary"] == "Fix ISSUE-116 deploy done"
    ctx.client.room_send.assert_awaited_once()
    content = ctx.client.room_send.call_args.kwargs["content"]
    assert content["body"] == "Fix ISSUE-116 deploy done"
    assert content["io.mindroom.thread_summary"]["summary"] == "Fix ISSUE-116 deploy done"
    tag_mock.assert_awaited_once_with(
        ctx.client,
        ctx.room_id,
        "$event",
        TEST_TAG,
        set_by=ctx.requester_id,
    )
    update_mock.assert_called_once_with(ctx.room_id, "$event", 1)


@pytest.mark.asyncio
async def test_sessions_spawn_rejects_invalid_tag(tmp_path: Path) -> None:
    """sessions_spawn should reject invalid tag names before spawning."""
    ctx = _make_context(tmp_path)

    with tool_runtime_context(ctx):
        payload = json.loads(
            await SubAgentsTools().sessions_spawn(task="do thing", summary=TEST_SUMMARY, tag="INVALID TAG!"),
        )

    assert payload["status"] == "error"
    assert payload["tool"] == "sessions_spawn"
    assert "lowercase letters, digits, or hyphens" in payload["message"]


@pytest.mark.asyncio
async def test_sessions_spawn_rejects_resolved_tag_before_matrix_send(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Spawn metadata must not create lifecycle state through generic tagging."""
    send_mock = AsyncMock(return_value="$event")
    tag_mock = AsyncMock()
    monkeypatch.setattr(subagents_module, "_send_matrix_text", send_mock)
    monkeypatch.setattr(subagents_module, "set_thread_tag", tag_mock)
    ctx = _make_context(tmp_path)

    with tool_runtime_context(ctx):
        payload = json.loads(
            await SubAgentsTools().sessions_spawn(task="do thing", summary=TEST_SUMMARY, tag="resolved"),
        )

    assert payload["status"] == "error"
    assert "lifecycle state" in payload["message"]
    send_mock.assert_not_awaited()
    tag_mock.assert_not_awaited()


@pytest.mark.asyncio
async def test_sessions_spawn_validates_before_matrix_send(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """sessions_spawn should validate required spawn metadata before any Matrix call."""
    send_mock = AsyncMock(return_value="$event")
    monkeypatch.setattr(subagents_module, "_send_matrix_text", send_mock)
    ctx = _make_context(tmp_path)

    with tool_runtime_context(ctx):
        payload = json.loads(
            await SubAgentsTools().sessions_spawn(task="do thing", summary=TEST_SUMMARY, tag="INVALID TAG!"),
        )

    assert payload["status"] == "error"
    send_mock.assert_not_awaited()


@pytest.mark.asyncio
async def test_sessions_spawn_sets_summary_after_spawn(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """sessions_spawn should write the requested thread summary after creating the thread."""
    send_mock = AsyncMock(return_value="$event")
    monkeypatch.setattr(subagents_module, "_send_matrix_text", send_mock)
    summary_mock, _, update_mock = _stub_spawn_followups(monkeypatch)
    ctx = _make_context(tmp_path)

    with tool_runtime_context(ctx):
        payload = json.loads(
            await SubAgentsTools().sessions_spawn(task="do thing", summary=TEST_SUMMARY, tag=TEST_TAG),
        )

    assert payload["status"] == "ok"
    summary_mock.assert_awaited_once_with(
        ctx.client,
        ctx.room_id,
        "$event",
        TEST_SUMMARY,
        1,
        "manual",
        ctx.conversation_cache,
        known_latest_thread_event_id="$event",
    )
    update_mock.assert_called_once_with(ctx.room_id, "$event", 1)


@pytest.mark.asyncio
async def test_sessions_spawn_summary_does_not_read_the_new_thread(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """The initial spawn summary should relate to the new root without reading thread history.

    A freshly created thread has no cache snapshot, so a latest-event lookup here
    starts a homeserver scan that races the spawned agent's first outbound event.
    """
    monkeypatch.setattr(subagents_module, "_send_matrix_text", AsyncMock(return_value="$spawn-root:localhost"))
    monkeypatch.setattr(subagents_module, "set_thread_tag", AsyncMock(return_value=SimpleNamespace()))
    monkeypatch.setattr(subagents_module, "update_last_summary_count", MagicMock())
    ctx = _make_context(tmp_path)
    ctx.client.room_send = AsyncMock(
        return_value=nio.RoomSendResponse(event_id="$summary:localhost", room_id=ctx.room_id),
    )
    ctx.conversation_cache.get_latest_thread_event_id_if_needed = AsyncMock(return_value="$never-read:localhost")

    with tool_runtime_context(ctx):
        payload = json.loads(
            await SubAgentsTools().sessions_spawn(task="do thing", summary=TEST_SUMMARY, tag=TEST_TAG),
        )

    assert payload["status"] == "ok"
    ctx.conversation_cache.get_latest_thread_event_id_if_needed.assert_not_awaited()
    relates_to = ctx.client.room_send.call_args.kwargs["content"]["m.relates_to"]
    assert relates_to["rel_type"] == "m.thread"
    assert relates_to["event_id"] == "$spawn-root:localhost"
    assert relates_to["m.in_reply_to"] == {"event_id": "$spawn-root:localhost"}


@pytest.mark.asyncio
async def test_sessions_spawn_reuse_still_resolves_the_latest_thread_event(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Reusing a labelled session targets an existing thread, so its tail must still be resolved."""
    summary_mock, _, _ = _stub_spawn_followups(monkeypatch)
    ctx = _make_context(tmp_path)
    monkeypatch.setattr(
        subagents_module,
        "_resolve_by_label",
        lambda *_args, **_kwargs: (
            create_session_id(ctx.room_id, "$existing-root:localhost"),
            {"target_agent": "code"},
        ),
    )

    with tool_runtime_context(ctx):
        payload = json.loads(
            await SubAgentsTools().sessions_spawn(
                task="do thing",
                summary=TEST_SUMMARY,
                tag=TEST_TAG,
                label="worklog",
            ),
        )

    assert payload["status"] == "ok"
    assert payload["reused"] is True
    assert summary_mock.await_args.kwargs["known_latest_thread_event_id"] is None


@pytest.mark.asyncio
async def test_sessions_spawn_sets_tag_after_spawn(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """sessions_spawn should write the requested thread tag after creating the thread."""
    send_mock = AsyncMock(return_value="$event")
    monkeypatch.setattr(subagents_module, "_send_matrix_text", send_mock)
    _, tag_mock, _ = _stub_spawn_followups(monkeypatch)
    ctx = _make_context(tmp_path)

    with tool_runtime_context(ctx):
        payload = json.loads(
            await SubAgentsTools().sessions_spawn(task="do thing", summary=TEST_SUMMARY, tag=TEST_TAG),
        )

    assert payload["status"] == "ok"
    tag_mock.assert_awaited_once_with(
        ctx.client,
        ctx.room_id,
        "$event",
        TEST_TAG,
        set_by=ctx.requester_id,
    )


@pytest.mark.asyncio
async def test_sessions_spawn_returns_warnings_on_summary_failure(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """sessions_spawn should still succeed when setting the summary fails."""
    send_mock = AsyncMock(return_value="$event")
    monkeypatch.setattr(subagents_module, "_send_matrix_text", send_mock)
    summary_mock, tag_mock, update_mock = _stub_spawn_followups(monkeypatch)
    summary_mock.side_effect = RuntimeError("boom")
    ctx = _make_context(tmp_path)

    with tool_runtime_context(ctx):
        payload = json.loads(
            await SubAgentsTools().sessions_spawn(task="do thing", summary=TEST_SUMMARY, tag=TEST_TAG),
        )

    assert payload["status"] == "ok"
    assert payload["warnings"] == ["Failed to set thread summary: boom"]
    tag_mock.assert_awaited_once()
    update_mock.assert_not_called()


@pytest.mark.asyncio
async def test_sessions_spawn_returns_warnings_when_summary_event_id_is_missing(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """sessions_spawn should warn when summary send returns no event id."""
    send_mock = AsyncMock(return_value="$event")
    monkeypatch.setattr(subagents_module, "_send_matrix_text", send_mock)
    summary_mock, tag_mock, update_mock = _stub_spawn_followups(monkeypatch)
    summary_mock.return_value = None
    ctx = _make_context(tmp_path)

    with tool_runtime_context(ctx):
        payload = json.loads(
            await SubAgentsTools().sessions_spawn(task="do thing", summary=TEST_SUMMARY, tag=TEST_TAG),
        )

    assert payload["status"] == "ok"
    assert payload["warnings"] == ["Failed to set thread summary."]
    tag_mock.assert_awaited_once()
    update_mock.assert_not_called()


@pytest.mark.asyncio
async def test_sessions_spawn_returns_warnings_on_tag_failure(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """sessions_spawn should still succeed when setting the tag fails."""
    send_mock = AsyncMock(return_value="$event")
    monkeypatch.setattr(subagents_module, "_send_matrix_text", send_mock)
    summary_mock, tag_mock, update_mock = _stub_spawn_followups(monkeypatch)
    tag_mock.side_effect = RuntimeError("no power")
    ctx = _make_context(tmp_path)

    with tool_runtime_context(ctx):
        payload = json.loads(
            await SubAgentsTools().sessions_spawn(task="do thing", summary=TEST_SUMMARY, tag=TEST_TAG),
        )

    assert payload["status"] == "ok"
    assert payload["warnings"] == ["Failed to set thread tag: no power"]
    summary_mock.assert_awaited_once()
    update_mock.assert_called_once_with(ctx.room_id, "$event", 1)


@pytest.mark.asyncio
async def test_sessions_spawn_response_includes_summary_and_tag(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """sessions_spawn should return normalized summary and tag values in the success payload."""
    send_mock = AsyncMock(return_value="$event")
    monkeypatch.setattr(subagents_module, "_send_matrix_text", send_mock)
    _stub_spawn_followups(monkeypatch)
    ctx = _make_context(tmp_path)

    with tool_runtime_context(ctx):
        payload = json.loads(
            await SubAgentsTools().sessions_spawn(
                task="do thing",
                summary="  test   summary  ",
                tag="Test-Tag",
            ),
        )

    assert payload["status"] == "ok"
    assert payload["summary"] == TEST_SUMMARY
    assert payload["tag"] == TEST_TAG
    assert "warnings" not in payload


@pytest.mark.asyncio
async def test_list_sessions_requires_runtime_context() -> None:
    """list_sessions should return a structured context-unavailable error outside runtime scope."""
    payload = json.loads(await SubAgentsTools().list_sessions())
    assert payload["status"] == "error"
    assert payload["tool"] == "list_sessions"
    assert "context" in payload["message"]


@pytest.mark.asyncio
async def test_list_sessions_returns_tracked_sessions(tmp_path: Path) -> None:
    """list_sessions should return sessions persisted via _record_session."""
    ctx = _make_context(tmp_path)
    session_key = create_session_id(ctx.room_id, "$child:localhost")
    subagents_module._record_session(ctx, session_key=session_key, label="my-task", target_agent="code")

    with tool_runtime_context(ctx):
        payload = json.loads(await SubAgentsTools().list_sessions())

    assert payload["status"] == "ok"
    assert payload["tool"] == "list_sessions"
    assert payload["total"] == 1
    session = payload["sessions"][0]
    assert session["session_key"] == session_key
    assert session["label"] == "my-task"
    assert session["target_agent"] == "code"


@pytest.mark.asyncio
async def test_list_sessions_empty_when_no_sessions(tmp_path: Path) -> None:
    """list_sessions should return an empty page when registry has no in-scope sessions."""
    ctx = _make_context(tmp_path)

    with tool_runtime_context(ctx):
        payload = json.loads(await SubAgentsTools().list_sessions())

    assert payload["status"] == "ok"
    assert payload["sessions"] == []
    assert payload["total"] == 0


def test_load_registry_handles_non_dict_payload(tmp_path: Path) -> None:
    """_load_registry should normalize non-dict JSON payloads to an empty mapping."""
    ctx = _make_context(tmp_path)
    registry_dir = tmp_path / "subagents"
    registry_dir.mkdir(parents=True)
    (registry_dir / "session_registry.json").write_text(json.dumps(["unexpected", "array"]))

    registry = subagents_module._load_registry(ctx)
    assert registry == {}


def test_load_registry_returns_existing_dict_without_migration(tmp_path: Path) -> None:
    """_load_registry should preserve dict payloads (including legacy shapes) as-is."""
    ctx = _make_context(tmp_path)
    registry_dir = tmp_path / "subagents"
    registry_dir.mkdir(parents=True)
    old_data = {
        "sessions": {
            "!room:localhost:$thread:localhost": {
                "label": "old-session",
                "target_agent": "code",
            },
        },
        "runs": {"run-1": {"status": "accepted"}},
    }
    (registry_dir / "session_registry.json").write_text(json.dumps(old_data))

    registry = subagents_module._load_registry(ctx)
    assert registry == old_data


def test_subagent_session_key_reverse_parse_matches_canonical_parser() -> None:
    """Subagent registry session keys should reverse-parse through the canonical parser."""
    room_id = "!room:with:colons:localhost"
    thread_id = "$thread$with$dollars:localhost"
    session_key = create_session_id(room_id, thread_id)

    assert parse_session_id(session_key) == (room_id, thread_id)
    assert subagents_module._session_key_to_room_thread(session_key) == (room_id, thread_id)


def test_subagent_session_key_reverse_parse_keeps_room_level_keys() -> None:
    """Subagent registry room-level session keys should not invent a thread id."""
    room_id = "!room:with:colons:localhost"

    assert parse_session_id(room_id) == (room_id, None)
    assert subagents_module._session_key_to_room_thread(room_id) == (room_id, None)


def test_resolve_by_label_require_thread_skips_entries_without_thread_id(tmp_path: Path) -> None:
    """Thread reuse should ignore malformed labeled entries without a stored thread id."""
    ctx = _make_context(tmp_path)
    session_key = create_session_id(ctx.room_id, "$child:localhost")
    subagents_module._record_session(ctx, session_key=session_key, label="work", target_agent="code")

    registry = subagents_module._load_registry(ctx)
    registry[session_key]["thread_id"] = None
    subagents_module._save_registry(ctx, registry)

    assert subagents_module._resolve_by_label(ctx, "work", require_thread=True) is None


@pytest.mark.asyncio
async def test_sessions_spawn_dedup_returns_existing_for_duplicate_label(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """sessions_spawn should return existing session when label already exists in scope."""
    send_mock = AsyncMock(return_value="$event")
    monkeypatch.setattr(subagents_module, "_send_matrix_text", send_mock)
    summary_mock, tag_mock, update_mock = _stub_spawn_followups(monkeypatch)
    ctx = _make_context(tmp_path)

    # First spawn creates the session
    with tool_runtime_context(ctx):
        first = json.loads(
            await SubAgentsTools().sessions_spawn(
                task="do thing",
                summary=TEST_SUMMARY,
                tag=TEST_TAG,
                label="work",
            ),
        )
    assert first["status"] == "ok"
    assert "reused" not in first
    send_mock.assert_awaited_once()
    summary_mock.assert_awaited_once_with(
        ctx.client,
        ctx.room_id,
        "$event",
        TEST_SUMMARY,
        1,
        "manual",
        ctx.conversation_cache,
        known_latest_thread_event_id="$event",
    )
    tag_mock.assert_awaited_once_with(
        ctx.client,
        ctx.room_id,
        "$event",
        TEST_TAG,
        set_by=ctx.requester_id,
    )
    update_mock.assert_called_once_with(ctx.room_id, "$event", 1)

    send_mock.reset_mock()
    summary_mock.reset_mock()
    tag_mock.reset_mock()
    update_mock.reset_mock()

    # Second spawn with same label should reuse, no Matrix message sent
    with tool_runtime_context(ctx):
        second = json.loads(
            await SubAgentsTools().sessions_spawn(
                task="do thing again",
                summary=TEST_SUMMARY,
                tag=TEST_TAG,
                label="work",
            ),
        )
    assert second["status"] == "ok"
    assert second["reused"] is True
    assert second["session_key"] == first["session_key"]
    assert second["target_agent"] == first["target_agent"]
    assert second["event_id"] == first["event_id"]
    assert second["summary"] == TEST_SUMMARY
    assert second["tag"] == TEST_TAG
    send_mock.assert_not_awaited()
    summary_mock.assert_awaited_once_with(
        ctx.client,
        ctx.room_id,
        "$event",
        TEST_SUMMARY,
        0,
        "manual",
        ctx.conversation_cache,
        known_latest_thread_event_id=None,
    )
    tag_mock.assert_awaited_once_with(
        ctx.client,
        ctx.room_id,
        "$event",
        TEST_TAG,
        set_by=ctx.requester_id,
    )
    update_mock.assert_not_called()


@pytest.mark.asyncio
async def test_sessions_spawn_skips_reuse_when_registry_entry_lacks_thread_id(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """sessions_spawn should create a fresh thread when a labeled entry is missing thread_id metadata."""
    send_mock = AsyncMock(return_value="$new-event")
    monkeypatch.setattr(subagents_module, "_send_matrix_text", send_mock)
    summary_mock, tag_mock, update_mock = _stub_spawn_followups(monkeypatch)
    ctx = _make_context(tmp_path)
    session_key = create_session_id(ctx.room_id, "$canonical-thread:localhost")
    subagents_module._record_session(ctx, session_key=session_key, label="work", target_agent="code")

    registry = subagents_module._load_registry(ctx)
    registry[session_key]["thread_id"] = None
    subagents_module._save_registry(ctx, registry)

    with tool_runtime_context(ctx):
        payload = json.loads(
            await SubAgentsTools().sessions_spawn(
                task="do thing",
                summary=TEST_SUMMARY,
                tag=TEST_TAG,
                label="work",
            ),
        )

    assert payload["status"] == "ok"
    assert "reused" not in payload
    assert payload["event_id"] == "$new-event"
    assert payload["session_key"] == create_session_id(ctx.room_id, "$new-event")
    send_mock.assert_awaited_once()
    summary_mock.assert_awaited_once_with(
        ctx.client,
        ctx.room_id,
        "$new-event",
        TEST_SUMMARY,
        1,
        "manual",
        ctx.conversation_cache,
        known_latest_thread_event_id="$new-event",
    )
    tag_mock.assert_awaited_once_with(
        ctx.client,
        ctx.room_id,
        "$new-event",
        TEST_TAG,
        set_by=ctx.requester_id,
    )
    update_mock.assert_called_once_with(ctx.room_id, "$new-event", 1)


@pytest.mark.asyncio
async def test_sessions_spawn_reuse_derives_thread_id_from_session_key(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """sessions_spawn should target the canonical thread id encoded in the session key when reusing."""
    send_mock = AsyncMock(return_value="$new-event")
    monkeypatch.setattr(subagents_module, "_send_matrix_text", send_mock)
    summary_mock, tag_mock, update_mock = _stub_spawn_followups(monkeypatch)
    ctx = _make_context(tmp_path)
    session_key = create_session_id(ctx.room_id, "$canonical-thread:localhost")
    subagents_module._record_session(ctx, session_key=session_key, label="work", target_agent="code")

    registry = subagents_module._load_registry(ctx)
    registry[session_key]["thread_id"] = "$stale-thread:localhost"
    subagents_module._save_registry(ctx, registry)

    with tool_runtime_context(ctx):
        payload = json.loads(
            await SubAgentsTools().sessions_spawn(
                task="do thing",
                summary=TEST_SUMMARY,
                tag=TEST_TAG,
                label="work",
            ),
        )

    assert payload["status"] == "ok"
    assert payload["reused"] is True
    assert payload["session_key"] == session_key
    assert payload["event_id"] == "$canonical-thread:localhost"
    send_mock.assert_not_awaited()
    summary_mock.assert_awaited_once_with(
        ctx.client,
        ctx.room_id,
        "$canonical-thread:localhost",
        TEST_SUMMARY,
        0,
        "manual",
        ctx.conversation_cache,
        known_latest_thread_event_id=None,
    )
    tag_mock.assert_awaited_once_with(
        ctx.client,
        ctx.room_id,
        "$canonical-thread:localhost",
        TEST_TAG,
        set_by=ctx.requester_id,
    )
    update_mock.assert_not_called()


@pytest.mark.asyncio
async def test_sessions_spawn_skips_room_level_reuse_candidates(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """sessions_spawn should create a new thread when the only labeled match is room-level."""
    send_mock = AsyncMock(return_value="$event")
    monkeypatch.setattr(subagents_module, "_send_matrix_text", send_mock)
    summary_mock, tag_mock, update_mock = _stub_spawn_followups(monkeypatch)
    ctx = _make_context(tmp_path)
    subagents_module._record_session(ctx, session_key=ctx.room_id, label="work", target_agent="code")

    with tool_runtime_context(ctx):
        payload = json.loads(
            await SubAgentsTools().sessions_spawn(
                task="do thing",
                summary=TEST_SUMMARY,
                tag=TEST_TAG,
                label="work",
            ),
        )

    assert payload["status"] == "ok"
    assert "reused" not in payload
    assert payload["event_id"] == "$event"
    send_mock.assert_awaited_once()
    summary_mock.assert_awaited_once_with(
        ctx.client,
        ctx.room_id,
        "$event",
        TEST_SUMMARY,
        1,
        "manual",
        ctx.conversation_cache,
        known_latest_thread_event_id="$event",
    )
    tag_mock.assert_awaited_once_with(
        ctx.client,
        ctx.room_id,
        "$event",
        TEST_TAG,
        set_by=ctx.requester_id,
    )
    update_mock.assert_called_once_with(ctx.room_id, "$event", 1)


@pytest.mark.asyncio
async def test_sessions_spawn_no_label_skips_dedup(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """sessions_spawn should always spawn new session when label is None."""
    send_mock = AsyncMock(return_value="$event1")
    monkeypatch.setattr(subagents_module, "_send_matrix_text", send_mock)
    _stub_spawn_followups(monkeypatch)
    ctx = _make_context(tmp_path)

    with tool_runtime_context(ctx):
        first = json.loads(
            await SubAgentsTools().sessions_spawn(task="do thing", summary=TEST_SUMMARY, tag=TEST_TAG),
        )
    assert first["status"] == "ok"

    send_mock.return_value = "$event2"

    with tool_runtime_context(ctx):
        second = json.loads(
            await SubAgentsTools().sessions_spawn(task="do thing again", summary=TEST_SUMMARY, tag=TEST_TAG),
        )
    assert second["status"] == "ok"
    assert "reused" not in second
    assert send_mock.await_count == 2


@pytest.mark.asyncio
async def test_sessions_spawn_dedup_scoped_by_context(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """sessions_spawn with same label but different scope should spawn new session."""
    send_mock = AsyncMock(return_value="$event1")
    monkeypatch.setattr(subagents_module, "_send_matrix_text", send_mock)
    _stub_spawn_followups(monkeypatch)
    ctx1 = _make_context(tmp_path, requester_id="@alice:localhost")

    with tool_runtime_context(ctx1):
        first = json.loads(
            await SubAgentsTools().sessions_spawn(
                task="do thing",
                summary=TEST_SUMMARY,
                tag=TEST_TAG,
                label="work",
            ),
        )
    assert first["status"] == "ok"

    send_mock.return_value = "$event2"
    ctx2 = _make_context(tmp_path, requester_id="@bob:localhost")

    with tool_runtime_context(ctx2):
        second = json.loads(
            await SubAgentsTools().sessions_spawn(
                task="do thing",
                summary=TEST_SUMMARY,
                tag=TEST_TAG,
                label="work",
            ),
        )
    assert second["status"] == "ok"
    assert "reused" not in second
    assert second["session_key"] != first["session_key"]
    assert send_mock.await_count == 2


@pytest.mark.asyncio
async def test_sessions_spawn_dedup_scoped_by_room(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """sessions_spawn with same label but different room should spawn new session."""
    send_mock = AsyncMock(return_value="$event1")
    monkeypatch.setattr(subagents_module, "_send_matrix_text", send_mock)
    _stub_spawn_followups(monkeypatch)
    ctx1 = _make_context(tmp_path, room_id="!room_a:localhost")

    with tool_runtime_context(ctx1):
        first = json.loads(
            await SubAgentsTools().sessions_spawn(
                task="do thing",
                summary=TEST_SUMMARY,
                tag=TEST_TAG,
                label="work",
            ),
        )
    assert first["status"] == "ok"

    send_mock.return_value = "$event2"
    ctx2 = _make_context(tmp_path, room_id="!room_b:localhost")

    with tool_runtime_context(ctx2):
        second = json.loads(
            await SubAgentsTools().sessions_spawn(
                task="do thing",
                summary=TEST_SUMMARY,
                tag=TEST_TAG,
                label="work",
            ),
        )
    assert second["status"] == "ok"
    assert "reused" not in second
    assert second["session_key"] != first["session_key"]
    assert send_mock.await_count == 2


def test_record_session_updates_existing_entry_fields(tmp_path: Path) -> None:
    """_record_session should update mutable fields without dropping existing target_agent."""
    ctx = _make_context(tmp_path)
    session_key = "!room:localhost:$thread:localhost"

    subagents_module._record_session(ctx, session_key=session_key, label="first", target_agent="code")
    subagents_module._record_session(ctx, session_key=session_key, label="second")

    registry = subagents_module._load_registry(ctx)
    assert registry[session_key]["label"] == "second"
    assert registry[session_key]["target_agent"] == "code"
