"""Unit tests for baseline service pure-logic helpers.

These tests cover ``_baseline_conversation_updater`` and ``_BaselineStreamState``
without requiring API keys, database connections, or network access.
"""

from unittest.mock import AsyncMock, MagicMock, patch

import pytest
from openai.types.chat import ChatCompletionToolParam

from backend.copilot.baseline.service import (
    _BUDGET_EXHAUSTED_FALLBACK_TEXT,
    _NATURAL_FINISH_EMPTY_FALLBACK_TEXT,
    _apply_skills_cache_breakpoint,
    _baseline_conversation_updater,
    _baseline_llm_caller,
    _BaselineStreamState,
    _budget_exhausted_notice_text,
    _build_budget_exhausted_fallback_events,
    _build_cached_system_message,
    _build_natural_finish_empty_fallback_events,
    _compress_session_messages,
    _fresh_anthropic_caching_headers,
    _fresh_ephemeral_cache_control,
    _is_anthropic_model,
    _mark_system_message_with_cache_control,
    _mark_tools_with_cache_control,
    _natural_finish_empty_notice_text,
    _split_user_message_after_skills_block,
    _supports_prompt_cache_markers,
)
from backend.copilot.model import ChatMessage
from backend.copilot.response_model import (
    StreamReasoningDelta,
    StreamReasoningEnd,
    StreamReasoningStart,
    StreamTextDelta,
    StreamTextEnd,
    StreamTextStart,
)
from backend.copilot.token_tracking import _extract_cache_creation_tokens
from backend.copilot.transcript_builder import TranscriptBuilder
from backend.util.prompt import CompressResult
from backend.util.tool_call_loop import LLMLoopResponse, LLMToolCall, ToolCallResult


class TestBaselineStreamState:
    def test_defaults(self):
        state = _BaselineStreamState()
        # ``pending_events`` is an asyncio.Queue now (live SSE channel).
        # The durable inspection view is ``emitted_events``.
        assert state.pending_events.empty()
        assert state.emitted_events == []
        assert state.assistant_text == ""
        assert state.text_started is False
        assert state.turn_prompt_tokens == 0
        assert state.turn_completion_tokens == 0
        assert state.text_block_id  # Should be a UUID string

    def test_mutable_fields(self):
        state = _BaselineStreamState()
        state.assistant_text = "hello"
        state.turn_prompt_tokens = 100
        state.turn_completion_tokens = 50
        assert state.assistant_text == "hello"
        assert state.turn_prompt_tokens == 100
        assert state.turn_completion_tokens == 50


class TestBaselineConversationUpdater:
    """Tests for _baseline_conversation_updater which updates the OpenAI
    message list and transcript builder after each LLM call."""

    def _make_transcript_builder(self) -> TranscriptBuilder:
        builder = TranscriptBuilder()
        builder.append_user("test question")
        return builder

    def test_text_only_response(self):
        """When the LLM returns text without tool calls, the updater appends
        a single assistant message and records it in the transcript."""
        messages: list = []
        builder = self._make_transcript_builder()
        response = LLMLoopResponse(
            response_text="Hello, world!",
            tool_calls=[],
            raw_response=None,
            prompt_tokens=0,
            completion_tokens=0,
        )

        _baseline_conversation_updater(
            messages,
            response,
            tool_results=None,
            transcript_builder=builder,
            model="test-model",
        )

        assert len(messages) == 1
        assert messages[0]["role"] == "assistant"
        assert messages[0]["content"] == "Hello, world!"
        # Transcript should have user + assistant
        assert builder.entry_count == 2
        assert builder.last_entry_type == "assistant"

    def test_tool_calls_response(self):
        """When the LLM returns tool calls, the updater appends the assistant
        message with tool_calls and tool result messages."""
        messages: list = []
        builder = self._make_transcript_builder()
        response = LLMLoopResponse(
            response_text="Let me search...",
            tool_calls=[
                LLMToolCall(
                    id="tc_1",
                    name="search",
                    arguments='{"query": "test"}',
                ),
            ],
            raw_response=None,
            prompt_tokens=0,
            completion_tokens=0,
        )
        tool_results = [
            ToolCallResult(
                tool_call_id="tc_1",
                tool_name="search",
                content="Found result",
            ),
        ]

        _baseline_conversation_updater(
            messages,
            response,
            tool_results=tool_results,
            transcript_builder=builder,
            model="test-model",
        )

        # Messages: assistant (with tool_calls) + tool result
        assert len(messages) == 2
        assert messages[0]["role"] == "assistant"
        assert messages[0]["content"] == "Let me search..."
        assert len(messages[0]["tool_calls"]) == 1
        assert messages[0]["tool_calls"][0]["id"] == "tc_1"
        assert messages[1]["role"] == "tool"
        assert messages[1]["tool_call_id"] == "tc_1"
        assert messages[1]["content"] == "Found result"

        # Transcript: user + assistant(tool_use) + user(tool_result)
        assert builder.entry_count == 3

    def test_tool_calls_without_text(self):
        """Tool calls without accompanying text should still work."""
        messages: list = []
        builder = self._make_transcript_builder()
        response = LLMLoopResponse(
            response_text=None,
            tool_calls=[
                LLMToolCall(id="tc_1", name="run", arguments="{}"),
            ],
            raw_response=None,
            prompt_tokens=0,
            completion_tokens=0,
        )
        tool_results = [
            ToolCallResult(tool_call_id="tc_1", tool_name="run", content="done"),
        ]

        _baseline_conversation_updater(
            messages,
            response,
            tool_results=tool_results,
            transcript_builder=builder,
            model="test-model",
        )

        assert len(messages) == 2
        assert "content" not in messages[0]  # No text content
        assert messages[0]["tool_calls"][0]["function"]["name"] == "run"

    def test_no_text_no_tools(self):
        """When the response has no text and no tool calls, nothing is appended."""
        messages: list = []
        builder = self._make_transcript_builder()
        response = LLMLoopResponse(
            response_text=None,
            tool_calls=[],
            raw_response=None,
            prompt_tokens=0,
            completion_tokens=0,
        )

        _baseline_conversation_updater(
            messages,
            response,
            tool_results=None,
            transcript_builder=builder,
            model="test-model",
        )

        assert len(messages) == 0
        # Only the user entry from setup
        assert builder.entry_count == 1

    def test_multiple_tool_calls(self):
        """Multiple tool calls in a single response are all recorded."""
        messages: list = []
        builder = self._make_transcript_builder()
        response = LLMLoopResponse(
            response_text=None,
            tool_calls=[
                LLMToolCall(id="tc_1", name="tool_a", arguments="{}"),
                LLMToolCall(id="tc_2", name="tool_b", arguments='{"x": 1}'),
            ],
            raw_response=None,
            prompt_tokens=0,
            completion_tokens=0,
        )
        tool_results = [
            ToolCallResult(tool_call_id="tc_1", tool_name="tool_a", content="result_a"),
            ToolCallResult(tool_call_id="tc_2", tool_name="tool_b", content="result_b"),
        ]

        _baseline_conversation_updater(
            messages,
            response,
            tool_results=tool_results,
            transcript_builder=builder,
            model="test-model",
        )

        # 1 assistant + 2 tool results
        assert len(messages) == 3
        assert len(messages[0]["tool_calls"]) == 2
        assert messages[1]["tool_call_id"] == "tc_1"
        assert messages[2]["tool_call_id"] == "tc_2"

    def test_invalid_tool_arguments_handled(self):
        """Tool call with invalid JSON arguments: the arguments field is
        stored as-is in the message, and orjson failure falls back to {}
        in the transcript content_blocks."""
        messages: list = []
        builder = self._make_transcript_builder()
        response = LLMLoopResponse(
            response_text=None,
            tool_calls=[
                LLMToolCall(id="tc_1", name="tool_x", arguments="not-json"),
            ],
            raw_response=None,
            prompt_tokens=0,
            completion_tokens=0,
        )
        tool_results = [
            ToolCallResult(tool_call_id="tc_1", tool_name="tool_x", content="ok"),
        ]

        _baseline_conversation_updater(
            messages,
            response,
            tool_results=tool_results,
            transcript_builder=builder,
            model="test-model",
        )

        # Should not raise — invalid JSON falls back to {} in transcript
        assert len(messages) == 2
        assert messages[0]["tool_calls"][0]["function"]["arguments"] == "not-json"


class TestCompressSessionMessagesPreservesToolCalls:
    """``_compress_session_messages`` must round-trip tool_calls + tool_call_id.

    Compression serialises ChatMessage to dict for ``compress_context`` and
    reifies the result back to ChatMessage.  A regression that drops
    ``tool_calls`` or ``tool_call_id`` would corrupt the OpenAI message
    list and break downstream tool-execution rounds.
    """

    @pytest.mark.asyncio
    async def test_compressed_output_keeps_tool_calls_and_ids(self):
        # Simulate compression that returns a summary + the most recent
        # assistant(tool_call) + tool(tool_result) intact.
        summary = {"role": "system", "content": "prior turns: user asked X"}
        assistant_with_tc = {
            "role": "assistant",
            "content": "calling tool",
            "tool_calls": [
                {
                    "id": "tc_abc",
                    "type": "function",
                    "function": {"name": "search", "arguments": '{"q":"y"}'},
                }
            ],
        }
        tool_result = {
            "role": "tool",
            "tool_call_id": "tc_abc",
            "content": "search result",
        }

        compress_result = CompressResult(
            messages=[summary, assistant_with_tc, tool_result],
            token_count=100,
            was_compacted=True,
            original_token_count=5000,
            messages_summarized=10,
            messages_dropped=0,
        )

        # Input: messages that should be compressed.
        input_messages = [
            ChatMessage(role="user", content="q1"),
            ChatMessage(
                role="assistant",
                content="calling tool",
                tool_calls=[
                    {
                        "id": "tc_abc",
                        "type": "function",
                        "function": {
                            "name": "search",
                            "arguments": '{"q":"y"}',
                        },
                    }
                ],
            ),
            ChatMessage(
                role="tool",
                tool_call_id="tc_abc",
                content="search result",
            ),
        ]

        with patch(
            "backend.copilot.baseline.service.compress_context",
            new=AsyncMock(return_value=compress_result),
        ):
            compressed = await _compress_session_messages(
                input_messages, model="openrouter/anthropic/claude-opus-4"
            )

        # Summary, assistant(tool_calls), tool(tool_call_id).
        assert len(compressed) == 3
        # Assistant message must keep its tool_calls intact.
        assistant_msg = compressed[1]
        assert assistant_msg.role == "assistant"
        assert assistant_msg.tool_calls is not None
        assert len(assistant_msg.tool_calls) == 1
        assert assistant_msg.tool_calls[0]["id"] == "tc_abc"
        assert assistant_msg.tool_calls[0]["function"]["name"] == "search"
        # Tool-role message must keep tool_call_id for OpenAI linkage.
        tool_msg = compressed[2]
        assert tool_msg.role == "tool"
        assert tool_msg.tool_call_id == "tc_abc"
        assert tool_msg.content == "search result"

    @pytest.mark.asyncio
    async def test_uncompressed_passthrough_keeps_fields(self):
        """When compression is a no-op (was_compacted=False), the original
        messages must be returned unchanged — including tool_calls."""
        input_messages = [
            ChatMessage(
                role="assistant",
                content="c",
                tool_calls=[
                    {
                        "id": "t1",
                        "type": "function",
                        "function": {"name": "f", "arguments": "{}"},
                    }
                ],
            ),
            ChatMessage(role="tool", tool_call_id="t1", content="ok"),
        ]

        noop_result = CompressResult(
            messages=[],  # ignored when was_compacted=False
            token_count=10,
            was_compacted=False,
        )

        with patch(
            "backend.copilot.baseline.service.compress_context",
            new=AsyncMock(return_value=noop_result),
        ):
            out = await _compress_session_messages(
                input_messages, model="openrouter/anthropic/claude-opus-4"
            )

        assert out is input_messages  # same list returned
        assert out[0].tool_calls is not None
        assert out[0].tool_calls[0]["id"] == "t1"
        assert out[1].tool_call_id == "t1"


