"""Integration tests for large message handling with streaming and regular messages."""

import asyncio
import json
from collections.abc import AsyncIterator
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, Mock

import nio
import pytest
from agno.models.response import ToolExecution
from agno.run.agent import ToolCallCompletedEvent, ToolCallStartedEvent

from mindroom.config.models import DefaultsConfig
from mindroom.constants import (
    AI_RUN_METADATA_KEY,
    STREAM_STATUS_KEY,
    STREAM_STATUS_STREAMING,
    RuntimePaths,
    resolve_runtime_paths,
)
from mindroom.matrix.client import edit_message_result, send_message_result
from mindroom.matrix.large_messages import (
    _NORMAL_MESSAGE_LIMIT,
    _oversized_nonterminal_streaming_edit_sent_at,
    prepare_large_message,
)
from mindroom.message_target import MessageTarget
from mindroom.streaming import (
    ReplacementStreamingResponse,
    StreamingResponse,
    StreamInputChunk,
    send_streaming_response,
)
from mindroom.tool_system.events import _TOOL_TRACE_KEY, StructuredStreamChunk, ToolTraceEntry


class MockClient:
    """Mock Matrix client for testing."""

    def __init__(self, should_upload_succeed: bool = True) -> None:
        room = MagicMock()
        room.encrypted = False
        self.rooms = {
            "!room:server": room,
            "!test:room": room,
        }
        self.messages_sent = []
        self.uploads: list[dict] = []
        self.should_upload_succeed = should_upload_succeed

    async def room_send(
        self,
        room_id: str,
        message_type: str,
        content: dict,
        *,
        ignore_unverified_devices: bool = False,
    ) -> MagicMock:
        """Mock sending a message."""
        assert message_type == "m.room.message"
        assert ignore_unverified_devices is True
        self.messages_sent.append(("send", room_id, content))

        # Create a mock that passes isinstance check
        response = MagicMock(spec=nio.RoomSendResponse)
        response.event_id = f"$event_{len(self.messages_sent)}"
        return response

    async def upload(self, **kwargs) -> tuple:  # noqa: ANN003
        """Mock file upload - returns tuple like nio."""
        if not self.should_upload_succeed:
            msg = "Upload failed"
            raise Exception(msg)  # noqa: TRY002

        # Capture the uploaded data for test inspection
        data_provider = kwargs.get("data_provider")
        data = data_provider(None, None) if data_provider else None
        self.uploads.append({"data": data, **{k: v for k, v in kwargs.items() if k != "data_provider"}})

        # Create a mock UploadResponse
        response = nio.UploadResponse.from_dict({"content_uri": f"mxc://server/file_{len(self.messages_sent)}"})
        return response, None  # nio returns (response, encryption_dict)


class MockConfig:
    """Mock config for testing."""

    def __init__(self) -> None:
        self.agents = {}
        self.defaults = DefaultsConfig()


def _runtime_paths() -> RuntimePaths:
    """Create an explicit runtime context for streaming tests."""
    return resolve_runtime_paths(config_path=Path("config.yaml"), process_env={})


# ============================================================================
# Non-Streaming Tests
# ============================================================================


@pytest.mark.asyncio
async def test_regular_message_under_limit() -> None:
    """Test that regular messages under the limit pass through unchanged."""
    client = MockClient()

    # Small message
    content = {"body": "Hello world", "msgtype": "m.text"}

    # Should pass through unchanged
    await send_message_result(client, "!room:server", content)

    assert len(client.messages_sent) == 1
    sent_content = client.messages_sent[0][2]
    assert sent_content["body"] == "Hello world"
    assert "io.mindroom.long_text" not in sent_content


