"""Regression test: DeepSeek V4 thinking mode reasoning_content echo.

DeepSeek V4-flash / V4-pro thinking mode requires ``reasoning_content`` on
every assistant message that carries ``tool_calls``. When a persisted
session replays an assistant tool-call turn that was recorded without the
field, DeepSeek rejects the next request with HTTP 400::

    The reasoning_content in the thinking mode must be passed back to the API.

Fix covers three paths:

1. ``_build_assistant_message`` — new tool-call messages without raw
   reasoning_content get ``" "`` pinned at creation time so nothing gets
   persisted poisoned.
2. ``_copy_reasoning_content_for_api`` — already-poisoned history replays
   with ``reasoning_content=" "`` injected defensively.
3. Detection covers three signals: ``provider == "deepseek"``,
   ``"deepseek" in model``, and ``api.deepseek.com`` host match. The third
   catches custom-provider setups pointing at DeepSeek.

The placeholder is a single space (not empty string) because DeepSeek V4 Pro
tightened validation and rejects empty-string reasoning_content with a
400 ("The reasoning content in the thinking mode must be passed back to
the API"). A space satisfies non-empty checks everywhere without leaking
fabricated reasoning.

Refs #15250 / #15353 / #17341.
"""

from __future__ import annotations

from types import SimpleNamespace

import pytest

from run_agent import AIAgent


def _make_agent(provider: str = "", model: str = "", base_url: str = "") -> AIAgent:
    agent = object.__new__(AIAgent)
    agent.provider = provider
    agent.model = model
    agent.base_url = base_url
    agent.verbose_logging = False
    agent.reasoning_callback = None
    agent.stream_delta_callback = None
    agent._stream_callback = None
    return agent


_ATTR_ABSENT = object()
_EXPECT_NOT_PRESENT = object()


def _sdk_tool_call(call_id: str = "c1", name: str = "terminal", arguments: str = "{}"):
    """Minimal SDK-shaped tool_call object that satisfies the builder's iteration."""
    return SimpleNamespace(
        id=call_id,
        call_id=call_id,
        type="function",
        function=SimpleNamespace(name=name, arguments=arguments),
        extra_content=None,
    )


def _build_sdk_message(reasoning_content=_ATTR_ABSENT, **extra):
    """SDK-shaped assistant message; ``reasoning_content`` defaults to absent."""
    kwargs = {"content": "", **extra}
    if reasoning_content is not _ATTR_ABSENT:
        kwargs["reasoning_content"] = reasoning_content
    return SimpleNamespace(**kwargs)


class TestNeedsDeepSeekToolReasoning:
    """_needs_deepseek_tool_reasoning() recognises all three detection signals."""

    def test_provider_deepseek(self) -> None:
        agent = _make_agent(provider="deepseek", model="deepseek-v4-flash")
        assert agent._needs_deepseek_tool_reasoning() is True







class TestCopyReasoningContentForApi:
    """_copy_reasoning_content_for_api pads reasoning_content for DeepSeek tool-calls."""

    def test_deepseek_tool_call_poisoned_history_gets_space_placeholder(self) -> None:
        """Already-poisoned history (no reasoning_content, no reasoning) gets ' '."""
        agent = _make_agent(provider="deepseek", model="deepseek-v4-flash")
        source = {
            "role": "assistant",
            "content": "",
            "tool_calls": [{"id": "c1", "function": {"name": "terminal"}}],
        }
        api_msg: dict = {}
        agent._copy_reasoning_content_for_api(source, api_msg)
        assert api_msg.get("reasoning_content") == " "




    def test_non_thinking_provider_strips_empty_reasoning_content(self) -> None:
        """Strict OpenAI-compatible providers (Mistral, Cerebras, …) reject ANY
        reasoning_content key in input messages — even an empty string — with
        HTTP 400/422. On a non-thinking provider the field must be stripped,
        not round-tripped. Refs #45655.
        """
        agent = _make_agent(
            provider="openrouter",
            model="anthropic/claude-sonnet-4.6",
            base_url="https://openrouter.ai/api/v1",
        )
        source = {
            "role": "assistant",
            "content": "hi",
            "reasoning_content": "",
        }
        api_msg: dict = {}
        agent._copy_reasoning_content_for_api(source, api_msg)
        assert "reasoning_content" not in api_msg