# ---- _filter_tools_by_permissions tests ---- #


def _make_tool(name: str) -> ChatCompletionToolParam:
    """Build a minimal OpenAI ChatCompletionToolParam."""
    return ChatCompletionToolParam(
        type="function",
        function={"name": name, "parameters": {}},
    )


class TestFilterToolsByPermissions:
    """Tests for _filter_tools_by_permissions."""

    @patch(
        "backend.copilot.permissions.all_known_tool_names",
        return_value=frozenset({"run_block", "web_fetch", "bash_exec"}),
    )
    def test_empty_permissions_returns_all(self, _mock_names):
        """Empty permissions (no filtering) returns every tool unchanged."""
        from backend.copilot.baseline.service import _filter_tools_by_permissions
        from backend.copilot.permissions import CopilotPermissions

        tools = [_make_tool("run_block"), _make_tool("web_fetch")]
        perms = CopilotPermissions()
        result = _filter_tools_by_permissions(tools, perms)
        assert result == tools

    @patch(
        "backend.copilot.permissions.all_known_tool_names",
        return_value=frozenset({"run_block", "web_fetch", "bash_exec"}),
    )
    def test_allowlist_keeps_only_matching(self, _mock_names):
        """Explicit allowlist (tools_exclude=False) keeps only listed tools."""
        from backend.copilot.baseline.service import _filter_tools_by_permissions
        from backend.copilot.permissions import CopilotPermissions

        tools = [
            _make_tool("run_block"),
            _make_tool("web_fetch"),
            _make_tool("bash_exec"),
        ]
        perms = CopilotPermissions(tools=["web_fetch"], tools_exclude=False)
        result = _filter_tools_by_permissions(tools, perms)
        assert len(result) == 1
        assert result[0]["function"]["name"] == "web_fetch"

    @patch(
        "backend.copilot.permissions.all_known_tool_names",
        return_value=frozenset({"run_block", "web_fetch", "bash_exec"}),
    )
    def test_blacklist_excludes_listed(self, _mock_names):
        """Blacklist (tools_exclude=True) removes only the listed tools."""
        from backend.copilot.baseline.service import _filter_tools_by_permissions
        from backend.copilot.permissions import CopilotPermissions

        tools = [
            _make_tool("run_block"),
            _make_tool("web_fetch"),
            _make_tool("bash_exec"),
        ]
        perms = CopilotPermissions(tools=["bash_exec"], tools_exclude=True)
        result = _filter_tools_by_permissions(tools, perms)
        names = [t["function"]["name"] for t in result]
        assert "bash_exec" not in names
        assert "run_block" in names
        assert "web_fetch" in names
        assert len(result) == 2

    @patch(
        "backend.copilot.permissions.all_known_tool_names",
        return_value=frozenset({"run_block", "web_fetch", "bash_exec"}),
    )
    def test_unknown_tool_name_filtered_out(self, _mock_names):
        """A tool whose name is not in all_known_tool_names is dropped."""
        from backend.copilot.baseline.service import _filter_tools_by_permissions
        from backend.copilot.permissions import CopilotPermissions

        tools = [_make_tool("run_block"), _make_tool("unknown_tool")]
        perms = CopilotPermissions(tools=["run_block"], tools_exclude=False)
        result = _filter_tools_by_permissions(tools, perms)
        names = [t["function"]["name"] for t in result]
        assert "unknown_tool" not in names
        assert names == ["run_block"]


# ---- _prepare_baseline_attachments tests ---- #


class TestPrepareBaselineAttachments:
    """Tests for _prepare_baseline_attachments."""

    @pytest.mark.asyncio
    async def test_empty_file_ids(self):
        """Empty file_ids returns empty hint and blocks."""
        from backend.copilot.baseline.service import _prepare_baseline_attachments

        hint, blocks = await _prepare_baseline_attachments([], "user1", "sess1", "/tmp")
        assert hint == ""
        assert blocks == []

    @pytest.mark.asyncio
    async def test_empty_user_id(self):
        """Empty user_id returns empty hint and blocks."""
        from backend.copilot.baseline.service import _prepare_baseline_attachments

        hint, blocks = await _prepare_baseline_attachments(
            ["file1"], "", "sess1", "/tmp"
        )
        assert hint == ""
        assert blocks == []

    @pytest.mark.asyncio
    async def test_image_file_returns_vision_blocks(self):
        """A PNG image within size limits is returned as a base64 vision block."""
        from backend.copilot.baseline.service import _prepare_baseline_attachments

        fake_info = AsyncMock()
        fake_info.name = "photo.png"
        fake_info.mime_type = "image/png"
        fake_info.size_bytes = 1024

        fake_manager = AsyncMock()
        fake_manager.get_file_info = AsyncMock(return_value=fake_info)
        fake_manager.read_file_by_id = AsyncMock(return_value=b"\x89PNG_FAKE_DATA")

        with patch(
            "backend.copilot.baseline.service.get_workspace_manager",
            new=AsyncMock(return_value=fake_manager),
        ):
            hint, blocks = await _prepare_baseline_attachments(
                ["fid1"], "user1", "sess1", "/tmp/workdir"
            )

        assert len(blocks) == 1
        assert blocks[0]["type"] == "image"
        assert blocks[0]["source"]["media_type"] == "image/png"
        assert blocks[0]["source"]["type"] == "base64"
        assert "photo.png" in hint
        assert "embedded as image" in hint

    @pytest.mark.asyncio
    async def test_non_image_file_saved_to_working_dir(self, tmp_path):
        """A non-image file is written to working_dir."""
        from backend.copilot.baseline.service import _prepare_baseline_attachments

        fake_info = AsyncMock()
        fake_info.name = "data.csv"
        fake_info.mime_type = "text/csv"
        fake_info.size_bytes = 42

        fake_manager = AsyncMock()
        fake_manager.get_file_info = AsyncMock(return_value=fake_info)
        fake_manager.read_file_by_id = AsyncMock(return_value=b"col1,col2\na,b")

        with patch(
            "backend.copilot.baseline.service.get_workspace_manager",
            new=AsyncMock(return_value=fake_manager),
        ):
            hint, blocks = await _prepare_baseline_attachments(
                ["fid1"], "user1", "sess1", str(tmp_path)
            )

        assert blocks == []
        assert "data.csv" in hint
        assert "saved to" in hint
        saved = tmp_path / "data.csv"
        assert saved.exists()
        assert saved.read_bytes() == b"col1,col2\na,b"

    @pytest.mark.asyncio
    async def test_file_not_found_skipped(self):
        """When get_file_info returns None the file is silently skipped."""
        from backend.copilot.baseline.service import _prepare_baseline_attachments

        fake_manager = AsyncMock()
        fake_manager.get_file_info = AsyncMock(return_value=None)

        with patch(
            "backend.copilot.baseline.service.get_workspace_manager",
            new=AsyncMock(return_value=fake_manager),
        ):
            hint, blocks = await _prepare_baseline_attachments(
                ["missing_id"], "user1", "sess1", "/tmp"
            )

        assert hint == ""
        assert blocks == []

    @pytest.mark.asyncio
    async def test_workspace_manager_error(self):
        """When get_workspace_manager raises, returns empty results."""
        from backend.copilot.baseline.service import _prepare_baseline_attachments

        with patch(
            "backend.copilot.baseline.service.get_workspace_manager",
            new=AsyncMock(side_effect=RuntimeError("connection failed")),
        ):
            hint, blocks = await _prepare_baseline_attachments(
                ["fid1"], "user1", "sess1", "/tmp"
            )

        assert hint == ""
        assert blocks == []


_COST_MISSING = object()


def _make_usage_chunk(
    *,
    prompt_tokens: int = 0,
    completion_tokens: int = 0,
    cost: float | str | None | object = _COST_MISSING,
    cached_tokens: int | None = None,
    cache_creation_input_tokens: int | None = None,
):
    """Build a mock streaming chunk carrying usage (and optionally cost).

    Provider-specific fields (``cost`` on usage, ``cache_creation_input_tokens``
    on prompt_tokens_details) are set on ``model_extra`` because that's where
    the baseline helper reads them from (typed ``CompletionUsage.model_extra``
    rather than ``getattr``). Pass ``cost=None`` to emit an explicit-null cost
    key; omit ``cost`` entirely to leave the key absent.
    """
    chunk = MagicMock()
    chunk.choices = []
    chunk.usage = MagicMock()
    chunk.usage.prompt_tokens = prompt_tokens
    chunk.usage.completion_tokens = completion_tokens
    usage_extras: dict[str, float | str | None] = {}
    if cost is not _COST_MISSING:
        usage_extras["cost"] = cost  # type: ignore[assignment]
    chunk.usage.model_extra = usage_extras

    if cached_tokens is not None or cache_creation_input_tokens is not None:
        # Build a real ``PromptTokensDetails`` so ``getattr(ptd,
        # "cache_write_tokens", None)`` returns ``None`` on this SDK version
        # (rather than a truthy MagicMock attribute) and the extraction
        # helper's typed-attr vs model_extra fallback resolves correctly.
        from openai.types.completion_usage import PromptTokensDetails

        ptd = PromptTokensDetails.model_validate({"cached_tokens": cached_tokens or 0})
        if cache_creation_input_tokens is not None:
            if ptd.model_extra is None:
                object.__setattr__(ptd, "__pydantic_extra__", {})
            assert ptd.model_extra is not None
            ptd.model_extra["cache_creation_input_tokens"] = cache_creation_input_tokens
        chunk.usage.prompt_tokens_details = ptd
    else:
        chunk.usage.prompt_tokens_details = None

    return chunk


def _make_stream_mock(*chunks):
    """Build an async streaming response mock that yields *chunks* in order."""
    stream = MagicMock()
    stream.close = AsyncMock()

    async def aiter():
        for c in chunks:
            yield c

    stream.__aiter__ = lambda self: aiter()
    return stream


