deerflow-code/offline-backend-20260512/backend/tests/test_agents_backend.py

from __future__ import annotations

import asyncio
import json
from types import SimpleNamespace
from uuid import uuid4

import pytest
import pytest_asyncio
import yaml
from sqlalchemy import select
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

from app.gateway.routers.agents import (
    AgentsListResponse,
    _agent_list_cache,
    _agent_list_inflight,
    _agent_owner_name_cache,
    _backfill_listing_cache,
    _cached_paginated_agent_list,
    _load_config_extras,
    _resolve_usernames,
    _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, normalize_agent_chat_link
from deerflow.persistence.agents import AgentRepository, MemoryAgentStore
from deerflow.persistence.agents.model import AgentFavoriteRow, AgentRow
from deerflow.persistence.base import Base
from deerflow.persistence.positions.memory import MemoryPositionStore
from deerflow.persistence.roundtable_chains.model import RoundtableChainRow
from deerflow.persistence.tags.memory import MemoryTagStore
from deerflow.services.position_sync import PositionSyncService


@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_asyncio.fixture
async def sql_session_factory(tmp_path):
    engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'agent-tests.db'}")
    async with engine.begin() as conn:
        # Importing PositionSyncService above registers the agent / skill /
        # scheduled-task rows it materializes into.
        await conn.run_sync(Base.metadata.create_all)
    sf = async_sessionmaker(engine, expire_on_commit=False)
    try:
        yield sf
    finally:
        await engine.dispose()


async def _add_sql_agent(sf, *, agent_id: str, user_id: str | None) -> None:
    async with sf() as session:
        session.add(
            AgentRow(
                id=agent_id,
                name=agent_id,
                description="",
                user_id=user_id,
                published=True,
                is_private=False,
                square_id="",
            )
        )
        await session.commit()


@pytest.mark.asyncio
async def test_sidebar_agent_query_returns_only_lightweight_visible_fields(sql_session_factory):
    await _add_sql_agent(sql_session_factory, agent_id="owned-agent", user_id="user-1")
    await _add_sql_agent(sql_session_factory, agent_id="public-agent", user_id=None)

    agents = await AgentRepository(sql_session_factory).list_sidebar("user-1", limit=3)

    assert agents == [
        {"id": "owned-agent", "name": "owned-agent"},
        {"id": "public-agent", "name": "public-agent"},
    ]
    assert all(set(item) == {"id", "name"} for item in agents)


@pytest.mark.asyncio
async def test_memory_sidebar_query_returns_only_lightweight_fields():
    store = MemoryAgentStore()
    await store.create({"id": "owned-agent", "name": "我的智能体", "user_id": "user-1"})

    agents = await store.list_sidebar("user-1", limit=3)

    assert agents == [{"id": "owned-agent", "name": "我的智能体"}]


async def _add_sql_favorite(sf, *, user_id: str, agent_id: str, origin: str) -> None:
    async with sf() as session:
        session.add(
            AgentFavoriteRow(
                id=uuid4().hex,
                user_id=user_id,
                agent_id=agent_id,
                origin=origin,
                position_id="default" if origin == "position" else None,
            )
        )
        await session.commit()


async def _add_sql_chain(sf, *, chain_id: str, agent_ids: list[str]) -> None:
    async with sf() as session:
        session.add(
            RoundtableChainRow(
                id=chain_id,
                user_id="chain-owner",
                title=chain_id,
                description="",
                seats=json.dumps(
                    [{"agent_id": agent_id, "name": agent_id, "description": ""} for agent_id in agent_ids],
                    ensure_ascii=False,
                ),
            )
        )
        await session.commit()


async def _sql_favorite_rows(sf, user_id: str) -> set[tuple[str, str]]:
    async with sf() as session:
        rows = (
            await session.execute(
                select(AgentFavoriteRow.agent_id, AgentFavoriteRow.origin).where(
                    AgentFavoriteRow.user_id == user_id
                )
            )
        ).all()
    return {(agent_id, origin) for agent_id, origin in rows}