class TestBuildAssistantMessageDeepSeekReasoningContent:
    """_build_assistant_message pins replay-safe DeepSeek tool-call state."""

    def test_deepseek_tool_call_reasoning_is_backfilled_into_reasoning_content(self) -> None:
        agent = _make_agent(provider="deepseek", model="deepseek-v4-flash")
        assistant_message = SimpleNamespace(
            content=None,
            reasoning="DeepSeek tool-call reasoning",
            reasoning_content=None,
            reasoning_details=None,
            codex_reasoning_items=None,
            codex_message_items=None,
            tool_calls=[
                SimpleNamespace(
                    id="call_1",
                    call_id=None,
                    response_item_id=None,
                    type="function",
                    function=SimpleNamespace(name="terminal", arguments="{}"),
                )
            ],
        )

        msg = agent._build_assistant_message(assistant_message, "tool_calls")

        assert msg["reasoning_content"] == "DeepSeek tool-call reasoning"
        assert msg["tool_calls"][0]["id"] == "call_1"




class TestBuildAssistantMessagePadsStrictProviders:
    """Regression for #17400: _build_assistant_message must pin reasoning_content
    on tool-call turns when the active provider enforces echo-back, regardless
    of whether the SDK exposed reasoning_content as None, omitted it entirely,
    or returned an empty thinking block.

    Prior to the fix, the pad branch was guarded by ``msg.get("tool_calls")``,
    which was always falsy because tool_calls were assigned later in the same
    method. Persisted history accumulated assistant tool-call turns with no
    reasoning_content; the next replay 400'd on DeepSeek/Kimi.
    """

    @pytest.mark.parametrize(
        "provider,model,base_url,sdk_reasoning_content,expected",
        [
            pytest.param(
                "deepseek", "deepseek-v4-pro", "",
                None, " ",
                id="deepseek-attr-none",
            ),
            pytest.param(
                "deepseek", "deepseek-v4-pro", "",
                _ATTR_ABSENT, " ",
                id="deepseek-attr-absent",
            ),
            pytest.param(
                "kimi-coding", "kimi-k2.6", "",
                None, " ",
                id="kimi-attr-none",
            ),
            pytest.param(
                "custom", "kimi-k2", "https://api.moonshot.ai/v1",
                _ATTR_ABSENT, " ",
                id="moonshot-base-url",
            ),
            pytest.param(
                "openrouter", "anthropic/claude-sonnet-4.6", "https://openrouter.ai/api/v1",
                _ATTR_ABSENT, _EXPECT_NOT_PRESENT,
                id="openrouter-no-pad",
            ),
        ],
    )
    def test_tool_call_reasoning_content_pad(
        self, provider, model, base_url, sdk_reasoning_content, expected,
    ) -> None:
        agent = _make_agent(provider=provider, model=model, base_url=base_url)
        msg_in = _build_sdk_message(
            reasoning_content=sdk_reasoning_content,
            tool_calls=[_sdk_tool_call()],
        )
        msg = agent._build_assistant_message(msg_in, finish_reason="tool_calls")
        if expected is _EXPECT_NOT_PRESENT:
            assert "reasoning_content" not in msg
        else:
            assert msg["reasoning_content"] == expected

    def test_tool_call_preserves_real_reasoning_content(self) -> None:
        agent = _make_agent(provider="deepseek", model="deepseek-v4-pro")
        msg_in = _build_sdk_message(
            reasoning_content="actual chain of thought",
            tool_calls=[_sdk_tool_call()],
        )
        msg = agent._build_assistant_message(msg_in, finish_reason="tool_calls")
        assert msg["reasoning_content"] == "actual chain of thought"

    def test_text_only_turn_not_padded_by_tool_call_branch(self) -> None:
        """Plain-text turns rely on _copy_reasoning_content_for_api at replay
        time, not on this builder's tool-call pad."""
        agent = _make_agent(provider="deepseek", model="deepseek-v4-pro")
        msg_in = SimpleNamespace(content="hello", tool_calls=None)
        msg = agent._build_assistant_message(msg_in, finish_reason="stop")
        assert "tool_calls" not in msg
        assert "reasoning_content" not in msg