class TestBaselineCostExtraction:
    """Tests for ``usage.cost`` extraction in ``_baseline_llm_caller``.

    Cost is read from the OpenRouter ``usage.cost`` field on the final
    streaming chunk when the request body includes ``usage: {include: true}``
    (handled by the baseline service via ``extra_body``).
    """

    @pytest.mark.asyncio
    async def test_cost_usd_extracted_from_usage_chunk(self):
        """state.cost_usd is set from chunk.usage.cost when present."""
        state = _BaselineStreamState(model="gpt-4o-mini")
        chunk = _make_usage_chunk(
            prompt_tokens=1000, completion_tokens=200, cost=0.0123
        )

        mock_client = MagicMock()
        mock_client.chat.completions.create = AsyncMock(
            return_value=_make_stream_mock(chunk)
        )

        with patch(
            "backend.copilot.baseline.service._get_main_client",
            return_value=mock_client,
        ):
            await _baseline_llm_caller(
                messages=[{"role": "user", "content": "hi"}],
                tools=[],
                state=state,
            )

        assert state.cost_usd == pytest.approx(0.0123)

    @pytest.mark.asyncio
    async def test_cost_usd_accumulates_across_calls(self):
        """cost_usd accumulates when _baseline_llm_caller is called multiple times."""
        state = _BaselineStreamState(model="gpt-4o-mini")

        mock_client = MagicMock()
        mock_client.chat.completions.create = AsyncMock(
            side_effect=[
                _make_stream_mock(_make_usage_chunk(prompt_tokens=500, cost=0.01)),
                _make_stream_mock(_make_usage_chunk(prompt_tokens=600, cost=0.02)),
            ]
        )

        with patch(
            "backend.copilot.baseline.service._get_main_client",
            return_value=mock_client,
        ):
            await _baseline_llm_caller(
                messages=[{"role": "user", "content": "first"}],
                tools=[],
                state=state,
            )
            await _baseline_llm_caller(
                messages=[{"role": "user", "content": "second"}],
                tools=[],
                state=state,
            )

        assert state.cost_usd == pytest.approx(0.03)

    @pytest.mark.asyncio
    async def test_cost_usd_accepts_string_value(self):
        """OpenRouter may emit cost as a string — it should still parse."""
        state = _BaselineStreamState(model="gpt-4o-mini")
        chunk = _make_usage_chunk(prompt_tokens=10, cost="0.005")

        mock_client = MagicMock()
        mock_client.chat.completions.create = AsyncMock(
            return_value=_make_stream_mock(chunk)
        )

        with patch(
            "backend.copilot.baseline.service._get_main_client",
            return_value=mock_client,
        ):
            await _baseline_llm_caller(
                messages=[{"role": "user", "content": "hi"}],
                tools=[],
                state=state,
            )

        assert state.cost_usd == pytest.approx(0.005)

    @pytest.mark.asyncio
    async def test_direct_mode_falls_back_to_rate_card_when_cost_missing(self):
        """In direct-Anthropic mode the OAI-compat chunk has no ``cost`` field
        (OpenRouter extension), so cost is computed locally from tokens ×
        rates via ``compute_anthropic_cost_usd``.  state.cost_usd ends up
        with a positive number rather than None."""
        state = _BaselineStreamState(model="anthropic/claude-sonnet-4")
        chunk = _make_usage_chunk(prompt_tokens=1000, completion_tokens=500)

        mock_client = MagicMock()
        mock_client.chat.completions.create = AsyncMock(
            return_value=_make_stream_mock(chunk)
        )

        with (
            patch(
                "backend.copilot.baseline.service._get_main_client",
                return_value=mock_client,
            ),
            patch(
                "backend.copilot.baseline.service.config.use_openrouter",
                False,
            ),
        ):
            await _baseline_llm_caller(
                messages=[{"role": "user", "content": "hi"}],
                tools=[],
                state=state,
            )

        assert state.cost_usd is not None
        assert state.cost_usd > 0
        # Token accumulators are still populated so the caller can log them.
        assert state.turn_prompt_tokens == 1000
        assert state.turn_completion_tokens == 500

    @pytest.mark.asyncio
    async def test_invalid_cost_string_leaves_cost_none(self):
        """A non-numeric cost value is rejected without raising."""
        state = _BaselineStreamState(model="gpt-4o-mini")
        chunk = _make_usage_chunk(prompt_tokens=10, cost="not-a-number")

        mock_client = MagicMock()
        mock_client.chat.completions.create = AsyncMock(
            return_value=_make_stream_mock(chunk)
        )

        with patch(
            "backend.copilot.baseline.service._get_main_client",
            return_value=mock_client,
        ):
            await _baseline_llm_caller(
                messages=[{"role": "user", "content": "hi"}],
                tools=[],
                state=state,
            )

        assert state.cost_usd is None

    @pytest.mark.asyncio
    async def test_negative_cost_is_ignored(self):
        """Guard against negative cost values (shouldn't happen but be safe)."""
        state = _BaselineStreamState(model="gpt-4o-mini")
        chunk = _make_usage_chunk(prompt_tokens=10, cost=-0.01)

        mock_client = MagicMock()
        mock_client.chat.completions.create = AsyncMock(
            return_value=_make_stream_mock(chunk)
        )

        with patch(
            "backend.copilot.baseline.service._get_main_client",
            return_value=mock_client,
        ):
            await _baseline_llm_caller(
                messages=[{"role": "user", "content": "hi"}],
                tools=[],
                state=state,
            )

        assert state.cost_usd is None

    @pytest.mark.asyncio
    async def test_explicit_null_cost_is_logged_and_ignored(self, caplog):
        """`{"cost": null}` is rejected and logged (not silently dropped)."""
        state = _BaselineStreamState(model="openrouter/auto")
        chunk = _make_usage_chunk(prompt_tokens=10, cost=None)

        mock_client = MagicMock()
        mock_client.chat.completions.create = AsyncMock(
            return_value=_make_stream_mock(chunk)
        )

        with (
            patch(
                "backend.copilot.baseline.service._get_main_client",
                return_value=mock_client,
            ),
            caplog.at_level("ERROR", logger="backend.copilot.baseline.service"),
        ):
            await _baseline_llm_caller(
                messages=[{"role": "user", "content": "hi"}],
                tools=[],
                state=state,
            )

        assert state.cost_usd is None
        assert any(
            "usage.cost is present but null" in rec.message for rec in caplog.records
        )

    @pytest.mark.asyncio
    async def test_cost_not_captured_when_stream_raises_mid_chunk(self):
        """If the stream aborts before emitting the usage chunk there is no cost."""
        state = _BaselineStreamState(model="gpt-4o-mini")

        stream = MagicMock()
        stream.close = AsyncMock()

        async def failing_aiter():
            raise RuntimeError("stream error")
            yield  # make it an async generator

        stream.__aiter__ = lambda self: failing_aiter()

        mock_client = MagicMock()
        mock_client.chat.completions.create = AsyncMock(return_value=stream)

        with (
            patch(
                "backend.copilot.baseline.service._get_main_client",
                return_value=mock_client,
            ),
            pytest.raises(RuntimeError, match="stream error"),
        ):
            await _baseline_llm_caller(
                messages=[{"role": "user", "content": "hi"}],
                tools=[],
                state=state,
            )

        # Stream aborted before yielding the usage chunk — cost stays None.
        assert state.cost_usd is None

    @pytest.mark.asyncio
    async def test_no_cost_when_api_call_raises_before_stream(self):
        """The helper is safe when the create() call itself raises."""
        state = _BaselineStreamState(model="gpt-4o-mini")

        mock_client = MagicMock()
        mock_client.chat.completions.create = AsyncMock(
            side_effect=RuntimeError("connection refused")
        )

        with (
            patch(
                "backend.copilot.baseline.service._get_main_client",
                return_value=mock_client,
            ),
            pytest.raises(RuntimeError, match="connection refused"),
        ):
            await _baseline_llm_caller(
                messages=[{"role": "user", "content": "hi"}],
                tools=[],
                state=state,
            )

        assert state.cost_usd is None

    @pytest.mark.asyncio
    async def test_cache_tokens_extracted_from_usage_details(self):
        """cache tokens are extracted from prompt_tokens_details.cached_tokens."""
        state = _BaselineStreamState(model="openai/gpt-4o")
        chunk = _make_usage_chunk(
            prompt_tokens=1000,
            completion_tokens=200,
            cost=0.01,
            cached_tokens=800,
        )

        mock_client = MagicMock()
        mock_client.chat.completions.create = AsyncMock(
            return_value=_make_stream_mock(chunk)
        )

        with patch(
            "backend.copilot.baseline.service._get_main_client",
            return_value=mock_client,
        ):
            await _baseline_llm_caller(
                messages=[{"role": "user", "content": "hi"}],
                tools=[],
                state=state,
            )

        assert state.turn_cache_read_tokens == 800
        assert state.turn_prompt_tokens == 1000

    @pytest.mark.asyncio
    async def test_cache_creation_tokens_extracted_from_usage_details(self):
        """cache_creation_input_tokens is extracted from prompt_tokens_details."""
        state = _BaselineStreamState(model="openai/gpt-4o")
        chunk = _make_usage_chunk(
            prompt_tokens=1000,
            completion_tokens=200,
            cost=0.01,
            cached_tokens=0,
            cache_creation_input_tokens=500,
        )

        mock_client = MagicMock()
        mock_client.chat.completions.create = AsyncMock(
            return_value=_make_stream_mock(chunk)
        )

        with patch(
            "backend.copilot.baseline.service._get_main_client",
            return_value=mock_client,
        ):
            await _baseline_llm_caller(
                messages=[{"role": "user", "content": "hi"}],
                tools=[],
                state=state,
            )

        assert state.turn_cache_creation_tokens == 500

    @pytest.mark.asyncio
    async def test_token_accumulators_track_across_multiple_calls(self):
        """Token accumulators grow correctly across multiple _baseline_llm_caller calls."""
        state = _BaselineStreamState(model="anthropic/claude-sonnet-4")

        mock_client = MagicMock()
        mock_client.chat.completions.create = AsyncMock(
            side_effect=[
                _make_stream_mock(
                    _make_usage_chunk(prompt_tokens=1000, completion_tokens=200)
                ),
                _make_stream_mock(
                    _make_usage_chunk(prompt_tokens=1100, completion_tokens=300)
                ),
            ]
        )

        with (
            patch(
                "backend.copilot.baseline.service._get_main_client",
                return_value=mock_client,
            ),
            patch(
                "backend.copilot.baseline.service.config.use_openrouter",
                False,
            ),
        ):
            await _baseline_llm_caller(
                messages=[{"role": "user", "content": "hi"}],
                tools=[],
                state=state,
            )
            await _baseline_llm_caller(
                messages=[{"role": "user", "content": "follow up"}],
                tools=[],
                state=state,
            )

        # In direct mode, missing usage.cost falls through to the rate-card
        # path so cost is computed locally and accumulates across both calls.
        assert state.cost_usd is not None
        assert state.cost_usd > 0
        assert state.turn_prompt_tokens == 2100
        assert state.turn_completion_tokens == 500

    @pytest.mark.parametrize(
        "tools",
        [
            pytest.param([], id="no_tools"),
            pytest.param([_make_tool("search")], id="with_tools"),
        ],
    )
    @pytest.mark.asyncio
    async def test_baseline_requests_usage_include_extra_body(
        self, tools: list[ChatCompletionToolParam]
    ):
        """The baseline call must pass extra_body={'usage': {'include': True}}.

        This guards the contract with OpenRouter that triggers inclusion of
        the authoritative cost on the final usage chunk. Without it the
        rate-limit counter stays at zero. Exercise both the no-tools and
        tool-calling branches so a regression in either path trips the test.
        """
        state = _BaselineStreamState(model="gpt-4o-mini")
        create_mock = AsyncMock(return_value=_make_stream_mock())
        mock_client = MagicMock()
        mock_client.chat.completions.create = create_mock

        with (
            patch(
                "backend.copilot.baseline.service._get_main_client",
                return_value=mock_client,
            ),
            patch(
                "backend.copilot.baseline.service.config.use_openrouter",
                True,
            ),
            patch(
                "backend.copilot.baseline.service.config.api_key",
                "or-key",
            ),
            patch(
                "backend.copilot.baseline.service.config.base_url",
                "https://openrouter.ai/api/v1",
            ),
        ):
            await _baseline_llm_caller(
                messages=[{"role": "user", "content": "hi"}],
                tools=tools,
                state=state,
            )

        create_mock.assert_awaited_once()
        await_args = create_mock.await_args
        assert await_args is not None
        assert await_args.kwargs["extra_body"] == {"usage": {"include": True}}
        assert await_args.kwargs["stream_options"] == {"include_usage": True}