async def _default_positions() -> MemoryPositionStore:
    positions = MemoryPositionStore()
    await positions.create_position(
        {
            "id": "default",
            "name": "Default",
            "description": "",
            "is_default": True,
        }
    )
    return positions


@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_position_sync_default_does_not_collect_builtin_agents(sql_session_factory) -> None:
    await _add_sql_agent(sql_session_factory, agent_id="builtin-auto", user_id=None)
    await _add_sql_agent(sql_session_factory, agent_id="builtin-manual", user_id=None)
    await _add_sql_favorite(
        sql_session_factory,
        user_id="user-a",
        agent_id="builtin-auto",
        origin="position",
    )
    await _add_sql_favorite(
        sql_session_factory,
        user_id="user-a",
        agent_id="builtin-manual",
        origin="user",
    )

    sync = PositionSyncService(
        sql_session_factory,
        await _default_positions(),
        MemoryTagStore(),
    )
    await sync.sync_user("user-a")

    assert await _sql_favorite_rows(sql_session_factory, "user-a") == {
        ("builtin-manual", "user")
    }


@pytest.mark.asyncio
async def test_position_sync_tagged_agents_are_not_auto_favorited(sql_session_factory) -> None:
    await _add_sql_agent(sql_session_factory, agent_id="builtin-tagged", user_id=None)
    await _add_sql_agent(sql_session_factory, agent_id="custom-tagged", user_id="owner")
    await _add_sql_agent(sql_session_factory, agent_id="chain-tagged", user_id="owner")
    await _add_sql_chain(sql_session_factory, chain_id="chain-1", agent_ids=["chain-tagged"])

    positions = await _default_positions()
    tags = MemoryTagStore()
    await tags.create_tag({"id": "agent-tag", "name": "Agent tag", "tag_type": "agent"})
    await tags.set_target_tags("agent", "builtin-tagged", ["agent-tag"])
    await tags.set_target_tags("agent", "custom-tagged", ["agent-tag"])
    await tags.set_target_tags("agent", "chain-tagged", ["agent-tag"])
    await positions.set_tag_bindings("default", {"agent": ["agent-tag"]})

    sync = PositionSyncService(sql_session_factory, positions, tags)
    await sync.sync_user("user-a")

    assert await _sql_favorite_rows(sql_session_factory, "user-a") == {
        ("custom-tagged", "position")
    }


@pytest.mark.asyncio
async def test_mine_scope_ignores_auto_agent_favorites_but_keeps_manual_builtin(sql_session_factory) -> None:
    await _add_sql_agent(sql_session_factory, agent_id="builtin-auto", user_id=None)
    await _add_sql_agent(sql_session_factory, agent_id="builtin-manual", user_id=None)
    await _add_sql_agent(sql_session_factory, agent_id="custom-auto", user_id="owner")
    await _add_sql_agent(sql_session_factory, agent_id="chain-auto", user_id="owner")
    await _add_sql_chain(sql_session_factory, chain_id="chain-1", agent_ids=["chain-auto"])
    await _add_sql_favorite(
        sql_session_factory,
        user_id="user-a",
        agent_id="builtin-auto",
        origin="position",
    )
    await _add_sql_favorite(
        sql_session_factory,
        user_id="user-a",
        agent_id="builtin-manual",
        origin="user",
    )
    await _add_sql_favorite(
        sql_session_factory,
        user_id="user-a",
        agent_id="custom-auto",
        origin="position",
    )
    await _add_sql_favorite(
        sql_session_factory,
        user_id="user-a",
        agent_id="chain-auto",
        origin="position",
    )

    store = AgentRepository(sql_session_factory)
    rows, total = await store.list_paginated(
        user_id="user-a",
        scope="mine",
        search=None,
        tag_ids=None,
        page=1,
        page_size=20,
    )

    assert total == 2
    assert {row["id"] for row in rows} == {"builtin-manual", "custom-auto"}
    assert await store.list_favorite_ids("user-a") == {"builtin-manual", "custom-auto"}