@pytest.mark.asyncio
async def test_regular_message_over_limit() -> None:
    """Test that large regular messages get uploaded to MXC."""
    client = MockClient()

    # Large message (100KB)
    large_text = "x" * 100000
    content = {"body": large_text, "msgtype": "m.text"}

    await send_message_result(client, "!room:server", content)

    assert len(client.messages_sent) == 1
    sent_content = client.messages_sent[0][2]

    # Should be an m.file message
    assert sent_content["msgtype"] == "m.file"
    assert sent_content["filename"] == "message-content.json"

    # Should have truncated body preview
    assert len(sent_content["body"]) < len(large_text)
    assert "[Message continues in attached file]" in sent_content["body"]

    # Should have metadata
    assert "io.mindroom.long_text" in sent_content
    assert sent_content["io.mindroom.long_text"]["version"] == 2
    assert sent_content["io.mindroom.long_text"]["encoding"] == "matrix_event_content_json"
    assert sent_content["io.mindroom.long_text"]["is_complete_content"] is True

    # Should have file URL
    assert "url" in sent_content or "file" in sent_content


@pytest.mark.asyncio
async def test_edit_message_with_lower_threshold() -> None:
    """Test that edit messages use the lower size threshold."""
    client = MockClient()

    # Message that's under normal limit but over edit limit (30KB)
    text = "y" * 30000
    content = {"body": text, "msgtype": "m.text", "formatted_body": f"<p>{text}</p>"}

    await edit_message_result(client, "!room:server", "$original", content, text)

    assert len(client.messages_sent) == 1
    sent_content = client.messages_sent[0][2]

    # Should be truncated due to edit limit
    # For edits, check m.new_content
    assert "m.new_content" in sent_content
    assert sent_content["m.new_content"]["msgtype"] == "m.file"
    assert "io.mindroom.long_text" in sent_content["m.new_content"]
    assert len(sent_content["m.new_content"]["body"]) < len(text)


@pytest.mark.asyncio
async def test_large_edit_preserves_mindroom_metadata_in_both_payload_layers() -> None:
    """Large edit previews should keep io.mindroom.* metadata on both edit payload layers."""
    client = MockClient()
    text = "z" * 30000
    extra_content = {
        AI_RUN_METADATA_KEY: {"version": 1, "usage": {"total_tokens": 10}},
        "io.mindroom.compaction": {"version": 3, "compacted": False},
    }
    content = {"body": text, "msgtype": "m.text", "formatted_body": f"<p>{text}</p>"}

    await edit_message_result(
        client,
        "!room:server",
        "$original",
        content,
        text,
        extra_content=extra_content,
    )

    assert len(client.messages_sent) == 1
    sent_content = client.messages_sent[0][2]
    assert "m.new_content" in sent_content
    assert sent_content["m.new_content"]["msgtype"] == "m.file"
    for key, value in extra_content.items():
        assert sent_content[key] == value
        assert sent_content["m.new_content"][key] == value

    uploaded_payload = json.loads(client.uploads[0]["data"].read().decode("utf-8"))
    for key, value in extra_content.items():
        assert uploaded_payload["m.new_content"][key] == value


@pytest.mark.asyncio
async def test_threaded_edit_strips_nested_relations_from_replacement_payload() -> None:
    """Threaded edits should keep the top-level replacement relation only."""
    client = MockClient()
    content = {
        "body": "Updated reply",
        "msgtype": "m.text",
        "formatted_body": "<p>Updated reply</p>",
        "m.relates_to": {
            "rel_type": "m.thread",
            "event_id": "$thread_root",
            "m.in_reply_to": {"event_id": "$latest"},
        },
    }

    await edit_message_result(
        client,
        "!room:server",
        "$original",
        content,
        "Updated reply",
    )

    assert len(client.messages_sent) == 1
    sent_content = client.messages_sent[0][2]
    assert sent_content["m.relates_to"] == {"rel_type": "m.replace", "event_id": "$original"}
    assert sent_content["m.new_content"]["body"] == "Updated reply"
    assert sent_content["m.new_content"]["msgtype"] == "m.text"
    assert "m.relates_to" not in sent_content["m.new_content"]


# ============================================================================
# Streaming Tests
# ============================================================================


@pytest.mark.asyncio
async def test_streaming_initial_message_under_limit() -> None:
    """Test streaming with initial message under limit."""
    client = MockClient()
    config = MockConfig()

    streaming = StreamingResponse(
        target=MessageTarget.resolve("!test:room", None, None),
        config=config,
        runtime_paths=_runtime_paths(),
    )

    # Small initial content
    await streaming.update_content("Hello streaming world", client)

    # Should trigger initial send
    assert len(client.messages_sent) == 1
    sent_content = client.messages_sent[0][2]
    assert "Hello streaming world" in sent_content["body"]
    assert "io.mindroom.long_text" not in sent_content