class TestMidLoopPendingFlushOrdering:
    """Regression test for the mid-loop pending drain ordering invariant.

    ``_baseline_conversation_updater`` records assistant+tool entries from
    each tool-call round into ``state.session_messages``; the finally block
    of ``stream_chat_completion_baseline`` batch-flushes them into
    ``session.messages`` at the end of the turn.

    The mid-loop pending drain appends pending user messages directly to
    ``session.messages``.  Without flushing ``state.session_messages`` first,
    the pending user message lands BEFORE the preceding round's assistant+
    tool entries in the final persisted ``session.messages`` — which
    produces a malformed tool-call/tool-result ordering on the next turn's
    replay.

    This test documents the invariant by replaying the production flush
    sequence against an in-memory state.
    """

    def test_flush_then_append_preserves_chronological_order(self):
        """Mid-loop drain must flush state.session_messages before appending
        the pending user message, so the final order matches the
        chronological execution order.
        """
        # Initial state: user turn already appended by maybe_append_user_message
        session_messages: list[ChatMessage] = [
            ChatMessage(role="user", content="original user turn"),
        ]
        state = _BaselineStreamState()

        # Round 1 completes: conversation_updater buffers assistant+tool
        # entries into state.session_messages (but does NOT write to
        # session.messages yet).
        builder = TranscriptBuilder()
        builder.append_user("original user turn")
        response = LLMLoopResponse(
            response_text="calling search",
            tool_calls=[LLMToolCall(id="tc_1", name="search", arguments="{}")],
            raw_response=None,
            prompt_tokens=0,
            completion_tokens=0,
        )
        tool_results = [
            ToolCallResult(
                tool_call_id="tc_1", tool_name="search", content="search output"
            ),
        ]
        openai_messages: list = []
        _baseline_conversation_updater(
            openai_messages,
            response,
            tool_results=tool_results,
            transcript_builder=builder,
            state=state,
            model="test-model",
        )
        # state.session_messages should now hold the round-1 assistant + tool
        assert len(state.session_messages) == 2
        assert state.session_messages[0].role == "assistant"
        assert state.session_messages[1].role == "tool"

        # --- Mid-loop pending drain (production code pattern) ---
        # Flush first, THEN append pending.  This is the ordering fix.
        for _buffered in state.session_messages:
            session_messages.append(_buffered)
        state.session_messages.clear()
        session_messages.append(
            ChatMessage(role="user", content="pending mid-loop message")
        )

        # Round 2 completes: new assistant+tool entries buffer again.
        response2 = LLMLoopResponse(
            response_text="another call",
            tool_calls=[LLMToolCall(id="tc_2", name="calc", arguments="{}")],
            raw_response=None,
            prompt_tokens=0,
            completion_tokens=0,
        )
        tool_results2 = [
            ToolCallResult(
                tool_call_id="tc_2", tool_name="calc", content="calc output"
            ),
        ]
        _baseline_conversation_updater(
            openai_messages,
            response2,
            tool_results=tool_results2,
            transcript_builder=builder,
            state=state,
            model="test-model",
        )

        # --- Finally-block flush (end of turn) ---
        for msg in state.session_messages:
            session_messages.append(msg)

        # Assert chronological order: original user, round-1 assistant,
        # round-1 tool, pending user, round-2 assistant, round-2 tool.
        assert [m.role for m in session_messages] == [
            "user",
            "assistant",
            "tool",
            "user",
            "assistant",
            "tool",
        ]
        assert session_messages[0].content == "original user turn"
        assert session_messages[3].content == "pending mid-loop message"
        # The assistant message carrying tool_call tc_1 must be immediately
        # followed by its tool result — no user message interposed.
        assert session_messages[1].role == "assistant"
        assert session_messages[1].tool_calls is not None
        assert session_messages[1].tool_calls[0]["id"] == "tc_1"
        assert session_messages[2].role == "tool"
        assert session_messages[2].tool_call_id == "tc_1"
        # Same invariant for the round after the pending user.
        assert session_messages[4].tool_calls is not None
        assert session_messages[4].tool_calls[0]["id"] == "tc_2"
        assert session_messages[5].tool_call_id == "tc_2"

    def test_flushed_assistant_text_len_prevents_duplicate_final_text(self):
        """After mid-loop drain clears state.session_messages, the finally
        block must not re-append assistant text from rounds already flushed.

        ``state.assistant_text`` accumulates ALL rounds' text, but
        ``state.session_messages`` only holds entries from rounds AFTER the
        last mid-loop flush.  Without ``_flushed_assistant_text_len``, the
        ``finally`` block's ``startswith(recorded)`` check fails because
        ``recorded`` only covers post-flush rounds, and the full
        ``assistant_text`` is appended — duplicating pre-flush rounds.
        """
        state = _BaselineStreamState()
        session_messages: list[ChatMessage] = [
            ChatMessage(role="user", content="user turn"),
        ]

        # Simulate round 1 text accumulation (as _bound_llm_caller does)
        state.assistant_text += "calling search"

        # Round 1 conversation_updater buffers structured entries
        builder = TranscriptBuilder()
        builder.append_user("user turn")
        response1 = LLMLoopResponse(
            response_text="calling search",
            tool_calls=[LLMToolCall(id="tc_1", name="search", arguments="{}")],
            raw_response=None,
            prompt_tokens=0,
            completion_tokens=0,
        )
        _baseline_conversation_updater(
            [],
            response1,
            tool_results=[
                ToolCallResult(
                    tool_call_id="tc_1", tool_name="search", content="result"
                )
            ],
            transcript_builder=builder,
            state=state,
            model="test-model",
        )

        # Mid-loop drain: flush + clear + record flushed text length
        for _buffered in state.session_messages:
            session_messages.append(_buffered)
        state.session_messages.clear()
        state._flushed_assistant_text_len = len(state.assistant_text)
        session_messages.append(ChatMessage(role="user", content="pending message"))

        # Simulate round 2 text accumulation
        state.assistant_text += "final answer"

        # Round 2: natural finish (no tool calls → no session_messages entry)

        # --- Finally block logic (production code) ---
        for msg in state.session_messages:
            session_messages.append(msg)

        final_text = state.assistant_text[state._flushed_assistant_text_len :]
        if state.session_messages:
            recorded = "".join(
                m.content or "" for m in state.session_messages if m.role == "assistant"
            )
            if final_text.startswith(recorded):
                final_text = final_text[len(recorded) :]
        if final_text.strip():
            session_messages.append(ChatMessage(role="assistant", content=final_text))

        # The final assistant message should only contain round-2 text,
        # not the round-1 text that was already flushed mid-loop.
        assistant_msgs = [m for m in session_messages if m.role == "assistant"]
        # Round-1 structured assistant (from mid-loop flush)
        assert assistant_msgs[0].content == "calling search"
        assert assistant_msgs[0].tool_calls is not None
        # Round-2 final text (from finally block)
        assert assistant_msgs[1].content == "final answer"
        assert assistant_msgs[1].tool_calls is None
        # Crucially: only 2 assistant messages, not 3 (no duplicate)
        assert len(assistant_msgs) == 2


class TestBuilderContextSplit:
    """Cross-helper composition: the guide must land in the system prompt via
    ``build_builder_system_prompt_suffix`` and NOT in the per-turn user prefix
    via ``build_builder_context_turn_prefix``.

    The baseline service composes these two blocks on each turn, so a drift
    here (guide leaking into both, or missing from both) would kill Claude's
    prompt-cache hit rate for builder sessions.
    """

    @pytest.mark.asyncio
    async def test_guide_lives_in_system_prompt_not_user_message(self):
        from backend.copilot.builder_context import (
            BUILDER_CONTEXT_TAG,
            BUILDER_SESSION_TAG,
            build_builder_context_turn_prefix,
            build_builder_system_prompt_suffix,
        )
        from backend.copilot.model import ChatSession

        session = MagicMock(spec=ChatSession)
        session.session_id = "s"
        session.metadata = MagicMock()
        session.metadata.builder_graph_id = "graph-1"

        agent_json = {
            "id": "graph-1",
            "name": "Demo",
            "version": 7,
            "nodes": [
                {
                    "id": "n1",
                    "block_id": "block-A",
                    "input_default": {"name": "Input"},
                    "metadata": {},
                }
            ],
            "links": [],
        }
        guide_body = "# UNIQUE_GUIDE_MARKER body"
        with (
            patch(
                "backend.copilot.builder_context.get_agent_as_json",
                new=AsyncMock(return_value=agent_json),
            ),
            patch(
                "backend.copilot.builder_context._load_guide",
                return_value=guide_body,
            ),
        ):
            suffix = await build_builder_system_prompt_suffix(session)
            prefix = await build_builder_context_turn_prefix(session, "user-1")

        # System prompt suffix carries <builder_session> and the guide.
        assert f"<{BUILDER_SESSION_TAG}>" in suffix
        assert guide_body in suffix
        # Dynamic bits must NOT be in the suffix — otherwise renames and
        # cross-graph sessions invalidate Claude's prompt cache.
        assert "graph-1" not in suffix
        assert "Demo" not in suffix

        # Per-turn prefix carries <builder_context> with the full live
        # snapshot (id, name, version, nodes) but NEVER the guide.
        assert f"<{BUILDER_CONTEXT_TAG}>" in prefix
        assert 'id="graph-1"' in prefix
        assert 'name="Demo"' in prefix
        assert 'version="7"' in prefix
        assert guide_body not in prefix
        assert "<building_guide>" not in prefix

        # Guide appears in the combined on-the-wire payload exactly ONCE.
        combined = suffix + "\n\n" + prefix
        assert combined.count(guide_body) == 1