@pytest.mark.asyncio
async def test_manual_favorite_upgrades_existing_auto_builtin_favorite(sql_session_factory) -> None:
    await _add_sql_agent(sql_session_factory, agent_id="builtin-agent", user_id=None)
    await _add_sql_favorite(
        sql_session_factory,
        user_id="user-a",
        agent_id="builtin-agent",
        origin="position",
    )

    store = AgentRepository(sql_session_factory)
    assert await store.list_favorite_ids("user-a") == set()

    await store.add_favorite("user-a", "builtin-agent")

    assert await _sql_favorite_rows(sql_session_factory, "user-a") == {
        ("builtin-agent", "user")
    }
    assert await store.list_favorite_ids("user-a") == {"builtin-agent"}
    rows, total = await store.list_paginated(
        user_id="user-a",
        scope="mine",
        search=None,
        tag_ids=None,
        page=1,
        page_size=20,
    )
    assert total == 1
    assert [row["id"] for row in rows] == ["builtin-agent"]


async def _add_sql_hidden_agent(sf, *, agent_id: str, user_id: str) -> None:
    """他人创建、普通列表不可见的条目（未发布 / 私密广场）。"""
    async with sf() as session:
        session.add(
            AgentRow(
                id=agent_id,
                name=agent_id,
                description="",
                user_id=user_id,
                published=not agent_id.startswith("hidden-unpublished"),
                is_private=agent_id.startswith("hidden-private"),
                square_id="",
            )
        )
        await session.commit()


@pytest.mark.asyncio
async def test_sql_list_paginated_admin_pages_every_agent(sql_session_factory) -> None:
    """管理端分页（GET /api/agents?admin=true&page=N 的 store 支撑）：
    admin=True 跳过可见性过滤 — 内置 / 他人未发布 / 私密广场条目全部参与分页，
    搜索在服务端过滤，越界页返回空页而非报错。"""
    await _add_sql_agent(sql_session_factory, agent_id="builtin-1", user_id=None)
    await _add_sql_agent(sql_session_factory, agent_id="owned-a", user_id="user-a")
    await _add_sql_hidden_agent(sql_session_factory, agent_id="hidden-unpublished", user_id="user-b")
    await _add_sql_hidden_agent(sql_session_factory, agent_id="hidden-private", user_id="user-b")

    store = AgentRepository(sql_session_factory)

    # 非 admin 对照：user-a 只看得到自己 + 内置（user-b 的两条不可见）。
    _, visible_total = await store.list_paginated(
        user_id="user-a",
        scope=None,
        search=None,
        tag_ids=None,
        page=1,
        page_size=20,
    )
    assert visible_total == 2

    page1, total = await store.list_paginated(
        user_id="user-a",
        scope=None,
        search=None,
        tag_ids=None,
        page=1,
        page_size=2,
        admin=True,
        sort="time_asc",
    )
    page2, total2 = await store.list_paginated(
        user_id="user-a",
        scope=None,
        search=None,
        tag_ids=None,
        page=2,
        page_size=2,
        admin=True,
        sort="time_asc",
    )
    assert total == 4 and total2 == 4
    assert len(page1) == 2 and len(page2) == 2
    assert {r["id"] for r in page1} | {r["id"] for r in page2} == {
        "builtin-1",
        "owned-a",
        "hidden-unpublished",
        "hidden-private",
    }

    # 越界页：空页但 total 不变（前端据此收敛回末页）。
    page3, total3 = await store.list_paginated(
        user_id="user-a",
        scope=None,
        search=None,
        tag_ids=None,
        page=3,
        page_size=2,
        admin=True,
    )
    assert total3 == 4 and page3 == []

    # 服务端搜索：管理端目录同样按名称过滤。
    hits, hits_total = await store.list_paginated(
        user_id="user-a",
        scope=None,
        search="hidden",
        tag_ids=None,
        page=1,
        page_size=20,
        admin=True,
    )
    assert hits_total == 2
    assert {r["id"] for r in hits} == {"hidden-unpublished", "hidden-private"}