@pytest.mark.asyncio
async def test_streaming_initial_message_over_limit() -> None:
    """Test streaming with initial message over limit."""
    client = MockClient()
    config = MockConfig()

    streaming = StreamingResponse(
        target=MessageTarget.resolve("!test:room", None, None),
        config=config,
        runtime_paths=_runtime_paths(),
    )

    # Large initial content (60KB - over normal limit)
    large_text = "a" * 60000
    streaming.accumulated_text = large_text
    streaming.last_update = float("-inf")  # Force immediate send

    await streaming._send_or_edit_message(client, is_final=True)

    # Should have sent with large message handling
    assert len(client.messages_sent) == 1
    sent_content = client.messages_sent[0][2]
    assert sent_content["msgtype"] == "m.file"
    assert len(sent_content["body"]) < 60000
    assert "io.mindroom.long_text" in sent_content


@pytest.mark.asyncio
async def test_streaming_large_initial_message_records_transformed_content_to_cache() -> None:
    """Streaming write-through should cache the exact oversized event content Matrix stored."""
    client = MockClient()
    config = MockConfig()
    conversation_cache = AsyncMock()
    conversation_cache.notify_outbound_message = Mock()

    streaming = StreamingResponse(
        target=MessageTarget.resolve("!test:room", None, None),
        config=config,
        runtime_paths=_runtime_paths(),
        conversation_cache=conversation_cache,
    )

    streaming.accumulated_text = "a" * 60000
    streaming.last_update = float("-inf")

    await streaming._send_or_edit_message(client, is_final=True)

    sent_content = client.messages_sent[0][2]
    conversation_cache.notify_outbound_message.assert_called_once_with(
        "!test:room",
        "$event_1",
        sent_content,
    )


@pytest.mark.asyncio
async def test_streaming_edit_grows_over_limit() -> None:
    """Test streaming where edit grows beyond limit."""
    client = MockClient()
    config = MockConfig()

    streaming = StreamingResponse(
        target=MessageTarget.resolve("!test:room", None, None),
        config=config,
        runtime_paths=_runtime_paths(),
    )

    # Start with small message
    streaming.accumulated_text = "Small start"
    streaming.last_update = float("-inf")
    await streaming._send_or_edit_message(client, is_final=False)

    # Should have an event ID now
    assert streaming.event_id is not None
    assert len(client.messages_sent) == 1

    # Now grow to large message (35KB - over edit limit)
    large_text = "b" * 35000
    streaming.accumulated_text = large_text

    # This should trigger edit with large message handling
    await streaming._send_or_edit_message(client, is_final=True)

    # Should have sent an edit
    assert len(client.messages_sent) == 2
    edit_content = client.messages_sent[1][2]

    # Edit should have large message handling
    assert "m.new_content" in edit_content
    assert edit_content["m.new_content"]["msgtype"] == "m.file"
    assert "io.mindroom.long_text" in edit_content["m.new_content"]
    assert len(edit_content["m.new_content"]["body"]) < 35000


@pytest.mark.asyncio
async def test_streaming_large_edit_records_transformed_content_to_cache() -> None:
    """Streaming edit write-through should cache the exact transformed edit event."""
    client = MockClient()
    config = MockConfig()
    conversation_cache = AsyncMock()
    conversation_cache.notify_outbound_message = Mock()

    streaming = StreamingResponse(
        target=MessageTarget.resolve("!test:room", None, None),
        config=config,
        runtime_paths=_runtime_paths(),
        conversation_cache=conversation_cache,
    )

    streaming.accumulated_text = "Small start"
    streaming.last_update = float("-inf")
    await streaming._send_or_edit_message(client, is_final=False)
    conversation_cache.notify_outbound_message.reset_mock()

    streaming.accumulated_text = "b" * 35000
    streaming.last_update = float("-inf")

    await streaming._send_or_edit_message(client, is_final=True)

    edit_content = client.messages_sent[1][2]
    conversation_cache.notify_outbound_message.assert_called_once_with(
        "!test:room",
        "$event_2",
        edit_content,
    )