class TestApplyPromptCacheMarkers:
    """Tests for _apply_prompt_cache_markers — Anthropic ephemeral
    cache_control markers on baseline OpenRouter requests."""

    def test_system_message_converted_to_content_blocks(self):
        messages = [
            {"role": "system", "content": "You are helpful."},
            {"role": "user", "content": "hello"},
        ]

        cached_messages = _mark_system_message_with_cache_control(messages)

        assert cached_messages[0]["role"] == "system"
        assert cached_messages[0]["content"] == [
            {
                "type": "text",
                "text": "You are helpful.",
                "cache_control": {"type": "ephemeral", "ttl": "1h"},
            }
        ]
        # User message must be untouched.
        assert cached_messages[1] == {"role": "user", "content": "hello"}

    def test_system_message_preserves_unknown_fields(self):
        # Future-proofing: a system message with extra keys (e.g. "name") must
        # keep them after the content-blocks conversion.
        messages = [
            {"role": "system", "content": "sys", "name": "developer"},
        ]

        cached_messages = _mark_system_message_with_cache_control(messages)

        assert cached_messages[0]["name"] == "developer"
        assert cached_messages[0]["role"] == "system"

    def test_last_tool_gets_cache_control(self):
        tools = [
            {"type": "function", "function": {"name": "a"}},
            {"type": "function", "function": {"name": "b"}},
        ]

        cached_tools = _mark_tools_with_cache_control(tools)

        assert "cache_control" not in cached_tools[0]
        assert cached_tools[-1]["cache_control"] == {
            "type": "ephemeral",
            "ttl": "1h",
        }
        # Last tool's other fields preserved.
        assert cached_tools[-1]["function"] == {"name": "b"}

    def test_does_not_mutate_input(self):
        messages = [{"role": "system", "content": "sys"}]
        tools = [{"type": "function", "function": {"name": "a"}}]

        _mark_system_message_with_cache_control(messages)
        _mark_tools_with_cache_control(tools)

        assert messages == [{"role": "system", "content": "sys"}]
        assert tools == [{"type": "function", "function": {"name": "a"}}]

    def test_no_system_message_safe(self):
        messages = [{"role": "user", "content": "hi"}]
        cached_messages = _mark_system_message_with_cache_control(messages)
        assert cached_messages == messages

    def test_empty_tools_safe(self):
        assert _mark_tools_with_cache_control([]) == []

    def test_non_string_system_content_left_untouched(self):
        # If the content is already a list of blocks (e.g. caller pre-marked),
        # the helper must not overwrite it.
        pre_marked = [
            {
                "type": "text",
                "text": "sys",
                "cache_control": {"type": "ephemeral", "ttl": "1h"},
            }
        ]
        messages = [{"role": "system", "content": pre_marked}]
        cached_messages = _mark_system_message_with_cache_control(messages)
        assert cached_messages[0]["content"] == pre_marked

    def test_is_anthropic_model_matches_claude_and_anthropic_prefix(self):
        assert _is_anthropic_model("anthropic/claude-sonnet-4-6")
        assert _is_anthropic_model("claude-3-5-sonnet-20241022")
        assert _is_anthropic_model("anthropic.claude-3-5-sonnet-20241022-v2:0")
        assert _is_anthropic_model("ANTHROPIC/Claude-Opus")  # case insensitive

    def test_is_anthropic_model_rejects_other_providers(self):
        assert not _is_anthropic_model("openai/gpt-4o")
        assert not _is_anthropic_model("openai/gpt-5")
        assert not _is_anthropic_model("google/gemini-2.5-pro")
        assert not _is_anthropic_model("xai/grok-4")
        assert not _is_anthropic_model("meta-llama/llama-3.3-70b-instruct")

    def test_is_anthropic_model_rejects_kimi_routes(self):
        """Regression guard: Kimi K2.6 is a reasoning route (reasoning
        extra_body is sent) but NOT an Anthropic route — Moonshot does
        its own auto prompt caching, so ``cache_control`` markers must
        NOT be applied. OpenRouter silently drops them today, but if
        they ever start failing fast we'd want the gate tight."""
        assert not _is_anthropic_model("moonshotai/kimi-k2.6")
        assert not _is_anthropic_model("moonshotai/kimi-k2-thinking")
        assert not _is_anthropic_model("kimi-k2-instruct")

    def test_cache_control_uses_configured_ttl(self, monkeypatch):
        """TTL comes from ChatConfig.baseline_prompt_cache_ttl — defaults
        to 1h so the static prefix (system + tools) stays warm across
        workspace users past the 5-min default window."""
        from backend.copilot.baseline import service as bsvc

        assert bsvc.config.baseline_prompt_cache_ttl == "1h"
        cc = bsvc._fresh_ephemeral_cache_control()
        assert cc == {"type": "ephemeral", "ttl": "1h"}
        monkeypatch.setattr(bsvc.config, "baseline_prompt_cache_ttl", "5m")
        assert bsvc._fresh_ephemeral_cache_control() == {
            "type": "ephemeral",
            "ttl": "5m",
        }

    def test_fresh_helpers_return_distinct_objects(self):
        """Regression guard: the `_fresh_*` helpers must return a NEW dict
        on every call.  A future refactor returning a module-level constant
        would silently reintroduce the shared-mutable-state bug flagged
        during earlier review cycles."""
        assert _fresh_ephemeral_cache_control() is not _fresh_ephemeral_cache_control()
        assert (
            _fresh_anthropic_caching_headers() is not _fresh_anthropic_caching_headers()
        )

    def test_extract_cache_creation_tokens_openrouter_typed_attr(self):
        """Newer ``openai-python`` declares ``cache_write_tokens`` as a
        typed attribute on ``PromptTokensDetails`` — it no longer lands in
        ``model_extra``.  Verified empirically against the production
        openai==1.113 installed in this venv: OpenRouter streaming
        response populates ``ptd.cache_write_tokens`` directly while
        ``ptd.model_extra`` is ``{}``.
        """
        from openai.types.completion_usage import PromptTokensDetails

        ptd = PromptTokensDetails.model_validate(
            {
                "audio_tokens": 0,
                "cached_tokens": 0,
                "cache_write_tokens": 4432,
                "video_tokens": 0,
            }
        )
        assert getattr(ptd, "cache_write_tokens", None) == 4432
        assert _extract_cache_creation_tokens(ptd) == 4432

    def test_extract_cache_creation_tokens_openrouter_model_extra(self):
        """Older SDKs that don't yet declare ``cache_write_tokens`` as a
        typed field leave it in ``model_extra`` — the helper must still
        find it there."""
        from openai.types.completion_usage import PromptTokensDetails

        ptd = PromptTokensDetails.model_validate({"cached_tokens": 0})
        # Force the value into model_extra (simulates the old SDK shape
        # where the field wasn't typed yet).
        if ptd.model_extra is None:
            # Pydantic v2 sometimes exposes __pydantic_extra__ as None when
            # extras are disabled; initialise to a dict to mutate safely.
            object.__setattr__(ptd, "__pydantic_extra__", {})
        assert ptd.model_extra is not None
        ptd.model_extra["cache_write_tokens"] = 7777
        assert _extract_cache_creation_tokens(ptd) == 7777

    def test_extract_cache_creation_tokens_anthropic_native_field(self):
        """Direct Anthropic API uses ``cache_creation_input_tokens`` —
        falls through as the final path when neither
        ``cache_write_tokens`` typed attr nor model_extra entry exists."""
        from openai.types.completion_usage import PromptTokensDetails

        ptd = PromptTokensDetails.model_validate({"cached_tokens": 0})
        if ptd.model_extra is None:
            object.__setattr__(ptd, "__pydantic_extra__", {})
        assert ptd.model_extra is not None
        ptd.model_extra["cache_creation_input_tokens"] = 2048
        assert _extract_cache_creation_tokens(ptd) == 2048

    def test_extract_cache_creation_tokens_absent(self):
        """Neither provider field present → 0 (non-Anthropic routes or
        cache-miss responses)."""
        from openai.types.completion_usage import PromptTokensDetails

        ptd = PromptTokensDetails.model_validate({"cached_tokens": 0})
        assert _extract_cache_creation_tokens(ptd) == 0

    def test_build_cached_system_message_applies_cache_control(self):
        """The single-message helper wraps the string content in a text block
        with an ephemeral cache_control marker."""
        out = _build_cached_system_message({"role": "system", "content": "hi"})
        assert out["role"] == "system"
        assert out["content"] == [
            {
                "type": "text",
                "text": "hi",
                "cache_control": {"type": "ephemeral", "ttl": "1h"},
            }
        ]

    def test_build_cached_system_message_preserves_extra_fields(self):
        """Unknown keys (e.g. ``name``) survive the transformation."""
        out = _build_cached_system_message(
            {"role": "system", "content": "sys", "name": "dev"}
        )
        assert out["name"] == "dev"
        assert out["role"] == "system"

    def test_build_cached_system_message_non_string_passthrough(self):
        """Pre-marked list content is returned as-is (shallow-copied)."""
        pre_marked = [
            {
                "type": "text",
                "text": "sys",
                "cache_control": {"type": "ephemeral", "ttl": "1h"},
            }
        ]
        out = _build_cached_system_message({"role": "system", "content": pre_marked})
        assert out["content"] is pre_marked

    @pytest.mark.asyncio
    async def test_baseline_llm_caller_memoises_cached_system_message(self):
        """The cached system dict is built once and reused across rounds.

        Guards against the perf regression where the entire (growing)
        ``messages`` list was copied on every tool-call iteration just to
        mark the static system prompt.
        """
        state = _BaselineStreamState(model="anthropic/claude-sonnet-4")
        chunk = _make_usage_chunk(prompt_tokens=10, completion_tokens=5)

        mock_client = MagicMock()
        mock_client.chat.completions.create = AsyncMock(
            side_effect=[_make_stream_mock(chunk), _make_stream_mock(chunk)]
        )

        messages: list[dict] = [
            {"role": "system", "content": "You are helpful."},
            {"role": "user", "content": "hi"},
        ]
        with patch(
            "backend.copilot.baseline.service._get_main_client",
            return_value=mock_client,
        ):
            await _baseline_llm_caller(messages=messages, tools=[], state=state)
            first_cached = state.cached_system_message
            assert first_cached is not None
            # Simulate the tool-call loop growing ``messages`` between rounds.
            messages.append({"role": "assistant", "content": "ok"})
            messages.append({"role": "user", "content": "follow up"})
            await _baseline_llm_caller(messages=messages, tools=[], state=state)

        # Same dict instance reused — not rebuilt per round.
        assert state.cached_system_message is first_cached

        # Second call's first message is the memoised system dict (not a new copy).
        second_call_messages = mock_client.chat.completions.create.call_args_list[1][1][
            "messages"
        ]
        assert second_call_messages[0] is first_cached
        # And the tail messages were spliced in, not re-copied.
        assert second_call_messages[1] is messages[1]
        assert second_call_messages[-1] is messages[-1]

    @pytest.mark.asyncio
    async def test_baseline_llm_caller_skips_memoisation_for_non_anthropic(self):
        """Non-Anthropic routes pass messages through unmodified — no cache
        dict is built, no list splicing happens."""
        state = _BaselineStreamState(model="openai/gpt-4o")
        chunk = _make_usage_chunk(prompt_tokens=10, completion_tokens=5)

        mock_client = MagicMock()
        mock_client.chat.completions.create = AsyncMock(
            return_value=_make_stream_mock(chunk)
        )

        messages: list[dict] = [
            {"role": "system", "content": "sys"},
            {"role": "user", "content": "hi"},
        ]
        with patch(
            "backend.copilot.baseline.service._get_main_client",
            return_value=mock_client,
        ):
            await _baseline_llm_caller(messages=messages, tools=[], state=state)

        assert state.cached_system_message is None
        # The exact same list object reaches the provider (no copy needed).
        call_messages = mock_client.chat.completions.create.call_args[1]["messages"]
        assert call_messages is messages


def _make_delta_chunk(
    *,
    content: str | None = None,
    reasoning: str | None = None,
    reasoning_details: list | None = None,
    reasoning_content: str | None = None,
    tool_calls: list | None = None,
):
    """Build a streaming chunk with a configurable ``delta`` payload.

    The ``delta`` is a real ``ChoiceDelta`` pydantic instance so OpenRouter
    extension fields land on ``delta.model_extra`` — which is how
    :class:`OpenRouterDeltaExtension` reads them in production.  Using a
    raw ``MagicMock`` here would leave ``model_extra`` unset and silently
    skip the reasoning parser.  ``tool_calls`` (when provided) must be
    ``MagicMock`` entries compatible with the service's streaming loop;
    they're set on the delta via ``object.__setattr__`` because pydantic
    would otherwise reject the non-schema types.
    """
    from openai.types.chat.chat_completion_chunk import ChoiceDelta

    payload: dict = {"role": "assistant"}
    if content is not None:
        payload["content"] = content
    if reasoning is not None:
        payload["reasoning"] = reasoning
    if reasoning_content is not None:
        payload["reasoning_content"] = reasoning_content
    if reasoning_details is not None:
        payload["reasoning_details"] = reasoning_details
    delta = ChoiceDelta.model_validate(payload)
    # ChoiceDelta's tool_calls schema expects OpenAI-typed entries; bypass
    # validation so tests can use MagicMocks that mimic the streaming shape.
    if tool_calls is not None:
        object.__setattr__(delta, "tool_calls", tool_calls)

    chunk = MagicMock()
    chunk.usage = None
    choice = MagicMock()
    choice.delta = delta
    chunk.choices = [choice]
    return chunk


def _make_tool_call_delta(*, index: int, call_id: str, name: str, arguments: str):
    """Build a ``delta.tool_calls[i]`` entry for streaming tool-use."""
    tc = MagicMock()
    tc.index = index
    tc.id = call_id
    function = MagicMock()
    function.name = name
    function.arguments = arguments
    tc.function = function
    return tc