@pytest.mark.asyncio
async def test_memory_store_admin_pagination() -> None:
    """内存版 store 的 admin 分页行为与 SQL 版一致（单元测试/回退路径）。"""
    store = MemoryAgentStore()
    await store.ensure_builtin({"id": "builtin-1", "name": "Built In"})
    await store.create({"id": "owned-a", "name": "Owned A", "user_id": "user-a"})
    await store.create(
        {"id": "hidden-b", "name": "Hidden B", "user_id": "user-b", "published": False}
    )
    await store.create(
        {
            "id": "private-b",
            "name": "Private B",
            "user_id": "user-b",
            # 内存版 _is_visible 不看 is_private，这里保持未发布，
            # 使「非 admin 对照 = 内置 + 自己」在两种 store 语义下都成立。
            "published": False,
            "is_private": True,
        }
    )

    _, visible_total = await store.list_paginated(
        user_id="user-a", scope=None, search=None, tag_ids=None, page=1, page_size=20
    )
    assert visible_total == 2

    page1, total = await store.list_paginated(
        user_id="user-a",
        scope=None,
        search=None,
        tag_ids=None,
        page=1,
        page_size=3,
        admin=True,
        sort="time_asc",
    )
    page2, total2 = await store.list_paginated(
        user_id="user-a",
        scope=None,
        search=None,
        tag_ids=None,
        page=2,
        page_size=3,
        admin=True,
        sort="time_asc",
    )
    assert total == 4 and total2 == 4
    assert len(page1) == 3 and len(page2) == 1
    assert {r["id"] for r in page1} | {r["id"] for r in page2} == {
        "builtin-1",
        "owned-a",
        "hidden-b",
        "private-b",
    }


async def _add_sql_hidden_flagged_agent(sf, *, agent_id: str, user_id: str | None) -> None:
    """创建时即 hidden=True 的条目（绕过 API，直接落库）。"""
    async with sf() as session:
        session.add(
            AgentRow(
                id=agent_id,
                name=agent_id,
                description="",
                user_id=user_id,
                published=True,
                is_private=False,
                square_id="",
                hidden=True,
            )
        )
        await session.commit()


@pytest.mark.asyncio
async def test_sql_hidden_agent_vanishes_from_non_admin_listings_but_stays_usable(
    sql_session_factory,
) -> None:
    """管理端隐藏开关的 store 语义：hidden=True 后从所有非 admin 列表消失
    （list_visible / 各分页 scope / selector / 私密广场），但单条访问
    （get_visible）不受影响，admin 列表仍可见且可按 hidden 筛选。"""
    await _add_sql_agent(sql_session_factory, agent_id="builtin-1", user_id=None)
    await _add_sql_agent(sql_session_factory, agent_id="owned-a", user_id="user-a")
    await _add_sql_agent(sql_session_factory, agent_id="published-b", user_id="user-b")
    await _add_sql_hidden_flagged_agent(sql_session_factory, agent_id="flagged-1", user_id="user-b")

    store = AgentRepository(sql_session_factory)

    # list_visible（默认列表）：flagged-1 消失
    visible_ids = {r["id"] for r in await store.list_visible("user-a")}
    assert visible_ids == {"builtin-1", "owned-a", "published-b"}

    # 非_admin 分页（默认 scope 与 square/mine scope 一致排除）
    for scope in (None, "square", "mine"):
        rows, total = await store.list_paginated(
            user_id="user-a", scope=scope, search=None, tag_ids=None, page=1, page_size=20,
        )
        assert "flagged-1" not in {r["id"] for r in rows}, scope
    # 属主自己也看不到自己被隐藏的智能体（非 admin 一律过滤）
    rows, _ = await store.list_paginated(
        user_id="user-b", scope="mine", search=None, tag_ids=None, page=1, page_size=20,
    )
    assert "flagged-1" not in {r["id"] for r in rows}

    # selector（featured）与私密广场同样排除
    await store.set_featured("flagged-1", 1)
    featured_ids = {r["id"] for r in await store.list_featured("user-a")}
    assert "flagged-1" not in featured_ids
    assert await store.update_any("flagged-1", {"is_private": True}) is not None
    private_ids = {r["id"] for r in await store.list_private()}
    assert "flagged-1" not in private_ids

    # 单条访问不受影响（隐藏不是禁用：既有会话/详情继续可用）
    assert await store.get_visible("flagged-1", "user-a") is not None
    assert await store.get_visible("flagged-1", "user-b") is not None

    # admin：全量可见，且可按 hidden 筛选
    rows, total = await store.list_paginated(
        user_id="user-a", scope=None, search=None, tag_ids=None,
        page=1, page_size=20, admin=True,
    )
    assert total == 4 and "flagged-1" in {r["id"] for r in rows}
    rows, total = await store.list_paginated(
        user_id="user-a", scope=None, search=None, tag_ids=None,
        page=1, page_size=20, admin=True, hidden=True,
    )
    assert total == 1 and {r["id"] for r in rows} == {"flagged-1"}
    rows, total = await store.list_paginated(
        user_id="user-a", scope=None, search=None, tag_ids=None,
        page=1, page_size=20, admin=True, hidden=False,
    )
    assert total == 3 and "flagged-1" not in {r["id"] for r in rows}

    # 开关往返：取消隐藏后回到列表。注：flagged-1 中途被挪进了私密广场
    #（is_private=True），普通列表本就不含私密条目，先移回公共广场再断言。
    assert (await store.update_any("flagged-1", {"hidden": False}))["hidden"] is False
    await store.update_any("flagged-1", {"is_private": False})
    visible_ids = {r["id"] for r in await store.list_visible("user-a")}
    assert "flagged-1" in visible_ids