@pytest.mark.asyncio
async def test_streaming_multiple_edits_with_growth(monkeypatch: pytest.MonkeyPatch) -> None:
    """Test streaming with multiple edits as message grows."""
    _oversized_nonterminal_streaming_edit_sent_at.clear()
    monotonic_values = iter([100.0, 106.0, 112.0])
    monkeypatch.setattr("mindroom.matrix.large_messages.monotonic", lambda: next(monotonic_values))
    client = MockClient()
    config = MockConfig()

    streaming = StreamingResponse(
        target=MessageTarget.resolve("!test:room", None, None),
        config=config,
        runtime_paths=_runtime_paths(),
    )

    # Simulate progressive growth
    sizes = [
        ("Initial", 100),
        ("Growing", 10000),
        ("Large", 28000),  # Over edit limit
        ("Larger", 35000),  # Way over edit limit
    ]

    for label, size in sizes:
        streaming.accumulated_text = "x" * size
        streaming.last_update = float("-inf")
        is_final = label == "Larger"

        await streaming._send_or_edit_message(client, is_final=is_final)

        # After first, should have event_id
        if label != "Initial":
            assert streaming.event_id is not None

    # Check final state
    assert len(client.messages_sent) == len(sizes)

    nonterminal_large_edit = client.messages_sent[-2][2]
    assert nonterminal_large_edit["msgtype"] == "m.notice"
    assert nonterminal_large_edit["m.new_content"]["msgtype"] == "m.notice"
    assert nonterminal_large_edit["m.new_content"][STREAM_STATUS_KEY] == STREAM_STATUS_STREAMING
    assert nonterminal_large_edit["m.new_content"]["format"] == "org.matrix.custom.html"
    assert "Streaming preview truncated" in nonterminal_large_edit["m.new_content"]["formatted_body"]
    assert nonterminal_large_edit["m.new_content"]["io.mindroom.long_text"]["version"] == 2
    assert nonterminal_large_edit["m.new_content"]["io.mindroom.long_text"]["encoding"] == "matrix_event_content_json"

    terminal_large_edit = client.messages_sent[-1][2]
    assert terminal_large_edit["m.new_content"]["msgtype"] == "m.file"
    assert "io.mindroom.long_text" in terminal_large_edit["m.new_content"]


@pytest.mark.asyncio
async def test_streaming_with_thread_context() -> None:
    """Test that streaming preserves thread context with large messages."""
    client = MockClient()
    config = MockConfig()

    streaming = StreamingResponse(
        target=MessageTarget.resolve("!test:room", "$thread_root", "$reply_to"),
        config=config,
        runtime_paths=_runtime_paths(),
    )

    # Large message
    large_text = "t" * 60000
    streaming.accumulated_text = large_text
    streaming.last_update = float("-inf")

    await streaming._send_or_edit_message(client, is_final=True)

    sent_content = client.messages_sent[0][2]

    # Should preserve thread context
    assert "m.relates_to" in sent_content
    # Thread relationship should be preserved
    relates_to = sent_content.get("m.relates_to", {})
    assert relates_to.get("event_id") == "$thread_root" or relates_to.get("rel_type") == "m.thread"

    # Should have large message handling
    assert sent_content["msgtype"] == "m.file"
    assert "io.mindroom.long_text" in sent_content


# ============================================================================
# Edge Cases
# ============================================================================


@pytest.mark.asyncio
async def test_message_exactly_at_limit() -> None:
    """Test message that's exactly at the size limit."""
    client = MockClient()

    # Create message exactly at normal limit
    # Account for JSON overhead (~2KB) in the calculation
    text_size = _NORMAL_MESSAGE_LIMIT - 2500
    text = "e" * text_size
    content = {"body": text, "msgtype": "m.text"}

    result = await prepare_large_message(client, "!room:server", content)

    # Should pass through unchanged (just under limit)
    assert result == content
    assert "io.mindroom.long_text" not in result