class TestBaselineReasoningStreaming:
    """End-to-end reasoning event emission through ``_baseline_llm_caller``."""

    @pytest.mark.asyncio
    async def test_reasoning_then_text_emits_paired_events(self):
        state = _BaselineStreamState(model="anthropic/claude-sonnet-4-6")

        chunks = [
            _make_delta_chunk(reasoning="thinking..."),
            _make_delta_chunk(reasoning=" more"),
            _make_delta_chunk(content="final answer"),
        ]

        mock_client = MagicMock()
        mock_client.chat.completions.create = AsyncMock(
            return_value=_make_stream_mock(*chunks)
        )

        with patch(
            "backend.copilot.baseline.service._get_main_client",
            return_value=mock_client,
        ):
            await _baseline_llm_caller(
                messages=[{"role": "user", "content": "hi"}],
                tools=[],
                state=state,
            )

        types = [type(e).__name__ for e in state.emitted_events]
        assert "StreamReasoningStart" in types
        assert "StreamReasoningDelta" in types
        assert "StreamReasoningEnd" in types

        # Reasoning must close before text opens — AI SDK v5 rejects
        # interleaved reasoning / text parts.
        reason_end = types.index("StreamReasoningEnd")
        text_start = types.index("StreamTextStart")
        assert reason_end < text_start

        # All reasoning deltas share a single block id; the text block uses
        # a fresh id after the reasoning-end rotation.
        reasoning_ids = {
            e.id
            for e in state.emitted_events
            if isinstance(
                e, (StreamReasoningStart, StreamReasoningDelta, StreamReasoningEnd)
            )
        }
        text_ids = {
            e.id
            for e in state.emitted_events
            if isinstance(e, (StreamTextStart, StreamTextDelta, StreamTextEnd))
        }
        assert len(reasoning_ids) == 1
        assert len(text_ids) == 1
        assert reasoning_ids.isdisjoint(text_ids)

        combined = "".join(
            e.delta for e in state.emitted_events if isinstance(e, StreamReasoningDelta)
        )
        assert combined == "thinking... more"

    @pytest.mark.asyncio
    async def test_reasoning_then_tool_call_closes_reasoning_first(self):
        """A tool_call arriving mid-reasoning must close the reasoning block
        before the tool-use is flushed — AI SDK v5 treats reasoning and
        tool-use as distinct UI parts and rejects interleaving."""
        state = _BaselineStreamState(model="anthropic/claude-sonnet-4-6")

        chunks = [
            _make_delta_chunk(reasoning="deliberating..."),
            _make_delta_chunk(
                tool_calls=[
                    _make_tool_call_delta(
                        index=0,
                        call_id="call_1",
                        name="search",
                        arguments='{"q":"x"}',
                    )
                ],
            ),
        ]

        mock_client = MagicMock()
        mock_client.chat.completions.create = AsyncMock(
            return_value=_make_stream_mock(*chunks)
        )

        with patch(
            "backend.copilot.baseline.service._get_main_client",
            return_value=mock_client,
        ):
            response = await _baseline_llm_caller(
                messages=[{"role": "user", "content": "hi"}],
                tools=[],
                state=state,
            )

        # A reasoning-end must have been emitted — this is the tool_calls
        # branch's responsibility, not the stream-end cleanup.
        types = [type(e).__name__ for e in state.emitted_events]
        assert "StreamReasoningStart" in types
        assert "StreamReasoningEnd" in types

        # The tool_call was collected — confirms the tool-use path executed
        # after reasoning closed (rather than silently dropping the tool).
        assert len(response.tool_calls) == 1
        assert response.tool_calls[0].name == "search"

        # No text events — this stream had no content deltas.
        assert "StreamTextStart" not in types

    @pytest.mark.asyncio
    async def test_reasoning_closed_on_mid_stream_exception(self):
        """Regression guard: an exception during the streaming loop must
        still emit ``StreamReasoningEnd`` (and ``StreamTextEnd`` when a
        text block is open) before ``StreamFinishStep`` — the frontend
        collapse relies on matched start/end pairs, and the outer handler
        no longer patches these after-the-fact."""
        state = _BaselineStreamState(model="anthropic/claude-sonnet-4-6")

        async def failing_stream():
            yield _make_delta_chunk(reasoning="thinking...")
            raise RuntimeError("boom")

        stream = MagicMock()
        stream.close = AsyncMock()
        stream.__aiter__ = lambda self: failing_stream()

        mock_client = MagicMock()
        mock_client.chat.completions.create = AsyncMock(return_value=stream)

        with patch(
            "backend.copilot.baseline.service._get_main_client",
            return_value=mock_client,
        ):
            with pytest.raises(RuntimeError):
                await _baseline_llm_caller(
                    messages=[{"role": "user", "content": "hi"}],
                    tools=[],
                    state=state,
                )

        types = [type(e).__name__ for e in state.emitted_events]
        # The reasoning block was opened, the exception fired, and the
        # finally block must have closed it before emitting the finish
        # step.
        assert "StreamReasoningStart" in types
        assert "StreamReasoningEnd" in types
        assert "StreamFinishStep" in types
        assert types.index("StreamReasoningEnd") < types.index("StreamFinishStep")
        # Emitter is reset so a retried round starts with fresh ids.
        assert state.reasoning_emitter.is_open is False

    @pytest.mark.asyncio
    async def test_reasoning_param_sent_on_anthropic_routes(self):
        """Anthropic route via OpenRouter gets ``reasoning.max_tokens``."""
        state = _BaselineStreamState(model="anthropic/claude-sonnet-4-6")

        mock_client = MagicMock()
        mock_client.chat.completions.create = AsyncMock(
            return_value=_make_stream_mock()
        )

        with (
            patch(
                "backend.copilot.baseline.service._get_main_client",
                return_value=mock_client,
            ),
            patch(
                "backend.copilot.baseline.service.config.use_openrouter",
                True,
            ),
            patch(
                "backend.copilot.baseline.service.config.api_key",
                "or-key",
            ),
            patch(
                "backend.copilot.baseline.service.config.base_url",
                "https://openrouter.ai/api/v1",
            ),
        ):
            await _baseline_llm_caller(
                messages=[{"role": "user", "content": "hi"}],
                tools=[],
                state=state,
            )

        extra_body = mock_client.chat.completions.create.call_args[1]["extra_body"]
        assert "reasoning" in extra_body
        assert extra_body["reasoning"]["max_tokens"] > 0

    @pytest.mark.asyncio
    async def test_thinking_param_sent_on_direct_anthropic_route(self):
        """Direct-Anthropic mode (OR off) swaps OR's ``reasoning`` for the
        Anthropic-native ``thinking`` parameter so extended-thinking
        survives the OR→Anthropic transport flip."""
        state = _BaselineStreamState(model="anthropic/claude-sonnet-4-6")

        mock_client = MagicMock()
        mock_client.chat.completions.create = AsyncMock(
            return_value=_make_stream_mock()
        )

        with (
            patch(
                "backend.copilot.baseline.service._get_main_client",
                return_value=mock_client,
            ),
            patch(
                "backend.copilot.baseline.service.config.use_openrouter",
                False,
            ),
        ):
            await _baseline_llm_caller(
                messages=[{"role": "user", "content": "hi"}],
                tools=[],
                state=state,
            )

        call_kwargs = mock_client.chat.completions.create.call_args[1]
        extra_body = call_kwargs["extra_body"]
        # Native Anthropic shape, not the OR ``reasoning`` wrapper.
        assert "reasoning" not in extra_body
        assert extra_body["thinking"]["type"] == "enabled"
        budget = extra_body["thinking"]["budget_tokens"]
        assert budget > 0
        # Anthropic's OpenAI-compat layer requires ``max_tokens > budget_tokens``
        # whenever ``thinking`` is enabled — without this the request 400s.
        assert call_kwargs["max_tokens"] > budget

    @pytest.mark.asyncio
    async def test_thinking_budget_clamped_to_model_max_in_direct_mode(self):
        """When the operator-configured thinking budget exceeds the model's
        ``max_output_tokens`` ceiling, both the ``thinking.budget_tokens``
        parameter and ``max_tokens`` must be clamped so the
        ``max_tokens > budget_tokens`` and ``max_tokens <= model_max``
        contracts both hold — otherwise Anthropic 400s the request."""
        state = _BaselineStreamState(model="anthropic/claude-opus-4-1")

        mock_client = MagicMock()
        mock_client.chat.completions.create = AsyncMock(
            return_value=_make_stream_mock()
        )

        with (
            patch(
                "backend.copilot.baseline.service._get_main_client",
                return_value=mock_client,
            ),
            patch(
                "backend.copilot.baseline.service.config.use_openrouter",
                False,
            ),
            patch(
                "backend.copilot.baseline.service.config.claude_agent_max_thinking_tokens",
                128_000,  # well above any current Claude max_output_tokens
            ),
            patch(
                "backend.copilot.baseline.service.get_max_output_tokens",
                return_value=32_000,  # opus-4-x published ceiling
            ),
        ):
            await _baseline_llm_caller(
                messages=[{"role": "user", "content": "hi"}],
                tools=[],
                state=state,
            )

        call_kwargs = mock_client.chat.completions.create.call_args[1]
        extra_body = call_kwargs["extra_body"]
        budget = extra_body["thinking"]["budget_tokens"]
        max_tokens = call_kwargs["max_tokens"]
        # Both stay at/under the model ceiling — overflow would 400 Anthropic.
        assert max_tokens <= 32_000
        assert budget < max_tokens
        # Budget is clamped to model_max - 1 so max_tokens=model_max satisfies
        # the strict ``max_tokens > budget`` requirement.
        assert budget == 31_999
        assert max_tokens == 32_000

    @pytest.mark.asyncio
    async def test_max_tokens_absent_on_openrouter_thinking_route(self):
        """OR proxy injects its own default ``max_tokens`` so we leave
        ``create_kwargs`` clean — only direct-Anthropic mode needs the
        explicit ``max_tokens > budget_tokens`` guard."""
        state = _BaselineStreamState(model="anthropic/claude-sonnet-4-6")

        mock_client = MagicMock()
        mock_client.chat.completions.create = AsyncMock(
            return_value=_make_stream_mock()
        )

        with (
            patch(
                "backend.copilot.baseline.service._get_main_client",
                return_value=mock_client,
            ),
            patch(
                "backend.copilot.baseline.service.config.use_openrouter",
                True,
            ),
            patch(
                "backend.copilot.baseline.service.config.api_key",
                "or-key",
            ),
            patch(
                "backend.copilot.baseline.service.config.base_url",
                "https://openrouter.ai/api/v1",
            ),
        ):
            await _baseline_llm_caller(
                messages=[{"role": "user", "content": "hi"}],
                tools=[],
                state=state,
            )

        call_kwargs = mock_client.chat.completions.create.call_args[1]
        assert "max_tokens" not in call_kwargs

    @pytest.mark.asyncio
    async def test_reasoning_param_absent_on_non_anthropic_routes(self):
        """Non-reasoning routes (e.g. OpenAI) must not receive ``reasoning``."""
        state = _BaselineStreamState(model="openai/gpt-4o")

        mock_client = MagicMock()
        mock_client.chat.completions.create = AsyncMock(
            return_value=_make_stream_mock()
        )

        with patch(
            "backend.copilot.baseline.service._get_main_client",
            return_value=mock_client,
        ):
            await _baseline_llm_caller(
                messages=[{"role": "user", "content": "hi"}],
                tools=[],
                state=state,
            )

        extra_body = mock_client.chat.completions.create.call_args[1]["extra_body"]
        assert "reasoning" not in extra_body

    @pytest.mark.asyncio
    async def test_kimi_route_sends_reasoning_and_cache_control(self):
        """Kimi K2.6 (Moonshot via OpenRouter's Anthropic-compat endpoint)
        accepts ``cache_control: {type: ephemeral}`` on the system block
        and the last tool — the endpoint honours the marker and lifts
        cache hit rate on continuation turns from near-zero (Moonshot's
        auto-caching drifts) to the Anthropic ~60-95% ballpark.  The
        ``anthropic-beta`` header stays off because Moonshot doesn't need
        it; OpenRouter would strip the unknown header anyway."""
        state = _BaselineStreamState(model="moonshotai/kimi-k2.6")

        mock_client = MagicMock()
        mock_client.chat.completions.create = AsyncMock(
            return_value=_make_stream_mock()
        )

        with (
            patch(
                "backend.copilot.baseline.service._get_main_client",
                return_value=mock_client,
            ),
            patch(
                "backend.copilot.baseline.service.config.use_openrouter",
                True,
            ),
            patch(
                "backend.copilot.baseline.service.config.api_key",
                "or-key",
            ),
            patch(
                "backend.copilot.baseline.service.config.base_url",
                "https://openrouter.ai/api/v1",
            ),
        ):
            await _baseline_llm_caller(
                messages=[
                    {"role": "system", "content": "you are a helpful assistant"},
                    {"role": "user", "content": "hi"},
                ],
                tools=[
                    {
                        "type": "function",
                        "function": {"name": "echo", "parameters": {}},
                    }
                ],
                state=state,
            )

        call_kwargs = mock_client.chat.completions.create.call_args[1]
        extra_body = call_kwargs["extra_body"]
        # Reasoning param on — the whole point of picking Kimi is the
        # cheap-but-still-reasoning-capable path.
        assert "reasoning" in extra_body
        assert extra_body["reasoning"]["max_tokens"] > 0
        # No ``anthropic-beta`` header — that beta is specifically for
        # native Anthropic endpoints; Moonshot's shim accepts
        # ``cache_control`` without it, and sending it would be wasted
        # bytes (OR strips it before forwarding to Moonshot).
        assert "extra_headers" not in call_kwargs or not call_kwargs.get(
            "extra_headers"
        )
        # System block MUST carry ``cache_control`` so Moonshot's cache
        # breakpoint is honoured.  The cached system-message builder
        # emits list-shape content with the marker on the first (and
        # only) block — assert on that shape.
        sys_msg = call_kwargs["messages"][0]
        sys_content = sys_msg.get("content")
        assert isinstance(
            sys_content, list
        ), "Cached system message should be a list-shape content block"
        assert any(
            "cache_control" in block for block in sys_content if isinstance(block, dict)
        ), "Kimi system message should now carry cache_control markers"
        # Tool-level cache marking is applied by ``stream_chat_completion_baseline``
        # (see ``_mark_tools_with_cache_control``) before tools reach
        # ``_baseline_llm_caller``, so this unit test doesn't exercise
        # that path — covered by the outer integration test.

    @pytest.mark.asyncio
    async def test_reasoning_only_stream_still_closes_block(self):
        """Regression: a stream with only reasoning (no text, no tool_call)
        must still emit a matching ``reasoning-end`` at stream close so the
        frontend Reasoning collapse finalises.  Exercised here against
        ``_baseline_llm_caller`` to cover the emitter's integration with
        the finally-block, not just the unit emitter in reasoning_test.py.
        """
        state = _BaselineStreamState(model="anthropic/claude-sonnet-4-6")

        mock_client = MagicMock()
        mock_client.chat.completions.create = AsyncMock(
            return_value=_make_stream_mock(
                _make_delta_chunk(reasoning="just thinking"),
            )
        )

        with patch(
            "backend.copilot.baseline.service._get_main_client",
            return_value=mock_client,
        ):
            await _baseline_llm_caller(
                messages=[{"role": "user", "content": "hi"}],
                tools=[],
                state=state,
            )

        types = [type(e).__name__ for e in state.emitted_events]
        assert "StreamReasoningStart" in types
        assert "StreamReasoningEnd" in types
        # No text was produced — no text events should be emitted.
        assert "StreamTextStart" not in types
        assert "StreamTextDelta" not in types

    @pytest.mark.asyncio
    async def test_reasoning_param_suppressed_when_thinking_tokens_zero(self):
        """Operator kill switch: setting ``claude_agent_max_thinking_tokens``
        to 0 removes both the OR ``reasoning`` and the Anthropic-native
        ``thinking`` fragments from ``extra_body`` regardless of transport.
        Restores the zero-disables behaviour the old
        ``baseline_reasoning_max_tokens`` config used to provide."""
        state = _BaselineStreamState(model="anthropic/claude-sonnet-4-6")

        mock_client = MagicMock()
        mock_client.chat.completions.create = AsyncMock(
            return_value=_make_stream_mock()
        )

        with (
            patch(
                "backend.copilot.baseline.service._get_main_client",
                return_value=mock_client,
            ),
            patch(
                "backend.copilot.baseline.service.config.claude_agent_max_thinking_tokens",
                0,
            ),
        ):
            await _baseline_llm_caller(
                messages=[{"role": "user", "content": "hi"}],
                tools=[],
                state=state,
            )

        extra_body = mock_client.chat.completions.create.call_args[1]["extra_body"]
        assert "reasoning" not in extra_body
        assert "thinking" not in extra_body

    @pytest.mark.asyncio
    async def test_reasoning_persists_to_state_session_messages(self):
        """Integration guard: ``_BaselineStreamState.__post_init__`` wires
        the emitter to ``state.session_messages``, so reasoning deltas
        flowing through ``_baseline_llm_caller`` must produce a
        ``role="reasoning"`` row on the state's session list.  Catches
        regressions where the wiring silently breaks (e.g. a refactor
        passes the wrong list reference)."""
        state = _BaselineStreamState(model="anthropic/claude-sonnet-4-6")

        mock_client = MagicMock()
        mock_client.chat.completions.create = AsyncMock(
            return_value=_make_stream_mock(
                _make_delta_chunk(reasoning="first "),
                _make_delta_chunk(reasoning="thought"),
                _make_delta_chunk(content="answer"),
            )
        )

        with patch(
            "backend.copilot.baseline.service._get_main_client",
            return_value=mock_client,
        ):
            await _baseline_llm_caller(
                messages=[{"role": "user", "content": "hi"}],
                tools=[],
                state=state,
            )

        reasoning_rows = [m for m in state.session_messages if m.role == "reasoning"]
        assert len(reasoning_rows) == 1
        assert reasoning_rows[0].content == "first thought"


