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, Request
from pydantic import BaseModel, Field

from app.gateway.deps import get_agent_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.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


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


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 _record_to_response(record: dict[str, Any], *, include_soul: bool = False) -> AgentResponse:
    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"],
    )


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) -> AgentsListResponse:
    _require_agents_api_enabled()

    try:
        store = get_agent_store(request)
        await _sync_legacy_agents(store)
        user_id = _current_user_id(request)
        agents = await store.list_visible(user_id)
        return AgentsListResponse(agents=[_record_to_response(a, include_soul=True) for a in agents])
    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/{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)
        record = await store.get_visible(agent_id, _current_user_id(request))
        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)
        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