"""Regression coverage for provider-safe transient context compression."""

from langchain.agents.middleware.types import ModelRequest, ModelResponse
from langchain_core.language_models.fake_chat_models import FakeListChatModel
from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage

from deerflow.agents.middlewares.summarization_middleware import DeerFlowSummarizationMiddleware, _ensure_latest_human_message


def test_compression_restores_latest_human_turn_when_only_tool_tail_remains():
    user = HumanMessage(content="请分析这份材料", id="user-1")
    source = [
        user,
        AIMessage(content="", id="ai-1"),
        ToolMessage(content="检索结果", tool_call_id="call-1", id="tool-1"),
    ]
    compressed = [SystemMessage(content="internal summary"), *source[1:]]

    result = _ensure_latest_human_message(source, compressed)

    assert result[0].type == "system"
    assert result[1] is user
    assert [message.type for message in result[2:]] == ["ai", "tool"]


def test_compression_does_not_duplicate_human_turn_already_in_tail():
    user = HumanMessage(content="继续", id="user-1")
    compressed = [SystemMessage(content="internal summary"), user]

    result = _ensure_latest_human_message([user], compressed)

    assert result == compressed


def test_compression_uses_single_model_request_system_slot_for_generic_adapters():
    model = FakeListChatModel(responses=["condensed history"])
    middleware = DeerFlowSummarizationMiddleware(
        model,
        trigger=("messages", 3),
        keep=("messages", 1),
    )
    request = ModelRequest(
        model=model,
        system_message=SystemMessage(content="agent prompt"),
        messages=[
            HumanMessage(content="first question", id="human-1"),
            AIMessage(content="first answer", id="ai-1"),
            HumanMessage(content="continue", id="human-2"),
        ],
    )
    captured: list[ModelRequest] = []

    def handler(model_request: ModelRequest) -> ModelResponse:
        captured.append(model_request)
        return ModelResponse(result=[AIMessage(content="ok")])

    middleware.wrap_model_call(request, handler)

    assert len(captured) == 1
    normalized = captured[0]
    assert normalized.system_message is not None
    assert "agent prompt" in normalized.system_message.text
    assert "condensed history" in normalized.system_message.text
    assert all(not isinstance(message, SystemMessage) for message in normalized.messages)
    assert [message.type for message in normalized.messages] == ["human"]
    assert normalized.messages[0].content == "continue"