@pytest.mark.asyncio
async def test_memory_hidden_agent_filtered_from_non_admin_listings() -> None:
    """内存版 store 的 hidden 行为与 SQL 版一致。"""
    store = MemoryAgentStore()
    await store.ensure_builtin({"id": "builtin-1", "name": "Built In"})
    await store.create({"id": "owned-a", "name": "Owned A", "user_id": "user-a"})
    await store.create(
        {"id": "flagged-1", "name": "Flagged", "user_id": "user-b", "published": True, "hidden": True}
    )

    assert {r["id"] for r in await store.list_visible("user-a")} == {"builtin-1", "owned-a"}
    rows, total = await store.list_paginated(
        user_id="user-a", scope=None, search=None, tag_ids=None, page=1, page_size=20,
    )
    assert total == 2 and "flagged-1" not in {r["id"] for r in rows}
    # 单条访问 + admin 可见 + hidden 筛选
    assert await store.get_visible("flagged-1", "user-a") is not None
    rows, total = await store.list_paginated(
        user_id="user-a", scope=None, search=None, tag_ids=None,
        page=1, page_size=20, admin=True,
    )
    assert total == 3 and "flagged-1" in {r["id"] for r in rows}
    rows, total = await store.list_paginated(
        user_id="user-a", scope=None, search=None, tag_ids=None,
        page=1, page_size=20, admin=True, hidden=True,
    )
    assert total == 1 and {r["id"] for r in rows} == {"flagged-1"}
    # 开关往返
    assert (await store.update_any("flagged-1", {"hidden": False}))["hidden"] is False
    assert "flagged-1" in {r["id"] for r in await store.list_visible("user-a")}


@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["model_prompt_suffix_enabled"] is True
    assert raw["tool_groups"] == ["tg-a"]


def test_write_agent_config_persists_prompt_suffix_switch(isolated_paths) -> None:
    _write_agent_config(
        "agent-suffix-off",
        name="助手",
        description="",
        model_prompt_suffix_enabled=False,
    )

    assert _load_config_extras("agent-suffix-off")["model_prompt_suffix_enabled"] is False


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,
        "model_prompt_suffix_enabled": True,
        "tool_groups": None,
        "common_questions": None,
        "template_name": None,
        "seat_skill_directive": None,
        "llmwiki_knowledge_base_ids": None,
        "chat_link": 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,
        "model_prompt_suffix_enabled": True,
        "tool_groups": None,
        "common_questions": None,
        "template_name": None,
        "seat_skill_directive": None,
        "llmwiki_knowledge_base_ids": None,
        "chat_link": 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
    assert "chat_link" not in raw