class TestSupportsPromptCacheMarkers:
    """``_supports_prompt_cache_markers`` is the widened gate for
    emitting ``cache_control`` markers on message content.  It's a
    superset of ``_is_anthropic_model`` that ALSO admits Moonshot
    (whose Anthropic-compat endpoint honours the marker) while keeping
    the False answer for OpenAI / Grok / Gemini (which 400 on the
    unknown field)."""

    @pytest.mark.parametrize(
        "model",
        [
            "anthropic/claude-sonnet-4-6",
            "claude-3-5-sonnet-20241022",
            "anthropic.claude-3-5-sonnet",
            "ANTHROPIC/Claude-Opus",
        ],
    )
    def test_anthropic_routes_are_supported(self, model):
        assert _supports_prompt_cache_markers(model) is True

    @pytest.mark.parametrize(
        "model",
        [
            "moonshotai/kimi-k2.6",
            "moonshotai/kimi-k2-thinking",
            "moonshotai/kimi-k2.5",
            "moonshotai/kimi-k3.0",  # future SKU
        ],
    )
    def test_moonshot_routes_are_supported(self, model):
        """The whole reason this predicate exists — Moonshot must be
        True even though ``_is_anthropic_model`` is False for it."""
        assert _supports_prompt_cache_markers(model) is True
        # Verify this is strictly wider than the anthropic-only check.
        assert _is_anthropic_model(model) is False

    @pytest.mark.parametrize(
        "model",
        [
            "openai/gpt-4o",
            "google/gemini-2.5-pro",
            "xai/grok-4",
            "meta-llama/llama-3.3-70b-instruct",
            "deepseek/deepseek-v3",
        ],
    )
    def test_other_providers_still_rejected(self, model):
        """Regression guard: OpenAI/Grok/Gemini still 400 on
        ``cache_control``, so the widened gate must keep them out."""
        assert _supports_prompt_cache_markers(model) is False


class TestBudgetExhaustedNoticeText:
    """Tests for the fallback-notice decision used when the tool-round
    budget is exhausted without a natural finish."""

    def test_empty_text_returns_fallback(self):
        assert _budget_exhausted_notice_text("") == _BUDGET_EXHAUSTED_FALLBACK_TEXT

    def test_whitespace_only_returns_fallback(self):
        """A string of only whitespace is still "no visible response"."""
        assert (
            _budget_exhausted_notice_text("   \n\t  ")
            == _BUDGET_EXHAUSTED_FALLBACK_TEXT
        )

    def test_non_empty_text_returns_none(self):
        """When the model already summarised, stay quiet — no extra notice."""
        assert _budget_exhausted_notice_text("Here is what I did...") is None

    def test_fallback_text_is_user_facing(self):
        """Guard against accidentally shipping an empty / internal string."""
        assert _BUDGET_EXHAUSTED_FALLBACK_TEXT.strip()
        assert "tool-call budget" in _BUDGET_EXHAUSTED_FALLBACK_TEXT
        assert "follow-up" in _BUDGET_EXHAUSTED_FALLBACK_TEXT


class TestBuildBudgetExhaustedFallbackEvents:
    """Tests for the helper that produces the stream events + text mutation
    for a budget-exhausted turn with no terminal-round text."""

    def test_empty_terminal_text_emits_three_events(self):
        events, to_append = _build_budget_exhausted_fallback_events("")
        assert to_append == _BUDGET_EXHAUSTED_FALLBACK_TEXT
        assert len(events) == 3
        assert isinstance(events[0], StreamTextStart)
        assert isinstance(events[1], StreamTextDelta)
        assert isinstance(events[2], StreamTextEnd)
        # All three events share the same block id so the frontend groups
        # them into a single text bubble.
        assert events[0].id == events[1].id == events[2].id
        # The delta carries the user-facing notice verbatim.
        assert events[1].delta == _BUDGET_EXHAUSTED_FALLBACK_TEXT

    def test_non_empty_terminal_text_returns_empty(self):
        """Model already produced visible final text → no fallback."""
        events, to_append = _build_budget_exhausted_fallback_events(
            "Here's what I did so far..."
        )
        assert events == []
        assert to_append == ""

    def test_whitespace_only_still_emits_fallback(self):
        events, to_append = _build_budget_exhausted_fallback_events("   \n\t  ")
        assert len(events) == 3
        assert to_append == _BUDGET_EXHAUSTED_FALLBACK_TEXT

    def test_each_call_uses_fresh_block_id(self):
        """Block IDs are UUIDs — two invocations must not collide."""
        events_a, _ = _build_budget_exhausted_fallback_events("")
        events_b, _ = _build_budget_exhausted_fallback_events("")
        assert events_a[0].id != events_b[0].id


class TestNaturalFinishEmptyNoticeText:
    """Fallback decision when the model finished naturally (under budget)
    but the terminal round produced no visible text — the baseline
    equivalent of the SDK's ``empty_completion`` StreamError. Without
    this fallback the FE shows nothing after a long thinking-only turn
    on baseline-routed sessions (e.g. some Kimi K2.6 cohorts)."""

    def test_empty_text_returns_fallback(self):
        assert (
            _natural_finish_empty_notice_text("") == _NATURAL_FINISH_EMPTY_FALLBACK_TEXT
        )

    def test_whitespace_only_returns_fallback(self):
        assert (
            _natural_finish_empty_notice_text("   \n\t  ")
            == _NATURAL_FINISH_EMPTY_FALLBACK_TEXT
        )

    def test_non_empty_text_returns_none(self):
        assert _natural_finish_empty_notice_text("Here is the answer.") is None

    def test_fallback_text_is_user_facing(self):
        """Guard against shipping an empty / internal string."""
        assert _NATURAL_FINISH_EMPTY_FALLBACK_TEXT.strip()
        # Distinct from the budget message — they describe different causes
        # so the user gets accurate signal about what happened.
        assert _NATURAL_FINISH_EMPTY_FALLBACK_TEXT != _BUDGET_EXHAUSTED_FALLBACK_TEXT


class TestBuildNaturalFinishEmptyFallbackEvents:
    def test_empty_terminal_text_emits_three_events(self):
        events, to_append = _build_natural_finish_empty_fallback_events("")
        assert to_append == _NATURAL_FINISH_EMPTY_FALLBACK_TEXT
        assert len(events) == 3
        assert isinstance(events[0], StreamTextStart)
        assert isinstance(events[1], StreamTextDelta)
        assert isinstance(events[2], StreamTextEnd)
        assert events[0].id == events[1].id == events[2].id
        assert events[1].delta == _NATURAL_FINISH_EMPTY_FALLBACK_TEXT

    def test_non_empty_terminal_text_returns_empty(self):
        events, to_append = _build_natural_finish_empty_fallback_events(
            "Here is the answer."
        )
        assert events == []
        assert to_append == ""


