from __future__ import annotations

import pytest
import yaml

from app.gateway.routers.agents import (
    _backfill_listing_cache,
    _load_config_extras,
    _write_agent_config,
)
from app.gateway.services import build_run_config
from deerflow.config import paths as paths_module
from deerflow.config.agents_config import AgentConfig
from deerflow.persistence.agents import MemoryAgentStore


@pytest.fixture
def isolated_paths(tmp_path, monkeypatch):
    """Redirect get_paths() at a fresh temp directory for the duration of one test."""
    monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path))
    monkeypatch.setattr(paths_module, "_paths", None)
    yield tmp_path
    monkeypatch.setattr(paths_module, "_paths", None)


@pytest.mark.asyncio
async def test_agent_store_allows_duplicate_chinese_names_and_filters_visibility() -> None:
    store = MemoryAgentStore()
    chinese_name = "\u4e2d\u6587\u52a9\u624b"

    await store.ensure_builtin({"id": "builtin-agent", "name": chinese_name, "description": "built in"})
    await store.create({"id": "private-a", "name": chinese_name, "description": "private", "user_id": "user-a"})
    await store.create({"id": "published-b", "name": chinese_name, "description": "published", "user_id": "user-b", "published": True})

    user_a_ids = {agent["id"] for agent in await store.list_visible("user-a")}
    assert user_a_ids == {"builtin-agent", "private-a", "published-b"}

    user_c_ids = {agent["id"] for agent in await store.list_visible("user-c")}
    assert user_c_ids == {"builtin-agent", "published-b"}


@pytest.mark.asyncio
async def test_agent_store_only_owner_can_update_or_delete() -> None:
    store = MemoryAgentStore()

    await store.ensure_builtin({"id": "builtin-agent", "name": "Built in"})
    await store.create({"id": "private-a", "name": "Owned", "user_id": "user-a"})

    assert await store.update("builtin-agent", "user-a", {"name": "Nope"}) is None
    assert await store.update("private-a", "user-b", {"name": "Nope"}) is None

    updated = await store.update("private-a", "user-a", {"name": "Renamed"})
    assert updated is not None
    assert updated["name"] == "Renamed"

    assert await store.delete("private-a", "user-b") is False
    assert await store.delete("private-a", "user-a") is True


@pytest.mark.asyncio
async def test_delete_builtin_only_targets_builtins() -> None:
    store = MemoryAgentStore()

    await store.ensure_builtin({"id": "builtin-agent", "name": "Built in"})
    await store.create({"id": "owned-agent", "name": "Owned", "user_id": "user-a"})

    # delete_builtin refuses to touch a user-owned agent
    assert await store.delete_builtin("owned-agent") is False
    assert await store.get_any("owned-agent") is not None

    # delete_builtin succeeds on a built-in
    assert await store.delete_builtin("builtin-agent") is True
    assert await store.get_any("builtin-agent") is None

    # And no-ops on a missing id
    assert await store.delete_builtin("does-not-exist") is False


@pytest.mark.asyncio
async def test_update_builtin_only_targets_builtins() -> None:
    store = MemoryAgentStore()

    await store.ensure_builtin({"id": "builtin-agent", "name": "Built in"})
    await store.create({"id": "owned-agent", "name": "Owned", "user_id": "user-a"})

    # update_builtin succeeds on a built-in agent
    updated = await store.update_builtin("builtin-agent", {"name": "Renamed by admin"})
    assert updated is not None
    assert updated["name"] == "Renamed by admin"

    # update_builtin refuses to touch a user-owned agent (returns None)
    assert await store.update_builtin("owned-agent", {"name": "Hijacked"}) is None

    # And on a missing agent
    assert await store.update_builtin("does-not-exist", {"name": "Ghost"}) is None