def test_write_agent_config_persists_chat_link_and_drops_blank(isolated_paths) -> None:
    agent_id = "agent-chat-link"
    link = {
        "url": "https://example.test/desk",
        "open_mode": "self",
        "route_params": [
            {"key": "userName", "paramType": "login", "source": "userName"},
            {"key": "theme", "paramType": "theme", "themeValue": "red=red;dark-blue=dark"},
        ],
    }
    _write_agent_config(agent_id, name="助手", description="", chat_link=link)

    extras = _load_config_extras(agent_id)
    assert extras["chat_link"]["url"] == "https://example.test/desk"
    assert extras["chat_link"]["open_mode"] == "self"
    assert extras["chat_link"]["route_params"][0]["key"] == "userName"
    assert extras["chat_link"]["route_params"][1]["paramType"] == "theme"

    _write_agent_config(agent_id, name="助手", description="", chat_link=None)
    raw = yaml.safe_load((paths_module.get_paths().agent_dir(agent_id) / "config.yaml").read_text(encoding="utf-8"))
    assert "chat_link" not in raw
    assert _load_config_extras(agent_id)["chat_link"] is None


def test_normalize_agent_chat_link_rejects_unsafe_urls() -> None:
    assert normalize_agent_chat_link({"url": ""}) is None
    assert normalize_agent_chat_link({"url": "javascript:alert(1)"}, strict=False) is None
    with pytest.raises(ValueError):
        normalize_agent_chat_link({"url": "javascript:alert(1)"}, strict=True)
    kept = normalize_agent_chat_link({"url": "/page/workspace/agents", "open_mode": "blank"})
    assert kept is not None
    assert kept["url"] == "/page/workspace/agents"
    loaded = AgentConfig(name="x", chat_link={"url": "javascript:alert(1)"})
    assert loaded.chat_link is None


@pytest.mark.asyncio
async def test_backfill_listing_cache_mirrors_chat_link() -> None:
    store = MemoryAgentStore()
    link = {"url": "https://example.test/a", "open_mode": "blank", "route_params": []}
    cfg = AgentConfig(id="a1", name="Agent", chat_link=link)
    await _backfill_listing_cache(store, [("a1", cfg)])
    extras = await store.get_extras_for(["a1"])
    assert extras["a1"]["chat_link"]["url"] == "https://example.test/a"

    await _backfill_listing_cache(store, [("a1", cfg)])
    assert (await store.get_extras_for(["a1"]))["a1"]["chat_link"]["url"] == "https://example.test/a"


@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


@pytest.mark.asyncio
async def test_paginated_agent_list_coalesces_concurrent_reads_and_returns_copies() -> None:
    _agent_list_cache.clear()
    _agent_list_inflight.clear()
    calls = 0

    async def loader() -> AgentsListResponse:
        nonlocal calls
        calls += 1
        await asyncio.sleep(0.01)
        return AgentsListResponse(agents=[], total=3, page=1, page_size=20)

    first, second = await asyncio.gather(
        _cached_paginated_agent_list(("user", 1), loader),
        _cached_paginated_agent_list(("user", 1), loader),
    )
    assert calls == 1
    assert {first[1], second[1]} == {"未命中", "合并等待"}

    first[0].total = 99
    cached, cache_status = await _cached_paginated_agent_list(("user", 1), loader)
    assert cache_status == "命中"
    assert cached.total == 3
    assert calls == 1


@pytest.mark.asyncio
async def test_agent_owner_usernames_are_batched_and_cached(monkeypatch) -> None:
    from app.gateway import deps as deps_module

    _agent_owner_name_cache.clear()

    class Provider:
        def __init__(self):
            self.calls = []

        async def get_users_by_ids(self, user_ids):
            self.calls.append(list(user_ids))
            return [SimpleNamespace(id="owner-1", email="alice@example.com")]

    provider = Provider()
    monkeypatch.setattr(deps_module, "get_local_provider", lambda: provider)

    assert await _resolve_usernames({"owner-1", "deleted-user"}) == {"owner-1": "alice"}
    assert await _resolve_usernames({"owner-1", "deleted-user"}) == {"owner-1": "alice"}
    assert provider.calls == [["deleted-user", "owner-1"]]