deerflow-code/offline-backend-20260512/backend/app/gateway/routers/agents.py

"""CRUD API for custom agents."""

from __future__ import annotations

import asyncio
import json
import logging
import os
import shutil
import tempfile
import time
from datetime import datetime
from typing import Any
from uuid import uuid4

import yaml
from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response
from pydantic import BaseModel, Field

from app.gateway.api_read_cache import (
    get_cache_version,
    get_cached_json_payload,
    release_cache_fill,
    set_cached_json,
    wait_for_cached_json_payload_or_claim_fill,
)
from app.gateway.auth.password import verify_password_async
from app.gateway.deps import get_agent_store, get_config, get_skill_store, get_tag_store
from app.gateway.perf_logging import finish_trace, trace_request, trace_stage
from app.gateway.routers._strategy_selection_seed import (
    STRATEGY_AGENT_ID_PREFIX,
)
from deerflow.agents.lead_agent.prompt import refresh_skills_system_prompt_cache_async
from deerflow.config.agents_api_config import get_agents_api_config
from deerflow.config.agents_config import AgentChatLinkConfig, CommonQuestion, list_custom_agents, load_agent_config, load_agent_soul, normalize_agent_chat_link, validate_agent_id
from deerflow.config.app_config import AppConfig
from deerflow.config.paths import get_paths
from deerflow.config.system_settings import load_system_settings, normalize_square_id, resolve_default_square_id
from deerflow.persistence.agents import AgentStore
from deerflow.persistence.skills import SkillStore
from deerflow.persistence.tags import TagStore
from deerflow.runtime.user_context import get_effective_user_id
from deerflow.skills.cascade import sync_agent_skills_to_available
from deerflow.skills.storage import get_or_new_skill_storage
from deerflow.skills.types import SkillCategory
from deerflow.skills.usage import all_entries as get_skill_usage_entries

logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api", tags=["agents"])


def _is_strategy_agent_id(agent_id: str) -> bool:
    return agent_id.casefold().startswith(STRATEGY_AGENT_ID_PREFIX.casefold())


def _require_canonical_record_id(requested_id: str, record: dict[str, Any]) -> str:
    """Reject case-insensitive database aliases before touching agent files.

    MySQL deployments may use a case-insensitive collation while Windows agent
    directories are also case-insensitive.  A lookup for a differently-cased
    id must never be allowed to select one row and mutate another path.
    """

    canonical_id = str(record.get("id") or "")
    if canonical_id != requested_id:
        raise HTTPException(
            status_code=409,
            detail=f"Agent id casing mismatch; use canonical id '{canonical_id}'",
        )
    return canonical_id


class AgentResponse(BaseModel):
    """Response model for an agent."""

    id: str = Field(..., description="Agent id used for routing and filesystem storage")
    name: str = Field(..., description="Display name. Chinese and duplicate names are allowed.")
    description: str = Field(default="", description="Agent description")
    user_id: str | None = Field(default=None, description="Owner user id. Null means built-in.")
    owner_username: str | None = Field(default=None, description="Publisher username (email local part). Null for built-in agents.")
    published: bool = Field(default=False, description="Whether a user-created agent is public")
    is_private: bool = Field(default=False, description="Whether the agent lives in the password-gated private square instead of the public one")
    hidden: bool = Field(
        default=False,
        description=(
            "Admin-set catalog switch: when True the agent disappears from every non-admin listing "
            "(gallery / 我的 / 广场 / selector / private square) but stays fully usable — only the "
            "admin agents console lists it"
        ),
    )
    square_id: str = Field(default="", description="Publish square (发布广场) the agent is filed under. Empty string = the default square. Independent of is_private.")
    builtin: bool = Field(default=False, description="True when user_id is null")
    created_at: datetime | str | None = None
    updated_at: datetime | str | None = None
    soul: str | None = Field(default=None, description="SOUL.md content")

    # Deprecated compatibility fields. Agent runtime configuration is no longer
    # stored in the agents table, but older frontends may still read these keys.
    model: str | None = None
    model_prompt_suffix_enabled: bool = Field(
        default=True,
        description="Whether this agent uses the administrator-configured prompt suffix for its selected model.",
    )
    tool_groups: list[str] | None = None
    skills: list[str] | None = None
    skill_labels: dict[str, str] = Field(
        default_factory=dict,
        description="Chinese display names for this agent's skills, keyed by skill name. List responses fill this from the skills table so gallery cards do not scan the skill catalog.",
    )

    common_questions: list[CommonQuestion] | None = Field(
        default=None,
        description="Recommended questions shown on the agent chat home / sidebar (常用问题). Each item is {title, prompt}; legacy string lists are coerced.",
    )

    template_name: str | None = Field(
        default=None,
        description="入库模板名 (external knowledge-ingest templateName). When set, markdown artifacts produced in this agent's chats can be published to the external knowledge-ingest API.",
    )

    seat_skill_directive: bool | None = Field(
        default=None,
        description="会商席位「技能识别强化」开关 (config.yaml). True = 本智能体作为会商席位时强化技能识别 (与业务链条同名开关取 OR). None/False = 关 (默认).",
    )

    llmwiki_knowledge_base_ids: list[str] | None = Field(
        default=None,
        description="DeerFlow LLMWiki mapping ids this agent may offer for conversation retrieval.",
    )

    chat_link: AgentChatLinkConfig | None = Field(
        default=None,
        description="Optional destination for the gallery 「对话」 button. Null keeps the in-app chat. open_mode is blank (new window) or self (current window). route_params match light-app route params.",
    )

    tags: list[dict] = Field(default_factory=list, description="Tags assigned to this agent")

    featured_order: int | None = Field(
        default=None,
        description="Admin-curated homepage rank. Null = not featured. Lower sorts first.",
    )

    is_favorite: bool = Field(
        default=False,
        description="Whether the requesting user has favorited (收藏) this agent.",
    )

    is_pinned: bool = Field(
        default=False,
        description="Whether the requesting user has pinned this agent.",
    )


class AgentsListResponse(BaseModel):
    """Response model for listing agents.

    ``total`` / ``page`` / ``page_size`` describe server-side pagination. When
    a caller does not opt into pagination they default to "everything on one
    page" (``total == len(agents)``, ``page == 1``), so existing unpaginated
    callers keep working unchanged.
    """

    agents: list[AgentResponse]
    total: int = Field(default=0, description="Total agents matching the query across all pages")
    page: int = Field(default=1, description="1-based page index of this response")
    page_size: int = Field(default=0, description="Page size; 0 means the result was not paginated")


_AGENT_LIST_CACHE_SECONDS = 5.0
_MAX_AGENT_LIST_CACHE_ENTRIES = 256
_agent_list_cache: dict[tuple[Any, ...], tuple[float, AgentsListResponse]] = {}
_agent_list_inflight: dict[tuple[Any, ...], asyncio.Task[AgentsListResponse]] = {}
_AGENT_OWNER_NAME_CACHE_SECONDS = 60.0
_MAX_AGENT_OWNER_NAME_CACHE_ENTRIES = 4096
_agent_owner_name_cache: dict[str, tuple[float, str | None]] = {}