class TestStreamOptionsGating:
    """stream_options must be present for OR and absent for direct Anthropic."""

    @pytest.mark.asyncio
    async def test_stream_options_absent_in_direct_mode(self):
        state = _BaselineStreamState(model="claude-sonnet-4-6")
        create_mock = AsyncMock(return_value=_make_stream_mock())
        mock_client = MagicMock()
        mock_client.chat.completions.create = create_mock

        with (
            patch(
                "backend.copilot.baseline.service._get_main_client",
                return_value=mock_client,
            ),
            patch("backend.copilot.baseline.service.config.use_openrouter", False),
            patch("backend.copilot.baseline.service.config.api_key", "ant-key"),
            patch("backend.copilot.baseline.service.config.base_url", None),
        ):
            await _baseline_llm_caller(
                messages=[{"role": "user", "content": "hi"}],
                tools=[],
                state=state,
            )

        create_mock.assert_awaited_once()
        assert create_mock.await_args is not None
        assert "stream_options" not in create_mock.await_args.kwargs

    @pytest.mark.asyncio
    async def test_stream_options_present_in_openrouter_mode(self):
        state = _BaselineStreamState(model="anthropic/claude-sonnet-4-6")
        create_mock = AsyncMock(return_value=_make_stream_mock())
        mock_client = MagicMock()
        mock_client.chat.completions.create = create_mock

        with (
            patch(
                "backend.copilot.baseline.service._get_main_client",
                return_value=mock_client,
            ),
            patch("backend.copilot.baseline.service.config.use_openrouter", True),
            patch("backend.copilot.baseline.service.config.api_key", "or-key"),
            patch(
                "backend.copilot.baseline.service.config.base_url",
                "https://openrouter.ai/api/v1",
            ),
        ):
            await _baseline_llm_caller(
                messages=[{"role": "user", "content": "hi"}],
                tools=[],
                state=state,
            )

        create_mock.assert_awaited_once()
        assert create_mock.await_args is not None
        assert create_mock.await_args.kwargs["stream_options"] == {
            "include_usage": True
        }


class TestDirectModeProviderLabel:
    """persist_and_record_usage must receive provider='anthropic' in direct mode
    and provider='open_router' in OR mode."""

    @pytest.mark.asyncio
    async def test_provider_label_is_anthropic_in_direct_mode(self):
        state = _BaselineStreamState(model="claude-sonnet-4-6")
        mock_client = MagicMock()
        mock_client.chat.completions.create = AsyncMock(
            return_value=_make_stream_mock(
                _make_usage_chunk(prompt_tokens=100, completion_tokens=50)
            )
        )

        with (
            patch(
                "backend.copilot.baseline.service._get_main_client",
                return_value=mock_client,
            ),
            patch("backend.copilot.baseline.service.config.use_openrouter", False),
            patch("backend.copilot.baseline.service.config.api_key", "ant-key"),
            patch("backend.copilot.baseline.service.config.base_url", None),
            patch(
                "backend.copilot.baseline.service.persist_and_record_usage"
            ) as mock_persist,
        ):
            await _baseline_llm_caller(
                messages=[{"role": "user", "content": "hi"}],
                tools=[],
                state=state,
            )

        # persist_and_record_usage is called from stream_chat_completion_baseline,
        # not _baseline_llm_caller, so the provider assertion belongs at the
        # higher level — but we can check state was updated with cost by the
        # lower-level call and that the config gate resolves correctly.
        # The direct integration is covered by TestBaselineCostExtraction;
        # here we just verify the gate expression evaluates correctly.
        assert not mock_persist.called  # called by outer stream fn, not llm_caller
        # The cost was computed from the rate card (direct mode, no OR extension)
        assert state.cost_usd is not None
        assert state.cost_usd > 0

    @pytest.mark.asyncio
    async def test_openrouter_active_false_when_no_base_url(self):
        """Config.openrouter_active is False when base_url is None (direct mode),
        so the provider='anthropic' branch is taken."""
        from backend.copilot.config import ChatConfig

        cfg = ChatConfig(
            use_openrouter=False,
            api_key="ant-key",
            base_url=None,
            aux_api_key=None,
            aux_base_url=None,
            title_model="anthropic/claude-haiku-4-5",
        )
        assert not cfg.openrouter_active


class TestDirectModeCostRecoveryOnMissingUsageChunk:
    """When the stream ends without a usage chunk, direct mode must still
    record a non-zero cost from the tiktoken fallback to prevent rate-limit
    bypass (token_tracking.py skips charging when cost_microdollars == 0)."""

    @pytest.mark.asyncio
    async def test_cost_computed_from_tiktoken_when_no_usage_chunk(self):
        state = _BaselineStreamState(model="claude-sonnet-4-6")
        # Stream produces a text delta but no usage chunk.
        no_usage_stream = _make_stream_mock()  # empty = no chunks at all

        mock_client = MagicMock()
        mock_client.chat.completions.create = AsyncMock(return_value=no_usage_stream)

        with (
            patch(
                "backend.copilot.baseline.service._get_main_client",
                return_value=mock_client,
            ),
            patch("backend.copilot.baseline.service.config.use_openrouter", False),
            patch("backend.copilot.baseline.service.config.api_key", "ant-key"),
            patch("backend.copilot.baseline.service.config.base_url", None),
        ):
            await _baseline_llm_caller(
                messages=[{"role": "user", "content": "hello world"}],
                tools=[],
                state=state,
            )

        # With no usage chunk, _baseline_llm_caller leaves cost_usd = None.
        # The final-fallback at stream_chat_completion_baseline level picks it
        # up from tiktoken. At _baseline_llm_caller level cost stays None — the
        # recovery is confirmed in TestBaselineCostExtraction (line ~749).
        # This test specifically confirms tokens are estimated (not zero) so
        # the outer fallback has something to work with.
        assert state.turn_prompt_tokens > 0 or state.turn_completion_tokens >= 0

    @pytest.mark.asyncio
    async def test_or_mode_leaves_cost_none_when_no_usage_chunk(self):
        """OR mode must NOT fabricate a cost from the rate card — under-billing
        is preferable to wrong-billing for the OR path (comment in service.py)."""
        state = _BaselineStreamState(model="anthropic/claude-sonnet-4-6")
        mock_client = MagicMock()
        mock_client.chat.completions.create = AsyncMock(
            return_value=_make_stream_mock()
        )

        with (
            patch(
                "backend.copilot.baseline.service._get_main_client",
                return_value=mock_client,
            ),
            patch("backend.copilot.baseline.service.config.use_openrouter", True),
            patch("backend.copilot.baseline.service.config.api_key", "or-key"),
            patch(
                "backend.copilot.baseline.service.config.base_url",
                "https://openrouter.ai/api/v1",
            ),
        ):
            await _baseline_llm_caller(
                messages=[{"role": "user", "content": "hi"}],
                tools=[],
                state=state,
            )

        # OR mode: no usage.cost in chunk → cost_usd stays None (expected).
        assert state.cost_usd is None


class TestSplitUserMessageAfterSkillsBlock:
    """Per-user cache breakpoint at the ``</available_skills>`` boundary.

    The skill index is stable per-user across turns, so caching its bytes
    separately from the user's variable typed text lifts cache hit rate
    on the otherwise-re-tokenised prefix.
    """

    def test_returns_none_when_no_skills_block(self):
        assert _split_user_message_after_skills_block("hello world") is None

    def test_splits_at_skills_block_boundary(self):
        content = (
            "<available_skills>\n"
            "- name: foo — bar\n"
            "</available_skills>\n\n"
            "actual user typing"
        )
        blocks = _split_user_message_after_skills_block(content)
        assert blocks is not None
        assert len(blocks) == 2
        # Prefix carries the cacheable static skill index.
        assert blocks[0]["type"] == "text"
        assert "<available_skills>" in blocks[0]["text"]
        assert blocks[0]["cache_control"]["type"] == "ephemeral"
        # Suffix is the variable user typing — no cache_control.
        assert blocks[1] == {"type": "text", "text": "actual user typing"}

    def test_skills_block_with_preceding_blocks_included_in_prefix(self):
        """When ``<memory_context>`` / ``<env_context>`` etc. precede the
        skills block, they ride along in the cacheable prefix — they're
        also stable per-user-per-session and a separate breakpoint would
        waste one of Anthropic's four available cache slots."""
        content = (
            "<memory_context>\nremember\n</memory_context>\n\n"
            "<available_skills>\n- name: foo — bar\n</available_skills>\n\n"
            "typing"
        )
        blocks = _split_user_message_after_skills_block(content)
        assert blocks is not None
        assert "<memory_context>" in blocks[0]["text"]
        assert "<available_skills>" in blocks[0]["text"]
        assert blocks[1]["text"] == "typing"

    def test_empty_suffix_drops_trailing_block(self):
        """Anthropic 400s on zero-length text blocks — when the user
        typed nothing after the auto-injected skills index, the trailing
        block must be omitted, not emitted empty."""
        content = "<available_skills>\nx\n</available_skills>\n\n"
        blocks = _split_user_message_after_skills_block(content)
        assert blocks is not None
        assert len(blocks) == 1
        assert blocks[0]["cache_control"]["type"] == "ephemeral"


class TestApplySkillsCacheBreakpoint:
    def test_first_user_message_with_skills_block_is_split(self):
        msgs = [
            {"role": "system", "content": "sys"},
            {
                "role": "user",
                "content": (
                    "<available_skills>\n- name: a — b\n</available_skills>\n\nhi"
                ),
            },
            {"role": "assistant", "content": "ok"},
        ]
        out = _apply_skills_cache_breakpoint(msgs)
        # System and assistant pass through untouched.
        assert out[0] == msgs[0]
        assert out[2] == msgs[2]
        # User message becomes a content-block list with the cache marker
        # on the static prefix.
        assert isinstance(out[1]["content"], list)
        assert out[1]["content"][0]["cache_control"]["type"] == "ephemeral"

    def test_no_skills_block_leaves_messages_untouched(self):
        msgs = [
            {"role": "system", "content": "sys"},
            {"role": "user", "content": "hi"},
        ]
        out = _apply_skills_cache_breakpoint(msgs)
        # No breakpoint applied — user content stays a plain string.
        assert out[1]["content"] == "hi"

    def test_only_first_user_message_is_split(self):
        """Later user turns never carry ``<available_skills>`` (it's
        only injected on the first turn), and even if a later message
        did contain it the helper should split the *first* match and
        stop — Anthropic's cache budget is four breakpoints total."""
        msgs = [
            {
                "role": "user",
                "content": "<available_skills>\nx\n</available_skills>\n\nhi",
            },
            {"role": "assistant", "content": "ok"},
            {
                "role": "user",
                "content": "<available_skills>\ny\n</available_skills>\n\nhi2",
            },
        ]
        out = _apply_skills_cache_breakpoint(msgs)
        # First user split; later user message left as a plain string.
        assert isinstance(out[0]["content"], list)
        assert isinstance(out[2]["content"], str)

    def test_pre_blocked_user_content_is_left_alone(self):
        """If a caller already converted the user content to a block
        list, do not double-mark.  Anthropic 400s on the same cache
        marker twice on adjacent blocks."""
        msgs = [
            {
                "role": "user",
                "content": [{"type": "text", "text": "pre-blocked"}],
            }
        ]
        out = _apply_skills_cache_breakpoint(msgs)
        assert out[0]["content"] == [{"type": "text", "text": "pre-blocked"}]

    def test_no_skills_block_preserves_dict_identity(self):
        """When no user message carries an ``<available_skills>`` block,
        every input dict must be returned by-reference so the memoised
        ``cached_system_message`` reference in ``_baseline_llm_caller``
        keeps its identity across rounds."""
        sys_msg = {"role": "system", "content": "sys"}
        usr_msg = {"role": "user", "content": "hi"}
        out = _apply_skills_cache_breakpoint([sys_msg, usr_msg])
        assert out[0] is sys_msg
        assert out[1] is usr_msg

    def test_skills_block_only_modifies_target_message(self):
        """Only the one user message that needs the breakpoint is
        shallow-copied; siblings (system, assistant, later user turns)
        keep their original dict identity."""
        sys_msg = {"role": "system", "content": "sys"}
        usr1 = {
            "role": "user",
            "content": "<available_skills>\n- a\n</available_skills>\n\nhi",
        }
        ast = {"role": "assistant", "content": "ok"}
        usr2 = {"role": "user", "content": "follow-up"}
        out = _apply_skills_cache_breakpoint([sys_msg, usr1, ast, usr2])
        assert out[0] is sys_msg
        assert out[1] is not usr1  # target — gets shallow-copied
        assert out[2] is ast
        assert out[3] is usr2