class TestNeedsKimiToolReasoning:
    """The extracted _needs_kimi_tool_reasoning() helper keeps Kimi behavior intact."""

    @pytest.mark.parametrize(
        "provider,base_url",
        [
            ("kimi-coding", ""),
            ("kimi-coding-cn", ""),
            ("custom", "https://api.kimi.com/v1"),
            ("custom", "https://api.moonshot.ai/v1"),
            ("custom", "https://api.moonshot.cn/v1"),
        ],
    )
    def test_kimi_signals(self, provider: str, base_url: str) -> None:
        agent = _make_agent(provider=provider, model="kimi-k2", base_url=base_url)
        assert agent._needs_kimi_tool_reasoning() is True



class TestReapplyReasoningEchoForProviderSwitch:
    """Mid-conversation fallover to a require-side provider must re-pad.

    ``api_messages`` is built once, before the retry loop, while the *primary*
    provider is active. When a fallback then switches to DeepSeek/Kimi/MiMo,
    assistant turns that were built under a non-require primary (e.g. Codex,
    which uses encrypted reasoning, not ``reasoning_content``) go out bare and
    the new provider 400s with "reasoning_content must be passed back".

    ``reapply_reasoning_echo_for_provider`` re-applies the pad against the
    *current* provider right before the request is built. It is idempotent and
    a no-op unless the active provider enforces echo-back.
    """

    @staticmethod
    def _codex_built_history() -> list[dict]:
        """Assistant turns as built under a Codex primary: some carry a
        reasoning summary (stored as reasoning_content), some are bare."""
        return [
            {"role": "system", "content": "sys"},
            {"role": "user", "content": "do the thing"},
            {  # turn that emitted a reasoning summary
                "role": "assistant",
                "content": "",
                "reasoning_content": "summary from codex",
                "tool_calls": [{"id": "c1", "function": {"name": "terminal"}}],
            },
            {"role": "tool", "tool_call_id": "c1", "content": "ok"},
            {  # bare tool-call turn (Codex emitted no summary)
                "role": "assistant",
                "content": "",
                "tool_calls": [{"id": "c2", "function": {"name": "terminal"}}],
            },
            {"role": "tool", "tool_call_id": "c2", "content": "ok"},
        ]

    def test_switch_to_deepseek_pads_bare_turns(self) -> None:
        from agent.agent_runtime_helpers import reapply_reasoning_echo_for_provider

        agent = _make_agent(provider="deepseek", model="deepseek-v4-pro")
        msgs = self._codex_built_history()
        padded = reapply_reasoning_echo_for_provider(agent, msgs)
        assert padded == 1
        bare = [m for m in msgs if m.get("role") == "assistant" and not m.get("reasoning_content")]
        assert bare == []
        # existing summary preserved verbatim, not clobbered with the pad
        assert msgs[2]["reasoning_content"] == "summary from codex"
        assert msgs[4]["reasoning_content"] == " "

    def test_strips_stale_pad_under_strict_provider(self) -> None:
        """Switching TO a strict provider (Codex/Mistral/Cerebras) must STRIP
        stale reasoning_content baked in under a reasoning primary, otherwise
        the fallback request 400/422s ("Extra inputs are not permitted").
        Refs #45655 — DeepSeek primary → Mistral fallback 422 on the " " pad.
        """
        from agent.agent_runtime_helpers import reapply_reasoning_echo_for_provider

        agent = _make_agent(
            provider="openai-codex",
            model="gpt-5.5",
            base_url="https://chatgpt.com/backend-api/codex",
        )
        msgs = self._codex_built_history()
        changed = reapply_reasoning_echo_for_provider(agent, msgs)
        # msgs[2] carried "summary from codex" — must be stripped for the
        # strict provider; the bare turn (msgs[4]) stays bare.
        assert changed == 1
        assert "reasoning_content" not in msgs[2]
        assert "reasoning_content" not in msgs[4]