@pytest.mark.asyncio
async def test_get_editable_or_raise_admin_can_edit_builtin() -> None:
    """Route helper: admin gets edit access to any agent; non-admin still 403."""
    from fastapi import HTTPException

    from app.gateway.routers.agents import _get_editable_or_raise

    store = MemoryAgentStore()
    await store.ensure_builtin({"id": "builtin-agent", "name": "Built in"})
    await store.create({"id": "owned-agent", "name": "Owned", "user_id": "user-a"})

    # Owner edits their own agent → returns ('owner', record)
    record, mode = await _get_editable_or_raise(
        store, "owned-agent", "user-a", is_admin=False
    )
    assert mode == "owner" and record["id"] == "owned-agent"

    # Admin edits a built-in agent → returns ('admin', record)
    record, mode = await _get_editable_or_raise(
        store, "builtin-agent", "user-a", is_admin=True
    )
    assert mode == "admin" and record["id"] == "builtin-agent"

    # Non-admin tries to edit a built-in → 403
    with pytest.raises(HTTPException) as exc:
        await _get_editable_or_raise(store, "builtin-agent", "user-a", is_admin=False)
    assert exc.value.status_code == 403

    # Admin edits someone else's custom agent → returns ('admin', record).
    # Behavior changed: admins manage the full agent catalog (built-in + any
    # user's agents), not just built-ins. Non-admins are still owner-only.
    await store.create(
        {"id": "published-other", "name": "Pub", "user_id": "user-x", "published": True}
    )
    record, mode = await _get_editable_or_raise(
        store, "published-other", "user-a", is_admin=True
    )
    assert mode == "admin" and record["id"] == "published-other"


def test_build_run_config_injects_custom_assistant_as_agent_id() -> None:
    config = build_run_config("thread-1", None, None, assistant_id="agent-123")

    assert config["configurable"]["thread_id"] == "thread-1"
    assert config["configurable"]["agent_id"] == "agent-123"
    assert "agent_name" not in config["configurable"]


def test_write_agent_config_persists_skills_into_config_yaml(isolated_paths) -> None:
    agent_id = "agent-001"
    _write_agent_config(
        agent_id,
        name="中文助手",
        description="desc",
        skills=["s1", "s2"],
        model="gpt-x",
        tool_groups=["tg-a"],
    )

    config_file = paths_module.get_paths().agent_dir(agent_id) / "config.yaml"
    raw = yaml.safe_load(config_file.read_text(encoding="utf-8"))

    assert raw["id"] == agent_id
    assert raw["name"] == "中文助手"
    assert raw["description"] == "desc"
    assert raw["skills"] == ["s1", "s2"]
    assert raw["model"] == "gpt-x"
    assert raw["tool_groups"] == ["tg-a"]


def test_load_config_extras_returns_values_written_to_yaml(isolated_paths) -> None:
    agent_id = "agent-002"
    _write_agent_config(
        agent_id,
        name="agent-002",
        description="",
        skills=["s1"],
        model=None,
        tool_groups=None,
    )

    extras = _load_config_extras(agent_id)
    assert extras == {"skills": ["s1"], "model": None, "tool_groups": None, "common_questions": None, "template_name": None, "seat_skill_directive": None}


def test_load_config_extras_missing_directory_returns_none_triple(isolated_paths) -> None:
    extras = _load_config_extras("does-not-exist")
    assert extras == {"skills": None, "model": None, "tool_groups": None, "common_questions": None, "template_name": None, "seat_skill_directive": None}


def test_write_agent_config_persists_common_questions(isolated_paths) -> None:
    agent_id = "agent-cq-1"
    # Structured {title, prompt} questions round-trip as clean dict mappings.
    questions = [
        {"title": "能做什么", "prompt": "这个智能体能做什么？"},
        {"title": "上传文件", "prompt": "如何上传文件？"},
    ]
    _write_agent_config(agent_id, name="助手", description="", common_questions=questions)

    config_file = paths_module.get_paths().agent_dir(agent_id) / "config.yaml"
    raw = yaml.safe_load(config_file.read_text(encoding="utf-8"))
    assert raw["common_questions"] == questions

    extras = _load_config_extras(agent_id)
    assert [q.model_dump() for q in extras["common_questions"]] == questions