@pytest.mark.asyncio
async def test_message_with_formatted_body_no_tools() -> None:
    """Large messages upload full source content JSON sidecar."""
    client = MockClient()

    # Large message with HTML body/format fields
    large_text = "f" * 100000
    large_html = f"<p>{'f' * 100000}</p>"
    content = {
        "body": large_text,
        "formatted_body": large_html,
        "msgtype": "m.text",
        "format": "org.matrix.custom.html",
    }

    result = await prepare_large_message(client, "!room:server", content)

    # Should be an m.file message with truncated preview
    assert result["msgtype"] == "m.file"
    assert len(result["body"]) < len(large_text)
    assert "io.mindroom.long_text" in result

    assert result["info"]["mimetype"] == "application/json"
    assert result["filename"] == "message-content.json"
    assert "format" not in result
    assert "formatted_body" not in result

    uploaded_data = client.uploads[0]["data"]
    uploaded_payload = json.loads(uploaded_data.read().decode("utf-8"))
    assert uploaded_payload["formatted_body"] == large_html
    assert uploaded_payload["format"] == "org.matrix.custom.html"


@pytest.mark.asyncio
async def test_large_message_with_plain_tool_markers_uploads_full_content_json() -> None:
    """Large-message sidecar stores full source content including tool trace metadata."""
    client = MockClient()

    body = "Here is the result:\n\n🔧 `web_search` [1]\n"
    body = body * 500  # Make it large enough to trigger long text
    formatted_body = "<p>Here is the result:</p>\n<p>🔧 <code>web_search</code> [1]</p>\n"
    formatted_body = formatted_body * 500

    content = {
        "body": body,
        "formatted_body": formatted_body,
        "msgtype": "m.text",
        "format": "org.matrix.custom.html",
        _TOOL_TRACE_KEY: {"version": 2, "events": [{"type": "tool_call_started", "tool_name": "web_search"}]},
    }

    result = await prepare_large_message(client, "!room:server", content)

    assert result["msgtype"] == "m.file"
    assert result["info"]["mimetype"] == "application/json"
    assert "format" not in result
    assert "formatted_body" not in result
    assert _TOOL_TRACE_KEY not in result

    # Uploaded sidecar should preserve full original content.
    uploaded_data = client.uploads[0]["data"]
    uploaded_payload = json.loads(uploaded_data.read().decode("utf-8"))
    assert uploaded_payload["formatted_body"] == formatted_body
    assert uploaded_payload[_TOOL_TRACE_KEY]["events"][0]["tool_name"] == "web_search"


@pytest.mark.asyncio
async def test_large_message_preview_uses_generic_truncation_with_plain_markers() -> None:
    """Plain-marker messages use generic truncation (no special tool-block shrinking)."""
    client = MockClient()

    # Non-tool text that should survive in full
    intro = "Important analysis result:\n" * 20  # ~520 bytes
    conclusion = "\nFinal conclusion here.\n"

    # A single huge plain-text body section that includes a visible tool marker
    tool_result = "x" * 80000
    body = f"{intro}🔧 `search` [1]\n{tool_result}\n{conclusion}"
    content = {
        "body": body,
        "msgtype": "m.text",
    }

    result = await prepare_large_message(client, "!room:server", content)

    assert result["msgtype"] == "m.file"

    # Intro text and the visible tool marker should survive in the body preview.
    assert "Important analysis result:" in result["body"]
    assert "[Message continues in attached file]" in result["body"]
    assert "🔧 `search` [1]" in result["body"]

    # Generic truncation is used now.
    assert "formatted_body" not in result


@pytest.mark.asyncio
async def test_streaming_finalize() -> None:
    """Test that streaming finalize properly handles large messages."""
    client = MockClient()
    config = MockConfig()

    streaming = StreamingResponse(
        target=MessageTarget.resolve("!test:room", None, None),
        config=config,
        runtime_paths=_runtime_paths(),
    )

    # Large content
    streaming.accumulated_text = "g" * 60000

    # Use finalize which should remove the in-progress marker
    await streaming.finalize(client)

    sent_content = client.messages_sent[0][2]

    # Should have large message handling
    assert sent_content["msgtype"] == "m.file"
    assert "io.mindroom.long_text" in sent_content
    # Should not have in-progress marker in final
    assert "⋯" not in sent_content["body"]