async def _cached_paginated_agent_list(
    key: tuple[Any, ...],
    loader,
) -> tuple[AgentsListResponse, str]:
    """Cache paginated responses briefly and coalesce identical DB reads."""
    now = time.monotonic()
    cached = _agent_list_cache.get(key)
    if cached is not None and cached[0] > now:
        return cached[1].model_copy(deep=True), "命中"
    if cached is not None:
        _agent_list_cache.pop(key, None)

    task = _agent_list_inflight.get(key)
    if task is not None:
        result = await asyncio.shield(task)
        return result.model_copy(deep=True), "合并等待"

    async def _load_and_cache() -> AgentsListResponse:
        try:
            result = await loader()
            cache_now = time.monotonic()
            if len(_agent_list_cache) >= _MAX_AGENT_LIST_CACHE_ENTRIES:
                expired = [cache_key for cache_key, value in _agent_list_cache.items() if value[0] <= cache_now]
                for cache_key in expired:
                    _agent_list_cache.pop(cache_key, None)
                if len(_agent_list_cache) >= _MAX_AGENT_LIST_CACHE_ENTRIES:
                    oldest = sorted(_agent_list_cache, key=lambda cache_key: _agent_list_cache[cache_key][0])
                    for cache_key in oldest[: max(1, len(oldest) // 2)]:
                        _agent_list_cache.pop(cache_key, None)
            _agent_list_cache[key] = (cache_now + _AGENT_LIST_CACHE_SECONDS, result.model_copy(deep=True))
            return result
        finally:
            _agent_list_inflight.pop(key, None)

    task = asyncio.create_task(_load_and_cache())
    _agent_list_inflight[key] = task
    result = await asyncio.shield(task)
    return result.model_copy(deep=True), "未命中"


class AgentCreateRequest(BaseModel):
    """Request body for creating a custom agent."""

    id: str | None = Field(
        default=None,
        description="Optional explicit storage id. Must match the storage id pattern. If omitted, a UUID is generated.",
    )
    name: str = Field(..., description="Display name; Chinese and duplicate names are allowed")
    description: str = Field(default="", description="Agent description")
    soul: str = Field(default="", description="SOUL.md content - agent personality and behavioral guardrails")
    published: bool = Field(default=False, description="Whether the new agent should be public")
    is_private: bool = Field(default=False, description="When True, the new agent goes into the password-gated private square instead of the public one")
    square_id: str | None = Field(default=None, description="Publish square (发布广场) to file the agent under. Omit / empty = the default square.")
    skills: list[str] | None = Field(default=None, description="Skill whitelist persisted to the agent's config.yaml")
    model: str | None = Field(default=None, description="Optional model override persisted to config.yaml")
    model_prompt_suffix_enabled: bool = Field(default=True, description="Whether this agent uses the administrator-configured model prompt suffix.")
    tool_groups: list[str] | None = Field(default=None, description="Optional tool group whitelist persisted to config.yaml")
    common_questions: list[CommonQuestion] | None = Field(default=None, description="Recommended questions (常用问题) persisted to config.yaml. Each item is {title, prompt}; legacy strings are coerced.")
    template_name: str | None = Field(default=None, description="入库模板名 persisted to config.yaml. Enables publishing markdown artifacts to the external knowledge-ingest API.")
    seat_skill_directive: bool | None = Field(default=None, description="会商席位「技能识别强化」开关 persisted to config.yaml. None/False = off (默认).")
    llmwiki_knowledge_base_ids: list[str] | None = Field(
        default=None,
        max_length=100,
        description="Allowed DeerFlow LLMWiki mapping ids persisted to config.yaml.",
    )
    chat_link: AgentChatLinkConfig | None = Field(
        default=None,
        description="Gallery 「对话」 jump target persisted to config.yaml. Null or an empty url keeps the in-app chat.",
    )


class AgentUpdateRequest(BaseModel):
    """Request body for updating a custom agent."""

    name: str | None = Field(default=None, description="Updated display name")
    description: str | None = Field(default=None, description="Updated description")
    soul: str | None = Field(default=None, description="Updated SOUL.md content")
    published: bool | None = Field(default=None, description="Updated publication state")
    is_private: bool | None = Field(default=None, description="Move the agent into / out of the private square (true / false)")
    square_id: str | None = Field(default=None, description="Publish square (发布广场) to file the agent under. Empty string = the default square.")
    skills: list[str] | None = Field(default=None, description="Updated skill whitelist (config.yaml)")
    model: str | None = Field(default=None, description="Updated model override (config.yaml)")
    model_prompt_suffix_enabled: bool | None = Field(default=None, description="Updated per-agent model prompt suffix switch (config.yaml).")
    tool_groups: list[str] | None = Field(default=None, description="Updated tool group whitelist (config.yaml)")
    common_questions: list[CommonQuestion] | None = Field(default=None, description="Updated recommended questions (常用问题, config.yaml). Each item is {title, prompt}; legacy strings are coerced.")
    template_name: str | None = Field(default=None, description="Updated 入库模板名 (config.yaml). Empty string clears it.")
    seat_skill_directive: bool | None = Field(default=None, description="Updated 会商席位「技能识别强化」开关 (config.yaml).")
    llmwiki_knowledge_base_ids: list[str] | None = Field(
        default=None,
        max_length=100,
        description="Updated allowed DeerFlow LLMWiki mapping ids (config.yaml).",
    )
    chat_link: AgentChatLinkConfig | None = Field(
        default=None,
        description="Updated gallery 「对话」 jump target (config.yaml). Null clears it.",
    )


class AgentSelectorResponse(BaseModel):
    """Response for the homepage agent selector dropdown.

    The list is fully curated by admins (``featured_order`` on each row).
    Non-admin users get a read-only view of the same list.
    """

    agents: list[AgentResponse] = Field(
        default_factory=list,
        description="Admin-curated agents, sorted by featured_order ascending.",
    )


class AgentSidebarItem(BaseModel):
    id: str
    name: str


class AgentSidebarResponse(BaseModel):
    agents: list[AgentSidebarItem] = Field(default_factory=list)


class AgentFeaturedRequest(BaseModel):
    """Request body for setting/clearing an agent's homepage featured rank."""

    order: int | None = Field(
        default=None,
        description="Featured rank. Lower sorts first. Pass null to remove the agent from the homepage selector.",
    )


def _require_agents_api_enabled() -> None:
    """Reject access unless the custom-agent management API is explicitly enabled."""
    if not get_agents_api_config().enabled:
        raise HTTPException(
            status_code=403,
            detail=("Custom-agent management API is disabled. Set agents_api.enabled=true to expose agent and user-profile routes over HTTP."),
        )


def _current_user_id(request: Request) -> str:
    user = getattr(request.state, "user", None)
    if user is not None:
        return str(user.id)
    return get_effective_user_id()


def _is_admin_user(request: Request) -> bool:
    """Return True when the request is authenticated as a system admin."""
    user = getattr(request.state, "user", None)
    return getattr(user, "system_role", None) == "admin"


async def _validate_llmwiki_bindings(
    request: Request,
    mapping_ids: list[str] | None,
) -> list[str] | None:
    """Normalize bindings and require read access for the editing actor."""

    if mapping_ids is None:
        return None
    normalized = list(dict.fromkeys(item.strip() for item in mapping_ids if item.strip()))
    if not normalized:
        return []
    store = getattr(request.app.state, "llmwiki_store", None)
    if store is None:
        raise HTTPException(status_code=503, detail="LLMWiki metadata store is unavailable")
    user_id = _current_user_id(request)
    is_admin = _is_admin_user(request)
    for mapping_id in normalized:
        row = await store.get_authorized(
            mapping_id,
            user_id,
            write=False,
            is_admin=is_admin,
        )
        if row is None:
            raise HTTPException(status_code=422, detail=f"LLMWiki knowledge base '{mapping_id}' is unavailable")
    return normalized


def _email_local_part(email: str | None) -> str | None:
    """Return the username portion (before ``@``) of an email address."""
    if not email:
        return None
    at = email.find("@")
    return email[:at] if at > 0 else email


async def _resolve_usernames(user_ids: set[str]) -> dict[str, str]:
    """Batch-resolve owner user ids to friendly usernames (email local part).

    One lookup per distinct id; tolerates missing accounts (deleted users) and
    an uninitialised auth provider by simply omitting those ids from the map.
    """
    user_ids = {uid for uid in user_ids if uid}
    if not user_ids:
        return {}
    try:
        from app.gateway.deps import get_local_provider

        provider = get_local_provider()
    except Exception:
        return {}
    now = time.monotonic()
    names: dict[str, str] = {}
    missing_ids: set[str] = set()
    for user_id in user_ids:
        cached = _agent_owner_name_cache.get(user_id)
        if cached is not None and cached[0] > now:
            if cached[1]:
                names[user_id] = cached[1]
        else:
            missing_ids.add(user_id)

    if missing_ids:
        try:
            users = await provider.get_users_by_ids(sorted(missing_ids))
        except Exception:
            logger.debug("agent owner username batch lookup failed", exc_info=True)
            return names
        loaded = {
            str(user.id): _email_local_part(getattr(user, "email", None))
            for user in users
        }
        expires_at = now + _AGENT_OWNER_NAME_CACHE_SECONDS
        for user_id in missing_ids:
            local = loaded.get(user_id)
            _agent_owner_name_cache[user_id] = (expires_at, local)
            if local:
                names[user_id] = local
        if len(_agent_owner_name_cache) > _MAX_AGENT_OWNER_NAME_CACHE_ENTRIES:
            expired = [key for key, value in _agent_owner_name_cache.items() if value[0] <= now]
            for key in expired:
                _agent_owner_name_cache.pop(key, None)
            if len(_agent_owner_name_cache) > _MAX_AGENT_OWNER_NAME_CACHE_ENTRIES:
                oldest = sorted(_agent_owner_name_cache, key=lambda key: _agent_owner_name_cache[key][0])
                excess = len(_agent_owner_name_cache) - _MAX_AGENT_OWNER_NAME_CACHE_ENTRIES
                for key in oldest[:excess]:
                    _agent_owner_name_cache.pop(key, None)
    return names


def _validate_display_name(name: str) -> str:
    normalized = name.strip()
    if not normalized:
        raise HTTPException(status_code=422, detail="Agent name cannot be empty")
    return normalized


def _validate_storage_id(agent_id: str) -> str:
    try:
        return validate_agent_id(agent_id) or agent_id
    except ValueError as exc:
        raise HTTPException(status_code=422, detail=str(exc)) from exc


def _validate_strategy_agent_model(
    raw_model: str | None,
    *,
    app_config: AppConfig,
) -> str | None:
    """Normalize a shared strategy role's model and reject stale names."""

    model = (raw_model or "").strip() or None
    if model is None:
        return None
    if model not in {entry.name for entry in app_config.models}:
        raise HTTPException(
            status_code=422,
            detail=f"Unknown strategy agent model: {model}",
        )
    return model


async def _validate_strategy_agent_skills(
    requested_skills: list[str] | None,
    *,
    app_config: AppConfig,
    skill_store: SkillStore,
    existing_skills: list[str] | None = None,
) -> list[str] | None:
    """Validate a global strategy role's skill allowlist.

    Strategy-selection agents are built-in roles shared by every project and
    every user.  Their allowlists may therefore reference only canonical skill
    names that exist on disk, are enabled/active, and are globally visible:
    deployment-owned public skills, explicit built-in rows, or custom skills
    currently published by their owner.  An update may retain a previously
    configured skill that was since disabled/archived, so administrators can
    remove it without the rest of the edit being rejected; newly attaching an
    unavailable skill remains forbidden.  In particular, an administrator
    must not be able to attach a different user's unpublished skill merely
    because ``?admin=true`` exposes it in the management catalog.
    """

    if requested_skills is None:
        return None

    normalized: list[str] = []
    for raw_name in requested_skills:
        name = raw_name.strip()
        if not name:
            raise HTTPException(
                status_code=422,
                detail="Strategy agent skill names cannot be blank",
            )
        if name not in normalized:
            normalized.append(name)

    try:
        skills = get_or_new_skill_storage(app_config=app_config).load_skills(enabled_only=False)
        ownership = {str(record.get("name") or ""): record for record in await skill_store.list_all() if record.get("name")}
        usage_entries = get_skill_usage_entries()
    except HTTPException:
        raise
    except Exception as exc:  # noqa: BLE001 - stable API response
        logger.exception("Failed to validate strategy-agent skill allowlist")
        raise HTTPException(status_code=503, detail="Skill catalog is unavailable") from exc

    by_name = {skill.name: skill for skill in skills}
    unknown = [name for name in normalized if name not in by_name]
    if unknown:
        raise HTTPException(
            status_code=422,
            detail=f"Unknown strategy agent skills: {', '.join(unknown)}",
        )

    retained_names = {str(name).strip() for name in (existing_skills or []) if str(name).strip()}
    unavailable = [
        name
        for name in normalized
        if (
            not by_name[name].enabled
            or str((usage_entries.get(name) or {}).get("state") or "active") != "active"
        )
        and name not in retained_names
    ]
    if unavailable:
        raise HTTPException(
            status_code=422,
            detail=f"Disabled or archived strategy agent skills: {', '.join(unavailable)}",
        )

    not_global: list[str] = []
    for name in normalized:
        skill = by_name[name]
        if skill.category != SkillCategory.CUSTOM:
            continue
        record = ownership.get(name)
        if record is not None and (record.get("owner_user_id") is None or bool(record.get("published"))):
            continue
        not_global.append(name)
    if not_global:
        raise HTTPException(
            status_code=422,
            detail=(f"Strategy agent skills must be globally available; unpublished or unregistered custom skills: {', '.join(not_global)}"),
        )

    return normalized


def _ensure_strategy_agent_deletable(agent_id: str) -> None:
    """Keep fixed strategy role ids alive; their SOUL/skills remain editable."""

    if _is_strategy_agent_id(agent_id):
        raise HTTPException(
            status_code=409,
            detail=(f"Strategy role agent '{agent_id}' cannot be deleted; edit its SOUL and skills instead"),
        )


def _chat_link_model(value: Any) -> AgentChatLinkConfig | None:
    normalized = normalize_agent_chat_link(value, strict=False)
    if normalized is None:
        return None
    return AgentChatLinkConfig.model_validate(normalized)


def _require_chat_link(value: Any) -> dict[str, Any] | None:
    raw = value.model_dump() if hasattr(value, "model_dump") else value
    try:
        return normalize_agent_chat_link(raw, strict=True)
    except ValueError as exc:
        raise HTTPException(status_code=422, detail=str(exc)) from exc


def _chat_link_fingerprint(value: Any) -> str | None:
    normalized = normalize_agent_chat_link(value, strict=False)
    if normalized is None:
        return None
    return json.dumps(normalized, sort_keys=True, ensure_ascii=False, separators=(",", ":"))


def _write_agent_config(
    agent_id: str,
    *,
    name: str,
    description: str,
    skills: list[str] | None = None,
    model: str | None = None,
    model_prompt_suffix_enabled: bool = True,
    tool_groups: list[str] | None = None,
    common_questions: list[CommonQuestion] | None = None,
    template_name: str | None = None,
    seat_skill_directive: bool | None = None,
    llmwiki_knowledge_base_ids: list[str] | None = None,
    chat_link: dict[str, Any] | None = None,
) -> None:
    agent_dir = get_paths().agent_dir(agent_id)
    agent_dir.mkdir(parents=True, exist_ok=True)
    config_file = agent_dir / "config.yaml"
    config_data: dict[str, Any] = {
        "id": agent_id,
        "name": name,
        "description": description or "",
    }
    if skills is not None:
        config_data["skills"] = skills
    if model is not None:
        config_data["model"] = model
    config_data["model_prompt_suffix_enabled"] = bool(model_prompt_suffix_enabled)
    if tool_groups is not None:
        config_data["tool_groups"] = tool_groups
    if common_questions is not None:
        # Persist as plain dicts (dropping blank rows) so config.yaml stays a
        # clean mapping rather than a pydantic repr. Accepts already-coerced
        # CommonQuestion models or raw dicts/strings.
        normalized = [q if isinstance(q, CommonQuestion) else CommonQuestion.model_validate(q) for q in common_questions]
        config_data["common_questions"] = [q.model_dump() for q in normalized if q.prompt.strip()]
    if template_name is not None and template_name.strip():
        config_data["template_name"] = template_name.strip()
    # 会商席位技能识别强化开关：仅当显式设置时写入（None = 不写, 由 AgentConfig 默认为关）。
    if seat_skill_directive is not None:
        config_data["seat_skill_directive"] = bool(seat_skill_directive)
    if llmwiki_knowledge_base_ids is not None:
        config_data["llmwiki_knowledge_base_ids"] = list(dict.fromkeys(item.strip() for item in llmwiki_knowledge_base_ids if item.strip()))
    if chat_link is not None:
        config_data["chat_link"] = chat_link
    _atomic_write_text(
        config_file,
        yaml.safe_dump(config_data, default_flow_style=False, allow_unicode=True),
    )


def _load_config_extras(agent_id: str) -> dict[str, Any]:
    """Read persisted agent configuration extras, tolerant of missing files."""
    empty = {
        "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,
    }
    try:
        cfg = load_agent_config(agent_id)
    except (FileNotFoundError, ValueError):
        return dict(empty)
    if cfg is None:
        return dict(empty)
    return {
        "skills": cfg.skills,
        "model": cfg.model,
        "model_prompt_suffix_enabled": cfg.model_prompt_suffix_enabled,
        "tool_groups": cfg.tool_groups,
        "common_questions": cfg.common_questions,
        "template_name": cfg.template_name,
        "seat_skill_directive": cfg.seat_skill_directive,
        "llmwiki_knowledge_base_ids": cfg.llmwiki_knowledge_base_ids,
        "chat_link": cfg.chat_link,
    }


def _write_agent_soul(agent_id: str, soul: str) -> None:
    agent_dir = get_paths().agent_dir(agent_id)
    agent_dir.mkdir(parents=True, exist_ok=True)
    _atomic_write_text(agent_dir / "SOUL.md", soul)


def _atomic_write_text(path, content: str) -> None:
    """Durably replace one small agent config file on the same filesystem."""

    path.parent.mkdir(parents=True, exist_ok=True)
    temp_name: str | None = None
    try:
        with tempfile.NamedTemporaryFile(
            mode="w",
            encoding="utf-8",
            dir=path.parent,
            prefix=f".{path.name}.",
            suffix=".tmp",
            delete=False,
        ) as handle:
            temp_name = handle.name
            handle.write(content)
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(temp_name, path)
        temp_name = None
    finally:
        if temp_name is not None:
            try:
                os.unlink(temp_name)
            except FileNotFoundError:
                pass


async def _sync_legacy_agents(store: AgentStore) -> int:
    """Import existing folder-backed agents as built-in DB records.

    This preserves old deployments where agents already existed under
    ``.deer-flow/agents/{name}`` before the agents table was introduced.

    Heavy: scans the agents directory + parses each ``config.yaml`` + issues a
    SELECT-then-INSERT/UPDATE per agent. Runs once at startup, and again only
    when an administrator clicks 「重新扫描智能体目录」. List and detail
    requests must not call it — a full directory walk is what made category
    switches stall.

    Returns the number of agent directories scanned. Failures are logged and
    do not propagate, so a bad directory cannot block Gateway startup.
    """
    scanned = 0
    try:
        pairs: list[tuple[str, Any]] = []
        for agent_cfg in list_custom_agents():
            agent_id = agent_cfg.id or agent_cfg.name
            if not agent_id:
                continue
            await store.ensure_builtin(
                {
                    "id": agent_id,
                    "name": agent_cfg.name,
                    "description": agent_cfg.description or "",
                    "published": True,
                }
            )
            pairs.append((agent_id, agent_cfg))
            scanned += 1
        await _backfill_listing_cache(store, pairs)
    except Exception:
        logger.exception("Failed to sync legacy folder-backed agents")
    return scanned


async def _backfill_listing_cache(store: AgentStore, pairs: list[tuple[str, Any]]) -> None:
    """Mirror each agent's config.yaml skills/tool_groups/model into the DB cache.

    The list endpoint reads skills/tool_groups from these cache rows instead of
    stat+parsing every ``config.yaml`` per request. We fetch the current cache
    in one batch and only re-write rows whose values actually changed, so a
    steady-state sync (every ``_LEGACY_SYNC_TTL_SECONDS``) is cheap and does not
    churn DELETE/INSERTs. For the cache, ``None`` and ``[]`` skills are
    equivalent ("no skills") — the tri-state only matters to the agent runtime,
    which reads config.yaml directly.
    """
    if not pairs:
        return
    cached = await store.get_extras_for([aid for aid, _ in pairs])
    for agent_id, cfg in pairs:
        want_skills = [s for s in (cfg.skills or []) if isinstance(s, str)]
        want_tool_groups = cfg.tool_groups
        want_model = cfg.model
        want_chat_link = cfg.chat_link
        existing = cached.get(agent_id)
        if existing is None or existing.get("skills") != want_skills:
            await store.replace_skills(agent_id, want_skills)
        link_changed = existing is None or _chat_link_fingerprint(existing.get("chat_link")) != _chat_link_fingerprint(want_chat_link)
        extras_changed = existing is None or existing.get("tool_groups") != want_tool_groups or existing.get("model") != want_model or link_changed
        if extras_changed:
            await store.set_extras(
                agent_id,
                tool_groups=want_tool_groups,
                model=want_model,
                chat_link=want_chat_link,
                update_chat_link=link_changed,
            )


# Kept so a manual directory scan can mark the process fresh. Request handlers
# no longer call :func:`_maybe_sync_legacy_agents`; administrators trigger
# :func:`_sync_legacy_agents` from the gallery button instead.
_LEGACY_SYNC_TTL_SECONDS = 3600.0
_legacy_sync_state: dict[str, Any] = {"last_at": 0.0, "lock": None}


async def _maybe_sync_legacy_agents(store: AgentStore) -> None:
    """TTL-gated wrapper around :func:`_sync_legacy_agents`.

    No request handler calls this anymore. The gallery sync button and the
    startup lifespan call :func:`_sync_legacy_agents` directly. This wrapper
    remains for callers that still want the hourly safety net.
    """
    now = time.monotonic()
    if now - float(_legacy_sync_state["last_at"]) < _LEGACY_SYNC_TTL_SECONDS:
        return
    lock = _legacy_sync_state["lock"]
    if lock is None:
        lock = asyncio.Lock()
        _legacy_sync_state["lock"] = lock
    async with lock:
        # Re-check under the lock — a sibling coroutine may have just refreshed.
        if time.monotonic() - float(_legacy_sync_state["last_at"]) < _LEGACY_SYNC_TTL_SECONDS:
            return
        await _sync_legacy_agents(store)
        _legacy_sync_state["last_at"] = time.monotonic()


def _mark_legacy_sync_fresh() -> None:
    """Called by the lifespan hook after running :func:`_sync_legacy_agents`."""
    _legacy_sync_state["last_at"] = time.monotonic()


def _ensure_agent_dir_from_record(record: dict[str, Any]) -> None:
    """Lazily restore an agent's on-disk artifacts from its DB record.

    The DB row is authoritative for the agent's existence. If the matching
    ``{base_dir}/agents/{id}/`` directory is missing (data dir moved between
    machines, filesystem wiped, etc.) we recreate it with a minimal
    ``config.yaml`` + empty ``SOUL.md`` synthesized from the row. This restores
    a usable agent without requiring the admin to manually delete + recreate.
    """
    agent_id = record.get("id")
    if not agent_id:
        return
    try:
        agent_dir = get_paths().agent_dir(agent_id)
        if agent_dir.exists():
            return
        logger.warning(
            "Agent dir missing for '%s' — rebuilding minimal config.yaml + SOUL.md from DB",
            agent_id,
        )
        _write_agent_config(
            agent_id,
            name=record.get("name") or agent_id,
            description=record.get("description") or "",
        )
        _write_agent_soul(agent_id, "")
    except Exception:
        # Self-heal is best-effort. ``load_agent_config`` already tolerates a
        # missing dir, so the worst case is the agent runs with default
        # settings until someone re-edits it.
        logger.exception("Failed to self-heal agent dir for '%s'", agent_id)


def _record_to_response(
    record: dict[str, Any],
    *,
    include_soul: bool = False,
    include_extras: bool = True,
) -> AgentResponse:
    """Convert a DB row to an API response.

    Args:
        include_soul: Read SOUL.md from disk and embed in ``soul``. Set False
            in list endpoints — the frontend (gallery / recommend / selector)
            never reads ``soul``, only the edit dialog does. Skipping this
            avoids one synchronous file read per agent.
        include_extras: Self-heal a missing agent dir and read deprecated
            ``model`` / ``tool_groups`` / ``skills`` from ``config.yaml``. Set
            False in list endpoints to skip a stat + yaml parse per agent —
            those fields are unused by current frontends and the dir restore
            can happen lazily when the agent is actually opened.
    """
    soul: str | None = None
    extras: dict[str, Any] = {
        "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,
    }
    if include_extras:
        _ensure_agent_dir_from_record(record)
        extras = _load_config_extras(record["id"])
    if include_soul:
        soul = load_agent_soul(record["id"]) or ""

    return AgentResponse(
        id=record["id"],
        name=record["name"],
        description=record.get("description") or "",
        user_id=record.get("user_id"),
        published=bool(record.get("published", False)),
        is_private=bool(record.get("is_private", False)),
        hidden=bool(record.get("hidden", False)),
        square_id=str(record.get("square_id") or ""),
        builtin=record.get("user_id") is None,
        created_at=record.get("created_at"),
        updated_at=record.get("updated_at"),
        soul=soul,
        model=extras["model"],
        model_prompt_suffix_enabled=extras["model_prompt_suffix_enabled"],
        tool_groups=extras["tool_groups"],
        skills=extras["skills"],
        common_questions=extras.get("common_questions"),
        template_name=extras.get("template_name"),
        seat_skill_directive=extras.get("seat_skill_directive"),
        llmwiki_knowledge_base_ids=extras.get("llmwiki_knowledge_base_ids"),
        chat_link=_chat_link_model(extras.get("chat_link")),
        featured_order=record.get("featured_order"),
    )


def _sort_agent_responses(responses: list[AgentResponse], sort: str) -> list[AgentResponse]:
    reverse = sort not in {"time_asc", "updated_asc", "name_asc"}
    if sort in {"name_asc", "name_desc"}:
        ordered = sorted(responses, key=lambda r: (r.name or "").lower(), reverse=reverse)
    elif sort in {"updated_desc", "updated_asc"}:
        ordered = sorted(responses, key=lambda r: str(r.updated_at or ""), reverse=reverse)
    else:
        ordered = sorted(responses, key=lambda r: str(r.created_at or ""), reverse=reverse)
    return sorted(ordered, key=lambda r: 0 if r.is_pinned else 1)


async def _record_to_response_async(
    record: dict[str, Any],
    *,
    include_soul: bool = False,
    include_extras: bool = True,
) -> AgentResponse:
    """Async wrapper: when we actually touch disk, hop to a worker thread so
    we don't block the asyncio loop. List endpoints (no soul, no extras) skip
    this overhead and stay on the calling task."""
    if not include_soul and not include_extras:
        return _record_to_response(record, include_soul=False, include_extras=False)
    return await asyncio.to_thread(
        _record_to_response,
        record,
        include_soul=include_soul,
        include_extras=include_extras,
    )


async def _get_owned_or_raise(store: AgentStore, agent_id: str, user_id: str) -> dict[str, Any]:
    owned = await store.get_owned(agent_id, user_id)
    if owned is not None:
        return owned
    visible = await store.get_visible(agent_id, user_id)
    if visible is None:
        raise HTTPException(status_code=404, detail=f"Agent '{agent_id}' not found")
    raise HTTPException(status_code=403, detail="You can only modify agents that you created")


async def _get_editable_or_raise(
    store: AgentStore,
    agent_id: str,
    user_id: str,
    *,
    is_admin: bool,
) -> tuple[dict[str, Any], str]:
    """Return (record, edit_mode) where edit_mode is 'owner' or 'admin'.

    Admins may edit any agent — built-in (``user_id IS NULL``) or owned by
    another user — so the admin console can manage the full agent catalog.
    Non-admin callers are still restricted to agents they own.
    """
    owned = await store.get_owned(agent_id, user_id)
    if owned is not None:
        return owned, "owner"
    if is_admin:
        record = await store.get_any(agent_id)
        if record is None:
            raise HTTPException(status_code=404, detail=f"Agent '{agent_id}' not found")
        return record, "admin"
    visible = await store.get_visible(agent_id, user_id)
    if visible is None:
        raise HTTPException(status_code=404, detail=f"Agent '{agent_id}' not found")
    raise HTTPException(status_code=403, detail="You can only modify agents that you created")


async def _lookup_skill_labels(skill_store: Any, names: set[str]) -> dict[str, str]:
    """Resolve Chinese skill names from the skills table, never the filesystem."""
    if not names or skill_store is None:
        return {}
    lookup = getattr(skill_store, "list_name_zh", None)
    if lookup is None:
        return {}
    try:
        labels = await lookup(list(names))
    except Exception:
        logger.debug("skill display-name lookup failed", exc_info=True)
        return {}
    if not isinstance(labels, dict):
        return {}
    return {str(name): str(label) for name, label in labels.items() if name and label}


async def _build_list_responses(
    store: AgentStore,
    tag_store: TagStore,
    user_id: str,
    records: list[dict[str, Any]],
    skill_store: Any = None,
    trace: dict[str, Any] | None = None,
) -> list[AgentResponse]:
    """Build gallery responses for a set of agent rows WITHOUT touching disk.

    skills / tool_groups / model come from the denormalized DB cache
    (``get_extras_for`` — one batched query) instead of stat+parsing every
    agent's ``config.yaml`` per request, which was the list-endpoint hot spot.
    SOUL.md is skipped (only the edit dialog reads it). An agent absent from
    the cache (never mirrored) keeps its ``None`` extras until the next sync.
    """
    stage_started = time.perf_counter()
    responses = [_record_to_response(a, include_soul=False, include_extras=False) for a in records]
    if trace is not None:
        trace_stage(trace, "agent_response_models", stage_started)
    ids = [r.id for r in responses]
    async def _timed(stage_name: str, awaitable):
        stage_started = time.perf_counter()
        result = await awaitable
        if trace is not None:
            trace_stage(trace, stage_name, stage_started)
        return result

    extras, assignments, favorite_ids, pin_ids, usernames = await asyncio.gather(
        _timed("agents_extras_db", store.get_extras_for(ids)),
        _timed("agent_tags_db", tag_store.list_assignments("agent", ids)),
        _timed("agent_favorites_db", store.list_favorite_ids(user_id)),
        _timed("agent_pins_db", store.list_pin_ids(user_id)),
        _timed("agent_usernames_db", _resolve_usernames({r.user_id for r in responses if r.user_id})),
    )
    skill_names: set[str] = set()
    for resp in responses:
        cached = extras.get(resp.id)
        if cached is not None:
            resp.skills = cached.get("skills")
            resp.tool_groups = cached.get("tool_groups")
            resp.model = cached.get("model")
            resp.chat_link = _chat_link_model(cached.get("chat_link"))
        for skill_name in resp.skills or []:
            if skill_name:
                skill_names.add(skill_name)
    stage_started = time.perf_counter()
    labels = await _lookup_skill_labels(skill_store, skill_names)
    if trace is not None:
        trace_stage(trace, "agent_skill_labels_db", stage_started)
    for resp in responses:
        resp.tags = assignments.get(resp.id, [])
        resp.is_favorite = resp.id in favorite_ids
        resp.is_pinned = resp.id in pin_ids
        if resp.user_id:
            resp.owner_username = usernames.get(resp.user_id)
        if labels and resp.skills:
            resp.skill_labels = {name: labels[name] for name in resp.skills if name in labels}
    return responses


@router.get(
    "/agents",
    response_model=AgentsListResponse,
    summary="List Agents",
    description="List built-in agents, the current user's agents, and published agents.",
)
async def list_agents(
    request: Request,
    search: str | None = Query(default=None, description="Case-insensitive fuzzy match on agent name (raw or desensitized display form)"),
    tag_ids: list[str] | None = Query(default=None, description="Keep agents carrying any of these tags (OR)"),
    scope: str | None = Query(
        default=None,
        pattern="^(mine|square|builtin)$",
        description="Pagination scope. 'mine' = owned + favorited; 'square' = built-in/published public; 'builtin' = 内置智能体 (user_id IS NULL, **admin only**). Omit for the default combined visibility list.",
    ),
    page: int | None = Query(default=None, ge=1, description="1-based page index. Providing this (or 'scope') enables server-side pagination."),
    page_size: int = Query(default=20, ge=1, le=200, description="Items per page when paginating."),
    sort: str = Query(
        default="time_desc",
        pattern="^(time_desc|time_asc|updated_desc|updated_asc|name_asc|name_desc)$",
        description="Sort order. Default is newest created first.",
    ),
    owner: str | None = Query(
        default=None,
        description="Keep only agents published by this user id (the card's publisher). Built-in agents never match.",
    ),
    square: str | None = Query(
        default=None,
        description="Keep only agents filed under this publish square (发布广场). Omit for all squares.",
    ),
    admin: bool = Query(
        default=False,
        description=("Admin-only override. When ``true`` and the caller is a system admin, every agent is returned (built-in + every user's agents + private-square entries) regardless of the usual ownership/visibility filter."),
    ),
    hidden: bool | None = Query(
        default=None,
        description=("Admin-only filter (requires ``admin=true``): keep only hidden (``true``) or only visible (``false``) agents. Omit for both. Ignored for non-admin callers, who never see hidden agents anyway."),
    ),
) -> AgentsListResponse:
    trace = trace_request(request, "agents_list")
    cache_fill_token: str | None = None
    cache_config = None
    cache_key = ""
    try:
        _require_agents_api_enabled()
        stage_started = time.perf_counter()
        store = get_agent_store(request)
        tag_store: TagStore = get_tag_store(request)
        skill_store = get_skill_store(request)
        user_id = _current_user_id(request)
        if admin and not _is_admin_user(request):
            raise HTTPException(status_code=403, detail="Only system admins can list all agents")
        # 「内置智能体」筛选仅限管理员（普通用户不应看到全部内置/功能型 agent 目录）。
        if scope == "builtin" and not _is_admin_user(request):
            raise HTTPException(status_code=403, detail="Only system admins can list built-in agents")
        trace_stage(trace, "request_setup_and_access_checks", stage_started)

        paginate = scope is not None or page is not None
        cacheable = (
            not admin
            and scope in (None, "mine", "square")
            and not (search or "").strip()
            and not tag_ids
            and not (owner or "").strip()
            and not (square or "").strip()
            and sort == "time_desc"
        )
        query_fingerprint = json.dumps(
            {
                "scope": scope,
                "page": page,
                "page_size": page_size,
                "sort": sort,
                "search": search,
                "tag_ids": sorted(tag_ids or []),
                "owner": owner,
                "square": square,
                "hidden": hidden,
            },
            ensure_ascii=False,
            sort_keys=True,
            separators=(",", ":"),
        )
        cache_key = f"user:{user_id}:query:{query_fingerprint}" if cacheable else ""
        cache_version = None
        request_app_state = getattr(getattr(request, "app", None), "state", None)
        cache_config = getattr(request_app_state, "config", None)
        if cacheable:
            stage_started = time.perf_counter()
            cache_version = await get_cache_version(cache_config, "agents")
            cached_payload = await get_cached_json_payload(
                cache_config,
                "agents",
                cache_key,
                version=cache_version,
            )
            if cached_payload is None:
                cached_payload, cache_fill_token = await wait_for_cached_json_payload_or_claim_fill(
                    cache_config,
                    "agents",
                    cache_key,
                    version=cache_version,
                )
            trace_stage(trace, "agents_redis_cache", stage_started)
            if cached_payload is not None:
                trace.setdefault("details", {})["cache"] = "redis_hit"
                return Response(content=cached_payload, media_type="application/json")
        square_id = square.strip() if square and square.strip() else None
        square_is_default = False
        if square_id:
            stage_started = time.perf_counter()
            sq = load_system_settings().publish_squares
            square_is_default = square_id == resolve_default_square_id(sq)
            trace_stage(trace, "load_publish_square_settings", stage_started)
        if paginate:
            eff_page = page or 1
            memory_cache_key = (
                user_id,
                bool(admin),
                scope,
                (search or "").strip(),
                tuple(sorted(tag_ids or [])),
                eff_page,
                page_size,
                sort,
                (owner or "").strip(),
                square_id or "",
                square_is_default,
                hidden if admin else None,
            )

            async def _load_page() -> AgentsListResponse:
                stage_started = time.perf_counter()
                rows, total = await store.list_paginated(
                    user_id=user_id,
                    scope=scope,
                    search=search,
                    tag_ids=tag_ids,
                    page=eff_page,
                    page_size=page_size,
                    admin=admin,
                    sort=sort,
                    owner=owner,
                    square_id=square_id,
                    square_is_default=square_is_default,
                    hidden=hidden if admin else None,
                )
                trace_stage(trace, "agents_page_query_db", stage_started)
                stage_started = time.perf_counter()
                responses = await _build_list_responses(store, tag_store, user_id, rows, skill_store, trace=trace)
                trace_stage(trace, "agents_page_assembly", stage_started)
                result = AgentsListResponse(agents=responses, total=total, page=eff_page, page_size=page_size)
                if cacheable:
                    await set_cached_json(
                        cache_config,
                        "agents",
                        cache_key,
                        result,
                        expected_version=cache_version,
                    )
                    trace.setdefault("details", {})["cache"] = "redis_miss"
                return result

            if admin:
                result = await _load_page()
                memory_cache_status = "bypass_admin"
            else:
                result, memory_cache_status = await _cached_paginated_agent_list(memory_cache_key, _load_page)
            trace.setdefault("details", {})["memory_cache"] = memory_cache_status
            return result

        # Legacy unpaginated path — returns everything (admin console, callers
        # that have not migrated to pagination). Extras now come from the DB
        # cache too, so this path no longer reads config.yaml per agent.
        stage_started = time.perf_counter()
        agents = await store.list_all() if admin else await store.list_visible(user_id)
        trace_stage(trace, "agents_full_list_query_db", stage_started)
        stage_started = time.perf_counter()
        responses = await _build_list_responses(store, tag_store, user_id, agents, skill_store, trace=trace)
        trace_stage(trace, "agents_full_list_assembly", stage_started)
        stage_started = time.perf_counter()
        if search:
            needle = search.strip().lower()
            responses = [r for r in responses if needle in r.name.lower()]
        if tag_ids:
            wanted = set(tag_ids)
            responses = [r for r in responses if any(t.get("id") in wanted for t in r.tags)]
        if owner and owner.strip():
            responses = [r for r in responses if r.user_id == owner.strip()]
        if square_id:
            responses = [r for r in responses if (r.square_id or "") == square_id or (square_is_default and not (r.square_id or ""))]
        if admin and hidden is not None:
            responses = [r for r in responses if r.hidden == hidden]
        responses = _sort_agent_responses(responses, sort)
        trace_stage(trace, "agents_filter_and_sort", stage_started)
        result = AgentsListResponse(agents=responses, total=len(responses), page=1, page_size=len(responses))
        if cacheable:
            await set_cached_json(
                cache_config,
                "agents",
                cache_key,
                result,
                expected_version=cache_version,
            )
            trace.setdefault("details", {})["cache"] = "redis_miss"
        return result
    except HTTPException:
        raise
    except Exception as e:
        logger.error("Failed to list agents: %s", e, exc_info=True)
        raise HTTPException(status_code=500, detail=f"Failed to list agents: {str(e)}") from e
    finally:
        if cache_fill_token is not None and cache_config is not None and cache_key:
            await release_cache_fill(cache_config, "agents", cache_key, cache_fill_token)
        finish_trace(trace)


@router.get(
    "/agents/check",
    summary="Check Agent Name",
    description="Validate an agent display name. Duplicate names are allowed.",
)
async def check_agent_name(name: str) -> dict[str, Any]:
    _require_agents_api_enabled()
    normalized = _validate_display_name(name)
    return {"available": True, "name": normalized}


@router.get(
    "/agents/selector",
    response_model=AgentSelectorResponse,
    summary="Agent Homepage Selector",
    description=("Return the admin-curated agent list shown on the new-chat homepage. Sorted by ``featured_order`` ascending. Visibility-filtered, so an agent the caller can't see (author un-published) is silently omitted."),
)
async def get_agent_selector(request: Request) -> AgentSelectorResponse:
    _require_agents_api_enabled()
    try:
        store = get_agent_store(request)

        user_id = _current_user_id(request)
        records = await store.list_featured(user_id)
        # Selector dropdown only renders name/description/featured_order — same
        # rationale as list_agents: skip soul + extras + self-heal.
        return AgentSelectorResponse(agents=[_record_to_response(r, include_soul=False, include_extras=False) for r in records])
    except HTTPException:
        raise
    except Exception as e:
        logger.error("Failed to load agent selector: %s", e, exc_info=True)
        raise HTTPException(status_code=500, detail=f"Failed to load agent selector: {str(e)}") from e


@router.get(
    "/agents/sidebar",
    response_model=AgentSidebarResponse,
    summary="Workspace sidebar agent names",
    description="Return at most three id/name rows for the workspace sidebar without loading descriptions, tags, skills, or filesystem data.",
)
async def get_agent_sidebar(request: Request) -> AgentSidebarResponse:
    _require_agents_api_enabled()
    try:
        records = await get_agent_store(request).list_sidebar(_current_user_id(request), limit=3)
        return AgentSidebarResponse(agents=records)
    except HTTPException:
        raise
    except Exception as exc:
        logger.error("Failed to load agent sidebar: %s", exc, exc_info=True)
        raise HTTPException(status_code=500, detail="Failed to load workspace sidebar agents") from exc


@router.get(
    "/agents/{agent_id}",
    response_model=AgentResponse,
    summary="Get Agent",
    description="Retrieve details and SOUL.md content for a visible agent.",
)
async def get_agent(agent_id: str, request: Request) -> AgentResponse:
    _require_agents_api_enabled()
    agent_id = _validate_storage_id(agent_id)

    try:
        store = get_agent_store(request)
        user_id = _current_user_id(request)
        record = await store.get_visible(agent_id, user_id)
        if record is None and _is_admin_user(request):
            # The admin console lists every agent (?admin=true), so an admin
            # must be able to open any of them — mirror that here.
            record = await store.get_any(agent_id)
        if record is None:
            raise HTTPException(status_code=404, detail=f"Agent '{agent_id}' not found")
        # Detail view legitimately reads SOUL.md + config.yaml — hop to a
        # worker thread so the asyncio loop stays responsive under concurrent
        # detail fetches.
        response = await _record_to_response_async(record, include_soul=True, include_extras=True)
        favorite_ids = await store.list_favorite_ids(user_id)
        pin_ids = await store.list_pin_ids(user_id)
        response.is_favorite = agent_id in favorite_ids
        response.is_pinned = agent_id in pin_ids
        return response
    except HTTPException:
        raise
    except Exception as e:
        logger.error("Failed to get agent '%s': %s", agent_id, e, exc_info=True)
        raise HTTPException(status_code=500, detail=f"Failed to get agent: {str(e)}") from e


@router.post(
    "/agents",
    response_model=AgentResponse,
    status_code=201,
    summary="Create Agent",
    description="Create a new user-owned agent. The display name can be Chinese and may duplicate existing names.",
)
async def create_agent_endpoint(request: AgentCreateRequest, http_request: Request) -> AgentResponse:
    _require_agents_api_enabled()
    name = _validate_display_name(request.name)
    store = get_agent_store(http_request)

    if request.id:
        agent_id = _validate_storage_id(request.id)
        if _is_strategy_agent_id(agent_id):
            raise HTTPException(
                status_code=409,
                detail=("The 'strategy-selection-' id namespace is reserved; create strategy angle agents through /api/strategy-agents/angles"),
            )
        if await store.get_any(agent_id) is not None:
            raise HTTPException(status_code=409, detail=f"Agent id '{agent_id}' already exists")
    else:
        agent_id = uuid4().hex
    agent_dir = get_paths().agent_dir(agent_id)
    llmwiki_ids = await _validate_llmwiki_bindings(
        http_request,
        request.llmwiki_knowledge_base_ids,
    )

    try:
        if agent_dir.exists():
            raise HTTPException(status_code=409, detail=f"Agent id '{agent_id}' already exists")

        chat_link = _require_chat_link(request.chat_link)
        _write_agent_config(
            agent_id,
            name=name,
            description=request.description or "",
            skills=request.skills,
            model=request.model,
            model_prompt_suffix_enabled=request.model_prompt_suffix_enabled,
            tool_groups=request.tool_groups,
            common_questions=request.common_questions,
            template_name=request.template_name,
            seat_skill_directive=request.seat_skill_directive,
            llmwiki_knowledge_base_ids=llmwiki_ids,
            chat_link=chat_link,
        )
        _write_agent_soul(agent_id, request.soul)

        record = await store.create(
            {
                "id": agent_id,
                "name": name,
                "description": request.description or "",
                "user_id": _current_user_id(http_request),
                "published": request.published,
                "is_private": request.is_private,
                # Normalize to a valid configured square id (default when unset).
                "square_id": normalize_square_id(request.square_id, load_system_settings().publish_squares),
            }
        )
        # Mirror skills/tool_groups/model into the listing cache so the gallery
        # can render this agent's badges without reading its config.yaml.
        await store.replace_skills(agent_id, request.skills)
        await store.set_extras(
            agent_id,
            tool_groups=request.tool_groups,
            model=request.model,
            chat_link=chat_link,
            update_chat_link=True,
        )
        logger.info("Created agent '%s' (%s) at %s", name, agent_id, agent_dir)
        return await _record_to_response_async(record, include_soul=True, include_extras=True)

    except HTTPException:
        raise
    except Exception as e:
        if agent_dir.exists():
            shutil.rmtree(agent_dir)
        logger.error("Failed to create agent '%s': %s", request.name, e, exc_info=True)
        raise HTTPException(status_code=500, detail=f"Failed to create agent: {str(e)}") from e


@router.put(
    "/agents/{agent_id}",
    response_model=AgentResponse,
    summary="Update Agent",
    description="Update an existing user-owned agent.",
)
async def update_agent(agent_id: str, request: AgentUpdateRequest, http_request: Request) -> AgentResponse:
    _require_agents_api_enabled()
    agent_id = _validate_storage_id(agent_id)
    if _is_strategy_agent_id(agent_id):
        raise HTTPException(
            status_code=409,
            detail=("Strategy agent configuration must be changed through /api/strategy-agents/{agent_id}"),
        )

    try:
        store = get_agent_store(http_request)
        user_id = _current_user_id(http_request)
        is_admin = _is_admin_user(http_request)
        current, edit_mode = await _get_editable_or_raise(store, agent_id, user_id, is_admin=is_admin)
        _require_canonical_record_id(agent_id, current)

        fields_set = request.model_fields_set
        update_data: dict[str, Any] = {}
        if "name" in fields_set:
            update_data["name"] = _validate_display_name(request.name or "")
        if "description" in fields_set:
            update_data["description"] = request.description or ""
        if "published" in fields_set:
            update_data["published"] = bool(request.published)
        if "is_private" in fields_set:
            update_data["is_private"] = bool(request.is_private)
        if "square_id" in fields_set:
            # Normalize to a valid configured square id (default when unset /
            # unknown), so the row never holds a dangling square reference.
            update_data["square_id"] = normalize_square_id(request.square_id, load_system_settings().publish_squares)
        # Mutual exclusion: an agent moved into the private square cannot
        # also be publicly published. Force ``published=false`` whenever the
        # same request sets ``is_private=true``, even if the client forgot
        # to clear it. (When the request only sets one of the two, the
        # other side is left untouched.)
        if update_data.get("is_private") is True:
            update_data["published"] = False

        next_name = update_data.get("name", current["name"])
        next_description = update_data.get("description", current.get("description") or "")

        existing_extras = _load_config_extras(agent_id)
        if "skills" in fields_set and _is_strategy_agent_id(agent_id):
            next_skills = await _validate_strategy_agent_skills(
                request.skills,
                app_config=get_config(http_request),
                skill_store=get_skill_store(http_request),
            )
        else:
            next_skills = request.skills if "skills" in fields_set else existing_extras["skills"]
        if "model" in fields_set and _is_strategy_agent_id(agent_id):
            next_model = _validate_strategy_agent_model(
                request.model,
                app_config=get_config(http_request),
            )
        else:
            next_model = request.model if "model" in fields_set else existing_extras["model"]
        next_model_prompt_suffix_enabled = (
            bool(request.model_prompt_suffix_enabled)
            if "model_prompt_suffix_enabled" in fields_set
            else existing_extras["model_prompt_suffix_enabled"]
        )
        next_tool_groups = request.tool_groups if "tool_groups" in fields_set else existing_extras["tool_groups"]
        next_common_questions = request.common_questions if "common_questions" in fields_set else existing_extras["common_questions"]
        next_template_name = request.template_name if "template_name" in fields_set else existing_extras["template_name"]
        next_seat_skill_directive = request.seat_skill_directive if "seat_skill_directive" in fields_set else existing_extras["seat_skill_directive"]
        next_llmwiki_ids = await _validate_llmwiki_bindings(http_request, request.llmwiki_knowledge_base_ids) if "llmwiki_knowledge_base_ids" in fields_set else existing_extras["llmwiki_knowledge_base_ids"]
        next_chat_link = _require_chat_link(request.chat_link) if "chat_link" in fields_set else existing_extras["chat_link"]

        config_dirty = bool(update_data) or any(
            key in fields_set
            for key in (
                "skills",
                "model",
                "model_prompt_suffix_enabled",
                "tool_groups",
                "common_questions",
                "template_name",
                "seat_skill_directive",
                "llmwiki_knowledge_base_ids",
                "chat_link",
            )
        )
        if config_dirty:
            _write_agent_config(
                agent_id,
                name=next_name,
                description=next_description,
                skills=next_skills,
                model=next_model,
                model_prompt_suffix_enabled=next_model_prompt_suffix_enabled,
                tool_groups=next_tool_groups,
                common_questions=next_common_questions,
                template_name=next_template_name,
                seat_skill_directive=next_seat_skill_directive,
                llmwiki_knowledge_base_ids=next_llmwiki_ids,
                chat_link=next_chat_link,
            )
            # Keep the listing cache in lock-step with config.yaml, but only
            # when the cached fields were actually touched — a partial update
            # (e.g. 常用问题 only) skips the DELETE/INSERT churn.
            if any(key in fields_set for key in ("skills", "model", "tool_groups", "chat_link")):
                if any(key in fields_set for key in ("skills", "model", "tool_groups")):
                    await store.replace_skills(agent_id, next_skills)
                await store.set_extras(
                    agent_id,
                    tool_groups=next_tool_groups,
                    model=next_model,
                    chat_link=next_chat_link,
                    update_chat_link="chat_link" in fields_set,
                )
        if request.soul is not None:
            _write_agent_soul(agent_id, request.soul)

        if not update_data:
            record = current
        elif edit_mode == "admin":
            # Admins may edit built-in (``user_id IS NULL``) and any user-owned
            # agent — ``update_any`` skips the ownership check that ``update``
            # and ``update_builtin`` enforce.
            record = await store.update_any(agent_id, update_data)
        else:
            record = await store.update(agent_id, user_id, update_data)
        if record is None:
            raise HTTPException(status_code=404, detail=f"Agent '{agent_id}' not found")

        logger.info("Updated agent '%s' (%s)", record["name"], agent_id)
        return await _record_to_response_async(record, include_soul=True, include_extras=True)

    except HTTPException:
        raise
    except Exception as e:
        logger.error("Failed to update agent '%s': %s", agent_id, e, exc_info=True)
        raise HTTPException(status_code=500, detail=f"Failed to update agent: {str(e)}") from e


class UserProfileResponse(BaseModel):
    """Response model for the global user profile (USER.md)."""

    content: str | None = Field(default=None, description="USER.md content, or null if not yet created")


class UserProfileUpdateRequest(BaseModel):
    """Request body for setting the global user profile."""

    content: str = Field(default="", description="USER.md content - describes the user's background and preferences")


@router.get(
    "/user-profile",
    response_model=UserProfileResponse,
    summary="Get User Profile",
    description="Read the global USER.md file that is injected into all custom agents.",
)
async def get_user_profile() -> UserProfileResponse:
    _require_agents_api_enabled()

    try:
        user_md_path = get_paths().user_md_file
        if not user_md_path.exists():
            return UserProfileResponse(content=None)
        raw = user_md_path.read_text(encoding="utf-8").strip()
        return UserProfileResponse(content=raw or None)
    except Exception as e:
        logger.error("Failed to read user profile: %s", e, exc_info=True)
        raise HTTPException(status_code=500, detail=f"Failed to read user profile: {str(e)}") from e


@router.put(
    "/user-profile",
    response_model=UserProfileResponse,
    summary="Update User Profile",
    description="Write the global USER.md file that is injected into all custom agents.",
)
async def update_user_profile(request: UserProfileUpdateRequest) -> UserProfileResponse:
    _require_agents_api_enabled()

    try:
        paths = get_paths()
        paths.base_dir.mkdir(parents=True, exist_ok=True)
        paths.user_md_file.write_text(request.content, encoding="utf-8")
        logger.info("Updated USER.md at %s", paths.user_md_file)
        return UserProfileResponse(content=request.content or None)
    except Exception as e:
        logger.error("Failed to update user profile: %s", e, exc_info=True)
        raise HTTPException(status_code=500, detail=f"Failed to update user profile: {str(e)}") from e


@router.delete(
    "/agents/{agent_id}",
    status_code=204,
    summary="Delete Agent",
    description="Delete a user-owned agent and all its files.",
)
async def delete_agent(agent_id: str, request: Request) -> None:
    _require_agents_api_enabled()
    agent_id = _validate_storage_id(agent_id)
    if _is_strategy_agent_id(agent_id):
        raise HTTPException(
            status_code=409,
            detail=("Strategy agents cannot be deleted through the generic API; use /api/strategy-agents/{agent_id} to disable an angle"),
        )

    try:
        store = get_agent_store(request)
        user_id = _current_user_id(request)
        is_admin = _is_admin_user(request)
        record, edit_mode = await _get_editable_or_raise(store, agent_id, user_id, is_admin=is_admin)
        _require_canonical_record_id(agent_id, record)
        _ensure_strategy_agent_deletable(agent_id)

        if edit_mode == "admin":
            deleted = await store.delete_any(agent_id)
        else:
            deleted = await store.delete(agent_id, user_id)
        if not deleted:
            raise HTTPException(status_code=404, detail=f"Agent '{agent_id}' not found")

        agent_dir = get_paths().agent_dir(agent_id)
        if agent_dir.exists():
            shutil.rmtree(agent_dir)
        try:
            await get_tag_store(request).unassign_all("agent", agent_id)
        except Exception:  # noqa: BLE001 — tag cleanup must not block deletion
            pass
        try:
            await store.delete_favorites_for_agent(agent_id)
        except Exception:  # noqa: BLE001 — favorite cleanup must not block deletion
            pass
        try:
            await store.delete_pins_for_agent(agent_id)
        except Exception:  # noqa: BLE001 - pin cleanup must not block deletion
            pass
        try:
            await store.delete_extras_for_agent(agent_id)
        except Exception:  # noqa: BLE001 — cache cleanup must not block deletion
            pass
        logger.info("Deleted agent '%s' from %s", agent_id, agent_dir)
    except HTTPException:
        raise
    except Exception as e:
        logger.error("Failed to delete agent '%s': %s", agent_id, e, exc_info=True)
        raise HTTPException(status_code=500, detail=f"Failed to delete agent: {str(e)}") from e


@router.post(
    "/agents/{agent_id}/favorite",
    status_code=204,
    summary="Favorite Agent",
    description="Bookmark a visible agent into the current user's 我的 tab. Idempotent.",
)
async def favorite_agent(agent_id: str, request: Request) -> None:
    _require_agents_api_enabled()
    agent_id = _validate_storage_id(agent_id)

    try:
        store = get_agent_store(request)
        user_id = _current_user_id(request)
        record = await store.get_visible(agent_id, user_id)
        if record is None:
            raise HTTPException(status_code=404, detail=f"Agent '{agent_id}' not found")
        await store.add_favorite(user_id, agent_id)
        logger.info("User '%s' favorited agent '%s'", user_id, agent_id)
    except HTTPException:
        raise
    except Exception as e:
        logger.error("Failed to favorite agent '%s': %s", agent_id, e, exc_info=True)
        raise HTTPException(status_code=500, detail=f"Failed to favorite agent: {str(e)}") from e


@router.delete(
    "/agents/{agent_id}/favorite",
    status_code=204,
    summary="Unfavorite Agent",
    description="Remove an agent from the current user's favorites. No-op when not favorited.",
)
async def unfavorite_agent(agent_id: str, request: Request) -> None:
    _require_agents_api_enabled()
    agent_id = _validate_storage_id(agent_id)

    try:
        store = get_agent_store(request)
        user_id = _current_user_id(request)
        await store.remove_favorite(user_id, agent_id)
        logger.info("User '%s' unfavorited agent '%s'", user_id, agent_id)
    except HTTPException:
        raise
    except Exception as e:
        logger.error("Failed to unfavorite agent '%s': %s", agent_id, e, exc_info=True)
        raise HTTPException(status_code=500, detail=f"Failed to unfavorite agent: {str(e)}") from e


@router.post(
    "/agents/{agent_id}/pin",
    status_code=204,
    summary="Pin Agent",
    description="Pin a visible agent to the top of the current user's lists. Idempotent.",
)
async def pin_agent(agent_id: str, request: Request) -> None:
    _require_agents_api_enabled()
    agent_id = _validate_storage_id(agent_id)

    try:
        store = get_agent_store(request)
        user_id = _current_user_id(request)
        record = await store.get_visible(agent_id, user_id)
        if record is None:
            raise HTTPException(status_code=404, detail=f"Agent '{agent_id}' not found")
        await store.add_pin(user_id, agent_id)
        logger.info("User '%s' pinned agent '%s'", user_id, agent_id)
    except HTTPException:
        raise
    except Exception as e:
        logger.error("Failed to pin agent '%s': %s", agent_id, e, exc_info=True)
        raise HTTPException(status_code=500, detail=f"Failed to pin agent: {str(e)}") from e


@router.delete(
    "/agents/{agent_id}/pin",
    status_code=204,
    summary="Unpin Agent",
    description="Remove the current user's pin from an agent. No-op when not pinned.",
)
async def unpin_agent(agent_id: str, request: Request) -> None:
    _require_agents_api_enabled()
    agent_id = _validate_storage_id(agent_id)

    try:
        store = get_agent_store(request)
        user_id = _current_user_id(request)
        await store.remove_pin(user_id, agent_id)
        logger.info("User '%s' unpinned agent '%s'", user_id, agent_id)
    except HTTPException:
        raise
    except Exception as e:
        logger.error("Failed to unpin agent '%s': %s", agent_id, e, exc_info=True)
        raise HTTPException(status_code=500, detail=f"Failed to unpin agent: {str(e)}") from e


@router.put(
    "/agents/{agent_id}/featured",
    response_model=AgentResponse,
    summary="Set Agent Featured Rank (admin)",
    description=(
        "Admin-only. Set or clear the homepage-selector featured rank for any agent. "
        "Requires the agent to be ``published=true`` when a non-null rank is supplied; "
        "passing ``order: null`` removes the agent from the selector and is always allowed."
    ),
)
async def set_agent_featured(
    agent_id: str,
    body: AgentFeaturedRequest,
    request: Request,
) -> AgentResponse:
    _require_agents_api_enabled()
    agent_id = _validate_storage_id(agent_id)

    if not _is_admin_user(request):
        raise HTTPException(status_code=403, detail="Only system admins can set the homepage featured rank")

    try:
        store = get_agent_store(request)
        record = await store.get_any(agent_id)
        if record is None:
            raise HTTPException(status_code=404, detail=f"Agent '{agent_id}' not found")

        if body.order is not None and not record.get("published"):
            raise HTTPException(status_code=409, detail="Only published agents can be featured on the homepage")

        updated = await store.set_featured(agent_id, body.order)
        if updated is None:
            raise HTTPException(status_code=404, detail=f"Agent '{agent_id}' not found")
        logger.info("Set agent '%s' featured_order=%s by admin", agent_id, body.order)
        return await _record_to_response_async(updated, include_soul=True, include_extras=True)
    except HTTPException:
        raise
    except Exception as e:
        logger.error("Failed to set featured for agent '%s': %s", agent_id, e, exc_info=True)
        raise HTTPException(status_code=500, detail=f"Failed to set featured: {str(e)}") from e


class AgentHiddenUpdateRequest(BaseModel):
    """Admin-only request body for hiding an agent from every non-admin listing."""

    hidden: bool = Field(..., description="Set true to hide the agent from all non-admin listings (gallery / 我的 / 广场 / selector / private square); false restores it. Purely a catalog switch — chats and runs keep working either way.")


@router.put(
    "/agents/{agent_id}/hidden",
    response_model=AgentResponse,
    summary="Set Agent Hidden (admin)",
    description=(
        "Admin-only. Hide an agent from every non-admin listing surface, or restore "
        "it. Hidden agents remain fully usable (existing chats, direct open, runs) — "
        "only catalog listings drop them; the admin agents console (?admin=true) "
        "keeps listing them so they can be un-hidden."
    ),
)
async def set_agent_hidden(
    agent_id: str,
    body: AgentHiddenUpdateRequest,
    request: Request,
) -> AgentResponse:
    _require_agents_api_enabled()
    agent_id = _validate_storage_id(agent_id)

    if not _is_admin_user(request):
        raise HTTPException(status_code=403, detail="Only system admins can hide or restore agents")

    try:
        store = get_agent_store(request)
        record = await store.get_any(agent_id)
        if record is None:
            raise HTTPException(status_code=404, detail=f"Agent '{agent_id}' not found")

        updated = await store.update_any(agent_id, {"hidden": body.hidden})
        if updated is None:
            raise HTTPException(status_code=404, detail=f"Agent '{agent_id}' not found")
        logger.info("Set agent '%s' hidden=%s by admin", agent_id, body.hidden)
        return await _record_to_response_async(updated, include_soul=True, include_extras=True)
    except HTTPException:
        raise
    except Exception as e:
        logger.error("Failed to set hidden for agent '%s': %s", agent_id, e, exc_info=True)
        raise HTTPException(status_code=500, detail=f"Failed to set hidden: {str(e)}") from e


class PrivateSquareListRequest(BaseModel):
    """Request body for listing all private-square agents."""

    password: str = Field(..., description="Plain-text private-square password. Verified against the admin-configured hash before the listing is returned.")
    search: str | None = Field(default=None, description="Case-insensitive fuzzy match on agent name (raw or desensitized).")
    tag_ids: list[str] | None = Field(default=None, description="Keep agents carrying any of these tags (OR).")
    page: int | None = Field(default=None, ge=1, description="1-based page index. Providing this enables server-side pagination.")
    page_size: int = Field(default=20, ge=1, le=200, description="Items per page when paginating.")
    sort: str = Field(
        default="time_desc",
        pattern="^(time_desc|time_asc|updated_desc|updated_asc|name_asc|name_desc)$",
        description="Sort order. Default is newest created first.",
    )
    owner: str | None = Field(default=None, description="Keep only agents published by this user id (the card's publisher).")


class AgentPrivateUpdateRequest(BaseModel):
    """Admin-only request body for moving an agent into / out of the private square."""

    is_private: bool = Field(..., description="Set true to move the agent into the private square, false to move it back into the public square.")


@router.post(
    "/agents/private-square/list",
    response_model=AgentsListResponse,
    summary="List Private-Square Agents",
    description=(
        "Verify the admin-configured private-square password and return every "
        "agent flagged ``is_private=true``. Returns 404 when no password is set, "
        "403 when the supplied password does not match, and the full agent list "
        "otherwise — regardless of ownership, since the password is the gate."
    ),
)
async def list_private_square(body: PrivateSquareListRequest, request: Request) -> AgentsListResponse:
    _require_agents_api_enabled()

    settings = load_system_settings().private_square
    if not settings.password_hash:
        raise HTTPException(status_code=404, detail="Private square password is not configured")
    if not body.password or not await verify_password_async(body.password, settings.password_hash):
        raise HTTPException(status_code=403, detail="Incorrect private square password")

    try:
        store = get_agent_store(request)
        tag_store: TagStore = get_tag_store(request)
        skill_store = get_skill_store(request)
        # user_id is unused by the "private" scope filter (the password is the
        # gate), but list_paginated requires it for its signature.
        user_id = _current_user_id(request)

        if body.page is not None:
            rows, total = await store.list_paginated(
                user_id=user_id,
                scope="private",
                search=body.search,
                tag_ids=body.tag_ids,
                page=body.page,
                page_size=body.page_size,
                sort=body.sort,
                owner=body.owner,
            )
            responses = await _build_list_responses(store, tag_store, user_id, rows, skill_store)
            return AgentsListResponse(agents=responses, total=total, page=body.page, page_size=body.page_size)

        records = await store.list_private()
        responses = await _build_list_responses(store, tag_store, user_id, records, skill_store)
        responses = _sort_agent_responses(responses, body.sort)
        return AgentsListResponse(agents=responses, total=len(responses), page=1, page_size=len(responses))
    except HTTPException:
        raise
    except Exception as e:
        logger.error("Failed to list private agents: %s", e, exc_info=True)
        raise HTTPException(status_code=500, detail=f"Failed to list private agents: {str(e)}") from e


@router.put(
    "/agents/{agent_id}/private",
    response_model=AgentResponse,
    summary="Move Agent Between Squares (admin)",
    description=(
        "Admin-only. Move any agent (built-in or user-owned) into or out of the "
        "password-gated private square. Owners can already toggle the flag on "
        "their own agents via the regular ``PUT /agents/{id}`` endpoint; this "
        "endpoint exists so admins can re-categorize other users' agents too."
    ),
)
async def set_agent_private(
    agent_id: str,
    body: AgentPrivateUpdateRequest,
    request: Request,
) -> AgentResponse:
    _require_agents_api_enabled()
    agent_id = _validate_storage_id(agent_id)

    if not _is_admin_user(request):
        raise HTTPException(status_code=403, detail="Only system admins can re-categorize other users' agents")

    try:
        store = get_agent_store(request)
        record = await store.get_any(agent_id)
        if record is None:
            raise HTTPException(status_code=404, detail=f"Agent '{agent_id}' not found")

        update: dict[str, Any] = {"is_private": bool(body.is_private)}
        # Mutual exclusion — see ``update_agent`` above. Moving an agent into
        # the private square also drops the public ``published`` flag so we
        # never end up with an agent that's "in both squares".
        if update["is_private"]:
            update["published"] = False
        if record.get("user_id") is None:
            updated = await store.update_builtin(agent_id, update)
        else:
            updated = await store.update(agent_id, str(record["user_id"]), update)
        if updated is None:
            raise HTTPException(status_code=404, detail=f"Agent '{agent_id}' not found")

        logger.info("Admin set agent '%s' is_private=%s", agent_id, body.is_private)
        return _record_to_response(updated, include_soul=True)
    except HTTPException:
        raise
    except Exception as e:
        logger.error("Failed to set is_private for agent '%s': %s", agent_id, e, exc_info=True)
        raise HTTPException(status_code=500, detail=f"Failed to update agent: {str(e)}") from e


class AgentFeaturedReorderRequest(BaseModel):
    """Request body for admin to bulk-reorder the homepage featured list."""

    agent_ids: list[str] = Field(
        default_factory=list,
        description=("Featured agent ids in the desired top→bottom display order. Ids not currently featured are silently skipped, so the client can send the current visible order without first reconciling with the server."),
    )


@router.put(
    "/agents/featured/order",
    status_code=204,
    summary="Reorder Featured Agents (admin)",
    description=("Admin-only. Bulk-assign ``featured_order`` to each id in the order received (``0`` is topmost). After this, the homepage selector renders agents in exactly the supplied order. Ids not already featured are ignored."),
)
async def reorder_featured_agents(
    body: AgentFeaturedReorderRequest,
    request: Request,
) -> None:
    _require_agents_api_enabled()
    if not _is_admin_user(request):
        raise HTTPException(status_code=403, detail="Only system admins can reorder the homepage selector")

    try:
        store = get_agent_store(request)
        for index, agent_id in enumerate(body.agent_ids):
            # ``set_featured`` is a no-op for unknown ids — it returns None and
            # we move on without raising, matching the docstring.
            await store.set_featured(agent_id, index)
        logger.info("Admin reordered featured agents (%d ids)", len(body.agent_ids))
    except HTTPException:
        raise
    except Exception as e:
        logger.error("Failed to reorder featured agents: %s", e, exc_info=True)
        raise HTTPException(status_code=500, detail=f"Failed to reorder featured agents: {str(e)}") from e


class AgentDirectorySyncResponse(BaseModel):
    scanned: int = Field(..., description="Number of on-disk agent directories scanned into the catalog")


@router.post(
    "/agents/sync-directory",
    response_model=AgentDirectorySyncResponse,
    summary="Rescan Agent Directory (admin)",
    description=(
        "Admin-only. Scan ``.deer-flow/agents/`` and upsert folder-backed agents "
        "into the catalog, including the skill/tool listing cache. This is the "
        "manual equivalent of the old hourly list-request sync."
    ),
)
async def sync_agent_directory(request: Request) -> AgentDirectorySyncResponse:
    _require_agents_api_enabled()
    if not _is_admin_user(request):
        raise HTTPException(status_code=403, detail="Only system admins can rescan the agent directory")

    store = get_agent_store(request)
    scanned = await _sync_legacy_agents(store)
    _mark_legacy_sync_fresh()
    try:
        from app.gateway.services import _mark_legacy_sync_fresh as _services_mark_sync_fresh

        _services_mark_sync_fresh()
    except Exception:
        logger.debug("Could not refresh the run-path legacy sync gate", exc_info=True)
    logger.info("Admin rescanned agent directory: scanned=%d", scanned)
    return AgentDirectorySyncResponse(scanned=scanned)


class AgentSkillRemovalEntry(BaseModel):
    agent_id: str = Field(..., description="Agent storage id whose config.yaml was rewritten")
    agent_name: str | None = Field(default=None, description="Display name from the DB, if any")
    user_id: str | None = Field(default=None, description="Owner user id (null = built-in)")
    removed_skills: list[str] = Field(..., description="Skill names that no longer exist on disk and were stripped")


class AgentSkillsSyncResponse(BaseModel):
    checked: int = Field(..., description="Number of agent config.yaml files scanned")
    modified: int = Field(..., description="Number of agents whose skills list was rewritten")
    removals: list[AgentSkillRemovalEntry] = Field(default_factory=list, description="Per-agent removal details")


@router.post(
    "/agents/admin/sync-skills",
    response_model=AgentSkillsSyncResponse,
    summary="Sync Agent Skills (admin)",
    description=(
        "Admin-only. Scan every agent's ``config.yaml`` and strip skill names that "
        "no longer exist on disk. Repairs historical drift where a skill was "
        "removed without the delete/unpublish cascade firing (e.g. deleted from "
        "the filesystem manually, or removed in an older version of the code)."
    ),
)
async def admin_sync_agent_skills(
    request: Request,
    config: AppConfig = Depends(get_config),
) -> AgentSkillsSyncResponse:
    _require_agents_api_enabled()
    if not _is_admin_user(request):
        raise HTTPException(status_code=403, detail="Only system admins can run agent-skills sync")

    try:
        storage = get_or_new_skill_storage(app_config=config)
        all_skills = storage.load_skills(enabled_only=False)
        available = {skill.name for skill in all_skills}

        result = await asyncio.to_thread(sync_agent_skills_to_available, available)

        # Enrich removals with display name + owner from the DB so the admin
        # UI can show "我的智能体 X 移除了技能 Y" instead of just the id.
        store = get_agent_store(request)
        enriched: list[AgentSkillRemovalEntry] = []
        for entry in result.get("removals", []):
            agent_id = entry["agent_id"]
            record = await store.get_any(agent_id)
            enriched.append(
                AgentSkillRemovalEntry(
                    agent_id=agent_id,
                    agent_name=(record.get("name") if record else None),
                    user_id=(record.get("user_id") if record else None),
                    removed_skills=entry["removed_skills"],
                )
            )

        if result.get("modified"):
            await refresh_skills_system_prompt_cache_async()

        logger.info(
            "Admin agent-skills sync: checked=%d modified=%d",
            result["checked"],
            result["modified"],
        )
        return AgentSkillsSyncResponse(
            checked=result["checked"],
            modified=result["modified"],
            removals=enriched,
        )
    except HTTPException:
        raise
    except Exception as e:
        logger.error("Failed to sync agent skills: %s", e, exc_info=True)
        raise HTTPException(status_code=500, detail=f"Failed to sync agent skills: {str(e)}") from e