def test_write_agent_config_coerces_legacy_string_questions(isolated_paths) -> None:
    """Legacy ``list[str]`` configs coerce to {title, prompt} with title=content."""
    agent_id = "agent-cq-legacy"
    _write_agent_config(
        agent_id, name="助手", description="", common_questions=["如何上传文件？"]
    )

    config_file = paths_module.get_paths().agent_dir(agent_id) / "config.yaml"
    raw = yaml.safe_load(config_file.read_text(encoding="utf-8"))
    assert raw["common_questions"] == [
        {"title": "如何上传文件？", "prompt": "如何上传文件？"}
    ]


def test_write_agent_config_omits_common_questions_when_none(isolated_paths) -> None:
    agent_id = "agent-cq-2"
    _write_agent_config(agent_id, name="x", description="")

    config_file = paths_module.get_paths().agent_dir(agent_id) / "config.yaml"
    raw = yaml.safe_load(config_file.read_text(encoding="utf-8"))
    assert "common_questions" not in raw
    assert _load_config_extras(agent_id)["common_questions"] is None


def test_write_agent_config_explicit_empty_common_questions_is_persisted(isolated_paths) -> None:
    agent_id = "agent-cq-3"
    _write_agent_config(agent_id, name="x", description="", common_questions=[])

    assert _load_config_extras(agent_id)["common_questions"] == []


def test_write_agent_config_explicit_empty_skills_is_persisted(isolated_paths) -> None:
    agent_id = "agent-003"
    _write_agent_config(agent_id, name="x", description="", skills=[])

    extras = _load_config_extras(agent_id)
    assert extras["skills"] == []


def test_write_agent_config_omits_skills_when_none(isolated_paths) -> None:
    agent_id = "agent-004"
    _write_agent_config(agent_id, name="x", description="")

    config_file = paths_module.get_paths().agent_dir(agent_id) / "config.yaml"
    raw = yaml.safe_load(config_file.read_text(encoding="utf-8"))

    assert "skills" not in raw
    assert "model" not in raw
    assert "tool_groups" not in raw


@pytest.mark.asyncio
async def test_backfill_listing_cache_mirrors_config_values() -> None:
    store = MemoryAgentStore()
    cfg = AgentConfig(id="a1", name="Agent", skills=["s1", "s2"], tool_groups=["g1"], model="m1")
    await _backfill_listing_cache(store, [("a1", cfg)])

    extras = await store.get_extras_for(["a1"])
    assert extras["a1"]["skills"] == ["s1", "s2"]
    assert extras["a1"]["tool_groups"] == ["g1"]
    assert extras["a1"]["model"] == "m1"


@pytest.mark.asyncio
async def test_backfill_listing_cache_is_idempotent_and_skips_unchanged(monkeypatch) -> None:
    store = MemoryAgentStore()
    cfg = AgentConfig(id="a1", name="Agent", skills=["s1"], tool_groups=None, model=None)
    await _backfill_listing_cache(store, [("a1", cfg)])

    # On a second run with identical config, no re-write should occur.
    calls = {"replace": 0, "set": 0}
    orig_replace = store.replace_skills
    orig_set = store.set_extras

    async def _replace(agent_id, skills):
        calls["replace"] += 1
        await orig_replace(agent_id, skills)

    async def _set(agent_id, *, tool_groups, model):
        calls["set"] += 1
        await orig_set(agent_id, tool_groups=tool_groups, model=model)

    monkeypatch.setattr(store, "replace_skills", _replace)
    monkeypatch.setattr(store, "set_extras", _set)

    await _backfill_listing_cache(store, [("a1", cfg)])
    assert calls == {"replace": 0, "set": 0}

    # Changing the config triggers a re-write.
    cfg2 = AgentConfig(id="a1", name="Agent", skills=["s1", "s2"], tool_groups=None, model=None)
    await _backfill_listing_cache(store, [("a1", cfg2)])
    assert calls["replace"] == 1