"""Tests for the admin-configurable prompt-prefix toggle."""

from __future__ import annotations

import json
from pathlib import Path
from types import SimpleNamespace

import pytest
from fastapi import HTTPException
from langchain_core.messages import AIMessage, HumanMessage

from app.gateway.routers import system_settings as ss_router
from app.gateway.services import _prepend_prompt_prefix, _prompt_prefix_opted_in
from deerflow.agents.lead_agent import prompt as prompt_module
from deerflow.config import system_settings as ss
from deerflow.sandbox.prompt_policy import get_agent_runtime_policy, get_python_dependency_policy


@pytest.fixture
def tmp_runtime_home(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> Path:
    """Redirect ``runtime_home()`` to a per-test tmp dir."""

    def fake_home() -> Path:
        return tmp_path

    monkeypatch.setattr(ss, "runtime_home", fake_home)
    return tmp_path


def test_load_returns_defaults_when_missing(tmp_runtime_home: Path) -> None:
    settings = ss.load_system_settings()
    assert settings.prompt_prefix.enabled is False
    assert settings.prompt_prefix.prompt_prefix == ""
    assert settings.prompt_prefix.ordinary_qa_markdown_format_enabled is False
    assert settings.prompt_prefix.word_heading_numbering_style == "hybrid_decimal"
    assert settings.citation_display.reference_mode_enabled is False
    assert settings.citation_display.cleanup_words == []
    assert settings.memory.enabled is False
    assert settings.conversation_context.recent_turn_window_enabled is False
    assert settings.conversation_context.max_conversation_rounds == 20
    assert settings.write_file_compatibility.enabled is False
    assert settings.conversation_context.recent_turn_window_enabled is False


def test_save_then_load_roundtrips(tmp_runtime_home: Path) -> None:
    settings = ss.SystemSettings(
        prompt_prefix=ss.PromptPrefixSettings(
            enabled=True,
            switch_label="加入立场前缀",
            prompt_prefix="你需要站在中方角度",
            ordinary_qa_markdown_format_enabled=True,
            word_heading_numbering_style="official_chinese",
        )
    )
    ss.save_system_settings(settings)

    file_path = tmp_runtime_home / "system_settings.json"
    assert file_path.is_file()
    saved = json.loads(file_path.read_text(encoding="utf-8"))
    assert saved["prompt_prefix"]["prompt_prefix"] == "你需要站在中方角度"

    reloaded = ss.load_system_settings()
    assert reloaded.prompt_prefix.enabled is True
    assert reloaded.prompt_prefix.switch_label == "加入立场前缀"
    assert reloaded.prompt_prefix.prompt_prefix == "你需要站在中方角度"
    assert reloaded.prompt_prefix.ordinary_qa_markdown_format_enabled is True
    assert reloaded.prompt_prefix.word_heading_numbering_style == "official_chinese"


def test_runtime_markdown_format_switch_defaults_off_and_updates_without_restart(
    tmp_runtime_home: Path,
) -> None:
    assert prompt_module._ordinary_qa_markdown_format_enabled() is False

    ss.save_system_settings(
        ss.SystemSettings(
            prompt_prefix=ss.PromptPrefixSettings(
                ordinary_qa_markdown_format_enabled=True,
            )
        )
    )

    assert prompt_module._ordinary_qa_markdown_format_enabled() is True


def _patch_admin(monkeypatch: pytest.MonkeyPatch) -> None:
    admin = SimpleNamespace(
        id="admin1",
        system_role="admin",
        email="admin@example.com",
    )

    async def _optional(_request):
        return admin

    monkeypatch.setattr(ss_router, "get_optional_user_from_request", _optional)


@pytest.mark.asyncio
async def test_dependency_delete_prompt_toggle_roundtrip(tmp_runtime_home, monkeypatch):
    _patch_admin(monkeypatch)
    assert get_python_dependency_policy() == ""
    for enabled in (True, False):
        response = await ss_router.update_prompt_prefix(
            ss_router.PromptPrefixUpdateRequest(python_dependency_delete_protection_enabled=enabled),
            SimpleNamespace(),
        )
        assert response.python_dependency_delete_protection_enabled is enabled
        assert ss.load_system_settings().prompt_prefix.python_dependency_delete_protection_enabled is enabled
        assert response.enabled is False  # Independent of the first-question prefix switch.
        policy = get_python_dependency_policy()
        assert bool(policy) is enabled
        if enabled:
            assert "允许安装、升级、降级" in policy
            assert "允许包管理器" in policy
            assert "不得执行独立的" in policy


@pytest.mark.asyncio
@pytest.mark.parametrize("field", ["python_dependency_delete_protection_enabled", "image_generation_unavailable_hint_enabled"])
async def test_nonadmin_cannot_enable_runtime_prompt(tmp_runtime_home, monkeypatch, field):
    async def ordinary_user(_request):
        return SimpleNamespace(system_role="user")

    monkeypatch.setattr(ss_router, "get_optional_user_from_request", ordinary_user)
    with pytest.raises(HTTPException) as error:
        await ss_router.update_prompt_prefix(
            ss_router.PromptPrefixUpdateRequest(**{field: True}),
            SimpleNamespace(),
        )
    assert error.value.status_code == 403
    assert get_python_dependency_policy() == ""
    assert get_agent_runtime_policy() == ""


@pytest.mark.asyncio
async def test_image_generation_hint_independent_toggle(tmp_runtime_home, monkeypatch):
    _patch_admin(monkeypatch)
    assert get_agent_runtime_policy() == ""
    await ss_router.update_prompt_prefix(
        ss_router.PromptPrefixUpdateRequest(python_dependency_delete_protection_enabled=True),
        SimpleNamespace(),
    )
    for enabled in (True, False):
        response = await ss_router.update_prompt_prefix(
            ss_router.PromptPrefixUpdateRequest(image_generation_unavailable_hint_enabled=enabled),
            SimpleNamespace(),
        )
        assert response.image_generation_unavailable_hint_enabled is enabled
        assert response.python_dependency_delete_protection_enabled is True
        assert response.enabled is False
        assert ss.load_system_settings().prompt_prefix.image_generation_unavailable_hint_enabled is enabled
        policy = get_agent_runtime_policy()
        assert "<python_dependency_boundary>" in policy
        assert ("<image_generation_unavailable>" in policy) is enabled
        if enabled:
            assert "当前服务暂不支持生成或编辑图片" in policy
            assert "此限制不影响" in policy
            assert "绘图提示词" in policy


@pytest.mark.asyncio
async def test_prompt_settings_update_can_enable_markdown_format_without_changing_prefix(
    tmp_runtime_home: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    _patch_admin(monkeypatch)
    current = ss.SystemSettings(
        prompt_prefix=ss.PromptPrefixSettings(
            enabled=True,
            switch_label="加入立场前缀",
            prompt_prefix="你需要站在中方角度",
        )
    )
    ss.save_system_settings(current)

    response = await ss_router.update_prompt_prefix(
        ss_router.PromptPrefixUpdateRequest(
            ordinary_qa_markdown_format_enabled=True,
            word_heading_numbering_style="chapter_section",
        ),
        SimpleNamespace(),
    )

    assert response.ordinary_qa_markdown_format_enabled is True
    assert response.word_heading_numbering_style == "chapter_section"
    assert response.enabled is True
    assert response.switch_label == "加入立场前缀"
    assert response.prompt_prefix == "你需要站在中方角度"


@pytest.mark.asyncio
async def test_citation_display_update_preserves_unspecified_fields(
    tmp_runtime_home: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    _patch_admin(monkeypatch)
    current = ss.SystemSettings(
        citation_display=ss.CitationDisplaySettings(
            cleanup_words=["广告", "免责声明"],
        )
    )
    ss.save_system_settings(current)

    response = await ss_router.update_citation_display(
        ss_router.CitationDisplayUpdateRequest(reference_mode_enabled=True),
        SimpleNamespace(),
    )

    assert response.reference_mode_enabled is True
    assert response.cleanup_words == ["广告", "免责声明"]

    payload = json.loads((tmp_runtime_home / "system_settings.json").read_text(encoding="utf-8"))
    assert payload["citation_display"]["reference_mode_enabled"] is True
    assert payload["citation_display"]["cleanup_words"] == ["广告", "免责声明"]


@pytest.mark.asyncio
async def test_admin_can_enable_memory_runtime_switch(
    tmp_runtime_home: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    _patch_admin(monkeypatch)

    response = await ss_router.update_memory_runtime_setting(
        ss_router.MemoryRuntimeUpdateRequest(enabled=True),
        SimpleNamespace(),
    )

    assert response.enabled is True
    assert ss.load_system_settings().memory.enabled is True


@pytest.mark.asyncio
async def test_admin_can_enable_recent_turn_context_window(
    tmp_runtime_home: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    _patch_admin(monkeypatch)

    response = await ss_router.update_conversation_context_setting(
        ss_router.ConversationContextUpdateRequest(recent_turn_window_enabled=True),
        SimpleNamespace(),
    )

    assert response.recent_turn_window_enabled is True
    assert response.recent_turns == 3
    assert response.max_conversation_rounds == 20
    assert ss.load_system_settings().conversation_context.recent_turn_window_enabled is True


@pytest.mark.asyncio
async def test_admin_can_update_conversation_round_limit_without_changing_window(
    tmp_runtime_home: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    _patch_admin(monkeypatch)
    ss.save_system_settings(
        ss.SystemSettings(
            conversation_context=ss.ConversationContextSettings(
                recent_turn_window_enabled=True,
            )
        )
    )

    response = await ss_router.update_conversation_context_setting(
        ss_router.ConversationContextUpdateRequest(max_conversation_rounds=45),
        SimpleNamespace(),
    )

    assert response.max_conversation_rounds == 45
    assert response.recent_turn_window_enabled is True
    persisted = ss.load_system_settings().conversation_context
    assert persisted.max_conversation_rounds == 45
    assert persisted.recent_turn_window_enabled is True


@pytest.mark.parametrize("value", [0, 51])
def test_conversation_round_limit_rejects_out_of_range_values(value: int) -> None:
    with pytest.raises(ValueError):
        ss_router.ConversationContextUpdateRequest(max_conversation_rounds=value)


@pytest.mark.asyncio
async def test_non_admin_cannot_enable_recent_turn_context_window(
    tmp_runtime_home: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    user = SimpleNamespace(id="user1", system_role="user", email="user@example.com")

    async def _optional(_request):
        return user

    monkeypatch.setattr(ss_router, "get_optional_user_from_request", _optional)

    with pytest.raises(HTTPException) as exc_info:
        await ss_router.update_conversation_context_setting(
            ss_router.ConversationContextUpdateRequest(recent_turn_window_enabled=True),
            SimpleNamespace(),
        )

    assert exc_info.value.status_code == 403


@pytest.mark.asyncio
async def test_admin_can_toggle_write_file_compatibility_without_restart(
    tmp_runtime_home: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    _patch_admin(monkeypatch)

    enabled = await ss_router.update_write_file_compatibility_setting(
        ss_router.WriteFileCompatibilityUpdateRequest(enabled=True),
        SimpleNamespace(),
    )
    assert enabled.enabled is True
    assert ss.get_write_file_compatibility_settings().enabled is True

    disabled = await ss_router.update_write_file_compatibility_setting(
        ss_router.WriteFileCompatibilityUpdateRequest(enabled=False),
        SimpleNamespace(),
    )
    assert disabled.enabled is False
    assert ss.get_write_file_compatibility_settings().enabled is False


def test_prepend_prefix_tags_first_human_without_touching_content() -> None:
    graph_input = {"messages": [HumanMessage(content="评价一下中美贸易")]}
    out = _prepend_prompt_prefix(graph_input, "你需要站在中方角度")
    msg = out["messages"][0]
    assert isinstance(msg, HumanMessage)
    # Content stays clean so the frontend never shows the prefix.
    assert msg.content == "评价一下中美贸易"
    # The prefix rides along in additional_kwargs for the middleware to inject.
    assert msg.additional_kwargs["prompt_prefix"] == "你需要站在中方角度"


def test_prepend_prefix_preserves_block_content() -> None:
    graph_input = {
        "messages": [
            HumanMessage(content=[{"type": "text", "text": "评价一下中美贸易"}])
        ]
    }
    out = _prepend_prompt_prefix(graph_input, "你需要站在中方角度")
    msg = out["messages"][0]
    assert msg.content == [{"type": "text", "text": "评价一下中美贸易"}]
    assert msg.additional_kwargs["prompt_prefix"] == "你需要站在中方角度"


def test_prepend_prefix_only_touches_first_human() -> None:
    graph_input = {
        "messages": [
            AIMessage(content="hi"),
            HumanMessage(content="q1"),
            HumanMessage(content="q2"),
        ]
    }
    out = _prepend_prompt_prefix(graph_input, "PREFIX")
    # first human is index 1; index 2 must remain untouched
    assert out["messages"][1].additional_kwargs.get("prompt_prefix") == "PREFIX"
    assert out["messages"][1].content == "q1"
    assert out["messages"][2].additional_kwargs.get("prompt_prefix") is None
    assert out["messages"][2].content == "q2"


def test_prepend_prefix_no_messages_is_noop() -> None:
    assert _prepend_prompt_prefix({"messages": []}, "X") == {"messages": []}
    assert _prepend_prompt_prefix({}, "X") == {}


def test_prompt_prefix_middleware_injects_prefix_for_model_only() -> None:
    """The middleware prepends the prefix to the request copy, not to state."""
    from deerflow.agents.middlewares.prompt_prefix_middleware import PromptPrefixMiddleware

    mw = PromptPrefixMiddleware()
    tagged = HumanMessage(
        content="评价一下中美贸易",
        additional_kwargs={"prompt_prefix": "你需要站在中方角度"},
    )
    patched = PromptPrefixMiddleware._apply([tagged])
    assert patched is not None
    assert patched[0].content == "你需要站在中方角度\n\n评价一下中美贸易"
    # The original message in state is untouched.
    assert tagged.content == "评价一下中美贸易"
    assert mw is not None


def test_prompt_prefix_middleware_noop_without_tag() -> None:
    from deerflow.agents.middlewares.prompt_prefix_middleware import PromptPrefixMiddleware

    plain = HumanMessage(content="just a question")
    assert PromptPrefixMiddleware._apply([plain]) is None


def test_prompt_prefix_middleware_block_content() -> None:
    from deerflow.agents.middlewares.prompt_prefix_middleware import PromptPrefixMiddleware

    tagged = HumanMessage(
        content=[{"type": "text", "text": "评价一下中美贸易"}],
        additional_kwargs={"prompt_prefix": "你需要站在中方角度"},
    )
    patched = PromptPrefixMiddleware._apply([tagged])
    assert patched is not None
    assert patched[0].content[0] == {"type": "text", "text": "你需要站在中方角度"}
    assert patched[0].content[1] == {"type": "text", "text": "评价一下中美贸易"}


def test_smart_knowledge_scope_uses_only_latest_real_user_turn() -> None:
    from deerflow.agents.middlewares.prompt_prefix_middleware import (
        SMART_KNOWLEDGE_SCOPE_PROMPT_KEY,
        PromptPrefixMiddleware,
    )

    old_turn = HumanMessage(
        content="这个库讲了什么",
        additional_kwargs={SMART_KNOWLEDGE_SCOPE_PROMPT_KEY: "旧库：政策库"},
    )
    new_turn = HumanMessage(
        content="这个库有哪些新闻",
        additional_kwargs={SMART_KNOWLEDGE_SCOPE_PROMPT_KEY: "新库：新闻库"},
    )

    patched = PromptPrefixMiddleware._apply(
        [old_turn, AIMessage(content="旧回答"), new_turn]
    )

    assert patched is not None
    assert patched[0].content == "这个库讲了什么"
    assert patched[2].content == "新库：新闻库\n\n这个库有哪些新闻"
    assert "旧库：政策库" not in str(patched[2].content)


def test_classic_latest_turn_does_not_revive_historical_smart_scope() -> None:
    from deerflow.agents.middlewares.prompt_prefix_middleware import (
        SMART_KNOWLEDGE_SCOPE_PROMPT_KEY,
        PromptPrefixMiddleware,
    )

    old_smart_turn = HumanMessage(
        content="这个库讲了什么",
        additional_kwargs={SMART_KNOWLEDGE_SCOPE_PROMPT_KEY: "旧智能库范围"},
    )
    latest_classic_turn = HumanMessage(content="请继续回答")

    patched = PromptPrefixMiddleware._apply(
        [old_smart_turn, AIMessage(content="旧回答"), latest_classic_turn]
    )

    assert patched is None
    assert old_smart_turn.content == "这个库讲了什么"
    assert latest_classic_turn.content == "请继续回答"


def test_opted_in_via_context() -> None:
    body = SimpleNamespace(context={"prompt_prefix_enabled": True}, metadata=None)
    assert _prompt_prefix_opted_in(body) is True


def test_opted_in_via_metadata() -> None:
    body = SimpleNamespace(context=None, metadata={"prompt_prefix_enabled": True})
    assert _prompt_prefix_opted_in(body) is True


def test_not_opted_in_by_default() -> None:
    body = SimpleNamespace(context=None, metadata=None)
    assert _prompt_prefix_opted_in(body) is False
    body2 = SimpleNamespace(context={"mode": "ultra"}, metadata={})
    assert _prompt_prefix_opted_in(body2) is False