@pytest.mark.asyncio
async def test_structured_stream_chunk_adds_tool_trace_metadata() -> None:
    """Structured streaming chunks should preserve tool trace metadata in sent content."""
    client = MockClient()
    config = MockConfig()

    async def stream() -> AsyncIterator[StreamInputChunk]:
        trace = [ToolTraceEntry(type="tool_call_started", tool_name="save_file", args_preview="file_name=a.py")]
        yield StructuredStreamChunk(content="🔧 `save_file` [1] ⏳", tool_trace=trace)

    outcome = await send_streaming_response(
        client=client,
        target=MessageTarget.resolve("!test:room", None, None),
        config=config,
        runtime_paths=_runtime_paths(),
        response_stream=stream(),
        streaming_cls=ReplacementStreamingResponse,
    )

    event_id = outcome.last_physical_stream_event_id
    assert event_id is not None
    assert len(client.messages_sent) >= 1
    last_content = client.messages_sent[-1][2]
    target_content = last_content.get("m.new_content", last_content)
    assert _TOOL_TRACE_KEY in target_content
    assert target_content[_TOOL_TRACE_KEY]["events"][0]["tool_name"] == "save_file"


@pytest.mark.asyncio
async def test_streaming_with_extra_content_metadata() -> None:
    """Streaming sender should merge custom metadata into final event content."""
    client = MockClient()
    config = MockConfig()
    extra_content: dict[str, object] = {}

    async def stream() -> AsyncIterator[StreamInputChunk]:
        yield "hello"
        extra_content[AI_RUN_METADATA_KEY] = {"version": 1, "usage": {"total_tokens": 10}}

    outcome = await send_streaming_response(
        client=client,
        target=MessageTarget.resolve("!test:room", None, None),
        config=config,
        runtime_paths=_runtime_paths(),
        response_stream=stream(),
        streaming_cls=ReplacementStreamingResponse,
        extra_content=extra_content,
    )

    event_id = outcome.last_physical_stream_event_id
    assert event_id is not None
    target_content = client.messages_sent[-1][2].get("m.new_content", client.messages_sent[-1][2])
    assert target_content[AI_RUN_METADATA_KEY]["usage"]["total_tokens"] == 10


@pytest.mark.asyncio
async def test_structured_stream_chunk_does_not_drop_trace_on_stale_snapshot() -> None:
    """Older structured snapshots should not remove already-seen tool trace entries."""
    client = MockClient()
    config = MockConfig()

    trace_full = [
        ToolTraceEntry(type="tool_call_started", tool_name="save_file"),
        ToolTraceEntry(type="tool_call_completed", tool_name="save_file"),
    ]
    trace_stale = [ToolTraceEntry(type="tool_call_started", tool_name="save_file")]

    async def stream() -> AsyncIterator[StreamInputChunk]:
        yield StructuredStreamChunk(content="🔧 `save_file` [1]", tool_trace=trace_full)
        yield StructuredStreamChunk(content="🔧 `save_file` [1]", tool_trace=trace_stale)

    outcome = await send_streaming_response(
        client=client,
        target=MessageTarget.resolve("!test:room", None, None),
        config=config,
        runtime_paths=_runtime_paths(),
        response_stream=stream(),
        streaming_cls=ReplacementStreamingResponse,
    )

    event_id = outcome.last_physical_stream_event_id
    assert event_id is not None
    target_content = client.messages_sent[-1][2].get("m.new_content", client.messages_sent[-1][2])
    assert _TOOL_TRACE_KEY in target_content
    assert len(target_content[_TOOL_TRACE_KEY]["events"]) == 2


@pytest.mark.asyncio
async def test_replacement_streaming_preserves_text_on_tool_completion() -> None:
    """ToolCallCompletedEvent through ReplacementStreamingResponse must not wipe accumulated_text."""
    client = MockClient()
    config = MockConfig()

    tool = ToolExecution(tool_name="save_file", tool_args={"file": "a.py"}, result="ok")

    async def stream() -> AsyncIterator[StreamInputChunk]:
        yield ToolCallStartedEvent(tool=ToolExecution(tool_name="save_file", tool_args={"file": "a.py"}))
        yield ToolCallCompletedEvent(tool=tool, content="ok")

    outcome = await send_streaming_response(
        client=client,
        target=MessageTarget.resolve("!test:room", None, None),
        config=config,
        runtime_paths=_runtime_paths(),
        response_stream=stream(),
        streaming_cls=ReplacementStreamingResponse,
    )

    event_id = outcome.last_physical_stream_event_id
    accumulated = outcome.rendered_body
    assert event_id is not None
    # The accumulated text must still contain the tool marker, not be empty
    assert accumulated is not None
    assert "save_file" in accumulated
    assert accumulated.strip() != ""


