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

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

from __future__ import annotations

import logging
import shutil
from datetime import datetime
from typing import Any
from uuid import uuid4

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

from app.gateway.deps import get_agent_store, get_tag_store
from deerflow.config.agents_api_config import get_agents_api_config
from deerflow.config.agents_config import list_custom_agents, load_agent_config, load_agent_soul, validate_agent_id
from deerflow.config.paths import get_paths
from deerflow.persistence.agents import AgentStore
from deerflow.persistence.tags import TagStore
from deerflow.runtime.user_context import get_effective_user_id

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


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.")
    published: bool = Field(default=False, description="Whether a user-created agent is public")
    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
    tool_groups: list[str] | None = None
    skills: list[str] | None = None

    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.",
    )


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

    agents: list[AgentResponse]


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

    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")
    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")
    tool_groups: list[str] | None = Field(default=None, description="Optional tool group whitelist persisted to config.yaml")


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")
    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)")
    tool_groups: list[str] | None = Field(default=None, description="Updated tool group whitelist (config.yaml)")


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 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"


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 _write_agent_config(
    agent_id: str,
    *,
    name: str,
    description: str,
    skills: list[str] | None = None,
    model: str | None = None,
    tool_groups: list[str] | 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
    if tool_groups is not None:
        config_data["tool_groups"] = tool_groups
    with open(config_file, "w", encoding="utf-8") as f:
        yaml.dump(config_data, f, default_flow_style=False, allow_unicode=True)


def _load_config_extras(agent_id: str) -> dict[str, Any]:
    """Read skills/model/tool_groups from the agent's config.yaml, tolerant of missing files."""
    try:
        cfg = load_agent_config(agent_id)
    except (FileNotFoundError, ValueError):
        return {"skills": None, "model": None, "tool_groups": None}
    if cfg is None:
        return {"skills": None, "model": None, "tool_groups": None}
    return {"skills": cfg.skills, "model": cfg.model, "tool_groups": cfg.tool_groups}


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)
    (agent_dir / "SOUL.md").write_text(soul, encoding="utf-8")


async def _sync_legacy_agents(store: AgentStore) -> None:
    """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.
    """
    try:
        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,
                }
            )
    except Exception:
        logger.exception("Failed to sync legacy folder-backed agents")


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,
) -> AgentResponse:
    _ensure_agent_dir_from_record(record)
    soul: str | None = None
    if include_soul:
        soul = load_agent_soul(record["id"]) or ""

    extras = _load_config_extras(record["id"])

    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)),
        builtin=record.get("user_id") is None,
        created_at=record.get("created_at"),
        updated_at=record.get("updated_at"),
        soul=soul,
        model=extras["model"],
        tool_groups=extras["tool_groups"],
        skills=extras["skills"],
        featured_order=record.get("featured_order"),
    )


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'.

    'admin' is granted only when the agent is built-in (user_id IS NULL) and the
    caller is a system admin. Custom agents owned by other users still 403.
    """
    owned = await store.get_owned(agent_id, user_id)
    if owned is not None:
        return owned, "owner"
    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")
    if is_admin and visible.get("user_id") is None:
        return visible, "admin"
    raise HTTPException(status_code=403, detail="You can only modify agents that you created")


@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"),
    tag_ids: list[str] | None = Query(default=None, description="Keep agents carrying any of these tags (OR)"),
) -> AgentsListResponse:
    _require_agents_api_enabled()

    try:
        store = get_agent_store(request)
        tag_store: TagStore = get_tag_store(request)
        await _sync_legacy_agents(store)
        user_id = _current_user_id(request)
        agents = await store.list_visible(user_id)
        responses = [_record_to_response(a, include_soul=True) for a in agents]

        assignments = await tag_store.list_assignments("agent", [r.id for r in responses])
        for resp in responses:
            resp.tags = assignments.get(resp.id, [])
        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)]
        return AgentsListResponse(agents=responses)
    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


@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)
        await _sync_legacy_agents(store)

        user_id = _current_user_id(request)
        records = await store.list_featured(user_id)
        return AgentSelectorResponse(agents=[_record_to_response(r) 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/{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)
        await _sync_legacy_agents(store)
        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")
        return _record_to_response(record, include_soul=True)
    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)
    agent_id = uuid4().hex
    agent_dir = get_paths().agent_dir(agent_id)

    try:
        store = get_agent_store(http_request)
        if agent_dir.exists():
            raise HTTPException(status_code=409, detail=f"Generated agent id '{agent_id}' already exists")

        _write_agent_config(
            agent_id,
            name=name,
            description=request.description or "",
            skills=request.skills,
            model=request.model,
            tool_groups=request.tool_groups,
        )
        _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,
            }
        )
        logger.info("Created agent '%s' (%s) at %s", name, agent_id, agent_dir)
        return _record_to_response(record, include_soul=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)

    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
        )

        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)

        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)
        next_skills = request.skills if "skills" in fields_set else existing_extras["skills"]
        next_model = request.model if "model" in fields_set else existing_extras["model"]
        next_tool_groups = request.tool_groups if "tool_groups" in fields_set else existing_extras["tool_groups"]

        config_dirty = bool(update_data) or any(key in fields_set for key in ("skills", "model", "tool_groups"))
        if config_dirty:
            _write_agent_config(
                agent_id,
                name=next_name,
                description=next_description,
                skills=next_skills,
                model=next_model,
                tool_groups=next_tool_groups,
            )
        if request.soul is not None:
            _write_agent_soul(agent_id, request.soul)

        if not update_data:
            record = current
        elif edit_mode == "admin":
            record = await store.update_builtin(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 _record_to_response(record, include_soul=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)

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

        if edit_mode == "admin":
            deleted = await store.delete_builtin(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
        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.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)
        await _sync_legacy_agents(store)
        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 _record_to_response(updated, include_soul=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 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