class TestReasoningPrimaryToStrictFallback:
    """Regression: reasoning primary → strict fallback must not 422.

    User report (HTTP 422): a DeepSeek V4 Pro primary pads tool-call turns
    with ``reasoning_content=" "``; a mid-session fallback to Mistral
    (mistral-small) replays those pads and Mistral rejects them with::

        body.messages.2.assistant.reasoning_content: Extra inputs are not
        permitted  (input: ' ')

    api_messages is built once under the primary, so the stale pad survives
    into the fallback request. reapply_reasoning_echo_for_provider() must
    strip it when the active provider doesn't enforce echo-back. Refs #45655.
    """

    @staticmethod
    def _deepseek_built_history() -> list[dict]:
        """Multi-turn history as built under a DeepSeek primary — tool-call
        turns padded with " " at indices 2 and 6 (matching the report)."""
        return [
            {"role": "system", "content": "sys"},
            {"role": "user", "content": "u1"},
            {"role": "assistant", "reasoning_content": " ",
             "tool_calls": [{"id": "a", "function": {"name": "terminal"}}]},
            {"role": "tool", "tool_call_id": "a", "content": "ok"},
            {"role": "assistant", "content": "done"},
            {"role": "user", "content": "u2"},
            {"role": "assistant", "reasoning_content": " ",
             "tool_calls": [{"id": "b", "function": {"name": "terminal"}}]},
            {"role": "tool", "tool_call_id": "b", "content": "ok"},
        ]

    def test_mistral_fallback_strips_space_pad(self) -> None:
        from agent.agent_runtime_helpers import reapply_reasoning_echo_for_provider

        mistral = _make_agent(
            provider="mistral",
            model="mistral-small-latest",
            base_url="https://api.mistral.ai/v1",
        )
        msgs = self._deepseek_built_history()
        changed = reapply_reasoning_echo_for_provider(mistral, msgs)
        assert changed == 2  # both padded tool-call turns
        leaks = [i for i, m in enumerate(msgs) if "reasoning_content" in m]
        assert leaks == []

    def test_roundtrip_back_to_deepseek_repads(self) -> None:
        """Strict fallback strips, then switching back to DeepSeek re-pads —
        no regression on the #15748 echo-back requirement."""
        from agent.agent_runtime_helpers import reapply_reasoning_echo_for_provider

        msgs = self._deepseek_built_history()
        mistral = _make_agent(
            provider="mistral", model="mistral-small-latest",
            base_url="https://api.mistral.ai/v1",
        )
        reapply_reasoning_echo_for_provider(mistral, msgs)
        deepseek = _make_agent(provider="deepseek", model="deepseek-v4-pro")
        reapply_reasoning_echo_for_provider(deepseek, msgs)
        assert msgs[2]["reasoning_content"] == " "
        assert msgs[6]["reasoning_content"] == " "

    def test_copy_strips_space_pad_for_mistral(self) -> None:
        """copy_reasoning_content_for_api strips the " " pad on the rebuild
        path too (covers fresh api_messages built under the strict provider)."""
        mistral = _make_agent(
            provider="mistral", model="mistral-small-latest",
            base_url="https://api.mistral.ai/v1",
        )
        source = {"role": "assistant", "reasoning_content": " ",
                  "tool_calls": [{"id": "a"}]}
        api_msg: dict = {"role": "assistant", "tool_calls": [{"id": "a"}]}
        mistral._copy_reasoning_content_for_api(source, api_msg)
        assert "reasoning_content" not in api_msg