@pytest.mark.asyncio
async def test_replacement_streaming_tool_start_preserves_prior_visible_text() -> None:
    """Visible tool-start markers must append to the current replacement snapshot, not replace it."""
    client = MockClient()
    config = MockConfig()

    async def stream() -> AsyncIterator[StreamInputChunk]:
        yield "hello"
        yield ToolCallStartedEvent(tool=ToolExecution(tool_name="save_file", tool_args={"file": "a.py"}))

    outcome = await send_streaming_response(
        client=client,
        target=MessageTarget.resolve("!test:room", None, None),
        config=config,
        runtime_paths=_runtime_paths(),
        response_stream=stream(),
        streaming_cls=ReplacementStreamingResponse,
    )
    accumulated = outcome.canonical_final_body_candidate or outcome.rendered_body or ""

    assert outcome.last_physical_stream_event_id is not None
    assert accumulated.startswith("hello")
    assert "save_file" in accumulated

    target_content = client.messages_sent[-1][2].get("m.new_content", client.messages_sent[-1][2])
    assert target_content["body"].startswith("hello")
    assert "save_file" in target_content["body"]


@pytest.mark.asyncio
async def test_replacement_streaming_preserves_visible_tool_marker_across_snapshots() -> None:
    """Replacement snapshots should not wipe a pending visible tool marker before completion."""
    client = MockClient()
    config = MockConfig()

    tool = ToolExecution(tool_name="save_file", tool_args={"file": "a.py"}, result="ok")

    async def stream() -> AsyncIterator[StreamInputChunk]:
        yield "hello"
        yield ToolCallStartedEvent(tool=ToolExecution(tool_name="save_file", tool_args={"file": "a.py"}))
        yield "hello world"
        yield ToolCallCompletedEvent(tool=tool, content="ok")

    outcome = await send_streaming_response(
        client=client,
        target=MessageTarget.resolve("!test:room", None, None),
        config=config,
        runtime_paths=_runtime_paths(),
        response_stream=stream(),
        streaming_cls=ReplacementStreamingResponse,
    )
    accumulated = outcome.canonical_final_body_candidate or outcome.rendered_body or ""

    assert outcome.last_physical_stream_event_id is not None
    assert accumulated.startswith("hello world")
    assert "save_file" in accumulated
    assert "⏳" not in accumulated

    target_content = client.messages_sent[-1][2].get("m.new_content", client.messages_sent[-1][2])
    assert target_content["body"].startswith("hello world")
    assert "save_file" in target_content["body"]
    assert "⏳" not in target_content["body"]


@pytest.mark.asyncio
async def test_hidden_tool_calls_coalesce_placeholder_spacing() -> None:
    """Hidden tool calls should not stack repeated blank-line placeholders."""
    client = MockClient()
    config = MockConfig()

    async def stream() -> AsyncIterator[StreamInputChunk]:
        yield ToolCallStartedEvent(tool=ToolExecution(tool_name="first_tool", tool_args={}))
        await asyncio.sleep(0)
        yield ToolCallStartedEvent(tool=ToolExecution(tool_name="second_tool", tool_args={}))
        await asyncio.sleep(0)
        yield "Done"

    outcome = await send_streaming_response(
        client=client,
        target=MessageTarget.resolve("!test:room", None, None),
        config=config,
        runtime_paths=_runtime_paths(),
        response_stream=stream(),
        show_tool_calls=False,
    )

    event_id = outcome.last_physical_stream_event_id
    accumulated = outcome.rendered_body
    assert event_id is not None
    assert accumulated is not None
    assert accumulated == "\n\nDone"
    bodies = [content.get("m.new_content", content)["body"] for _, _, content in client.messages_sent]
    assert bodies == ["Thinking...", "\n\nDone"]
