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

"""Admin-only window into another user's activity.

Lets an administrator inspect any user's conversation history (threads + the
Q&A messages inside them) and their accumulated memory (global + per-agent
buckets). All endpoints require ``system_role == "admin"``; they are strictly
read-only.

The underlying stores normally auto-scope every query to the *calling* user
via a contextvar. These endpoints pass an explicit ``user_id`` so the admin
reads the *target* user's data instead — which is exactly what those store
methods' ``user_id`` parameter is designed for.
"""

from __future__ import annotations

import logging

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

from app.gateway.deps import (
    get_agent_store,
    get_checkpointer,
    get_current_user_from_request,
    get_thread_store,
)
from deerflow.agents.memory import BuiltinFileProvider
from deerflow.config.paths import get_paths
from deerflow.runtime import serialize_channel_values
from app.gateway.routers.memory import _build_v1_provider, _is_v1

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


async def _require_admin(request: Request) -> None:
    """Reject non-admin callers. In no-auth mode there is no user → allow."""
    try:
        user = await get_current_user_from_request(request)
    except HTTPException:
        raise
    if user is None:
        return
    if getattr(user, "system_role", None) != "admin":
        raise HTTPException(status_code=403, detail="Admin only")


# ── Threads ────────────────────────────────────────────────────────────────


class AdminThreadItem(BaseModel):
    thread_id: str
    display_name: str = ""
    status: str = "idle"
    created_at: str = ""
    updated_at: str = ""


@router.get("/{user_id}/threads", response_model=list[AdminThreadItem])
async def list_user_threads(
    user_id: str,
    request: Request,
    limit: int = Query(default=100, ge=1, le=500),
    offset: int = Query(default=0, ge=0),
) -> list[AdminThreadItem]:
    """List one user's conversation threads, newest first."""
    await _require_admin(request)
    repo = get_thread_store(request)
    rows = await repo.search(limit=limit, offset=offset, user_id=user_id)
    return [
        AdminThreadItem(
            thread_id=r["thread_id"],
            display_name=r.get("display_name") or "",
            status=r.get("status", "idle"),
            created_at=str(r.get("created_at") or ""),
            updated_at=str(r.get("updated_at") or ""),
        )
        for r in rows
    ]


@router.get("/{user_id}/threads/{thread_id}/messages", response_model=list[dict])
async def list_user_thread_messages(
    user_id: str,
    thread_id: str,
    request: Request,
    limit: int = Query(default=400, ge=1, le=1000),
) -> list[dict]:
    """Return the conversation messages of one of the user's threads.

    Read straight from the LangGraph checkpoint — the same source the chat UI
    renders — rather than the optional ``run_events`` audit log, which may be
    unconfigured (``run_events.backend=memory``) or empty for older threads.
    """
    await _require_admin(request)

    repo = get_thread_store(request)
    # ``user_id=None`` bypasses the owner filter so the admin can fetch any row;
    # we then assert the thread really belongs to the target user.
    thread = await repo.get(thread_id, user_id=None)
    if thread is None:
        raise HTTPException(status_code=404, detail="Thread not found")
    owner = thread.get("user_id")
    if owner is not None and owner != user_id:
        raise HTTPException(status_code=404, detail="Thread does not belong to this user")

    checkpointer = get_checkpointer(request)
    config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
    try:
        checkpoint_tuple = await checkpointer.aget_tuple(config)
    except Exception as exc:  # noqa: BLE001 - surface a clean error
        logger.error("Failed to read checkpoint for thread %s: %s", thread_id, exc, exc_info=True)
        raise HTTPException(status_code=500, detail="Failed to read thread state") from exc
    if checkpoint_tuple is None:
        return []

    checkpoint = getattr(checkpoint_tuple, "checkpoint", {}) or {}
    channel_values = checkpoint.get("channel_values", {}) if isinstance(checkpoint, dict) else {}
    values = serialize_channel_values(channel_values)
    messages = values.get("messages", [])
    if not isinstance(messages, list):
        return []
    # Keep the most recent ``limit`` messages — long threads stay bounded.
    return messages[-limit:]


# ── Memory ─────────────────────────────────────────────────────────────────


class AdminMemoryBucket(BaseModel):
    name: str = Field(..., description="Bucket name: 'global' or the agent id (used for CRUD)")
    display_name: str = Field("", description="Human-readable name shown in the UI")
    data: dict = Field(default_factory=dict, description="Raw memory data for the bucket")


class AdminMemoryEntryAddRequest(BaseModel):
    content: str = Field(..., min_length=1, description="Entry content to add")
    agent: str | None = Field(None, description="Agent name; None = global USER.md bucket")


class AdminMemoryEntryReplaceRequest(BaseModel):
    old_text: str = Field(..., min_length=1, description="Short unique substring to locate the entry")
    content: str = Field(..., min_length=1, description="New content to replace with")
    agent: str | None = Field(None, description="Agent name; None = global USER.md bucket")


class AdminMemoryResponse(BaseModel):
    user_id: str
    version: str = Field(default="v2", description="记忆版本：v1 或 v2")
    buckets: list[AdminMemoryBucket] = Field(default_factory=list)


def _read_memory_bucket(user_id: str, agent_id: str | None) -> dict:
    """Read one Memory V2 file into a JSON-able ``{"entries": [...]}`` dict."""
    provider = BuiltinFileProvider()
    provider.initialize(user_id=user_id, agent_id=agent_id)
    target = "user" if agent_id is None else "memory"
    return {"entries": provider.read(target)}


def _read_v1_memory_bucket(user_id: str, agent_id: str | None) -> dict:
    """Read one V1 memory.json into its raw dict (facts / user / history)."""
    provider = _build_v1_provider(user_id, agent_id)
    return provider.read_v1_data()


@router.get("/{user_id}/memory", response_model=AdminMemoryResponse)
async def get_user_memory(user_id: str, request: Request) -> AdminMemoryResponse:
    """Return every memory bucket for one user (V1 or V2 depending on config)."""
    await _require_admin(request)

    # Source agents from the database: only agents owned by this specific user.
    agent_store = get_agent_store(request)
    all_agents = await agent_store.list_visible(user_id)
    owned_agents = sorted(
        [a for a in all_agents if a.get("user_id") == user_id],
        key=lambda a: (a.get("name") or a["id"]).lower(),
    )

    if _is_v1(user_id):
        # V1 模式：每个桶读 memory.json，内容为 {facts, user, history}
        buckets: list[AdminMemoryBucket] = []
        try:
            buckets.append(AdminMemoryBucket(name="default", display_name="默认助手", data=_read_v1_memory_bucket(user_id, None)))
        except Exception:
            logger.exception("Failed to read V1 global memory for user %s", user_id)
            buckets.append(AdminMemoryBucket(name="default", display_name="默认助手", data={}))

        for agent in owned_agents:
            agent_id: str = agent["id"]
            display_name: str = agent.get("name") or agent_id
            try:
                data = _read_v1_memory_bucket(user_id, agent_id)
            except Exception:
                data = {}
            buckets.append(AdminMemoryBucket(name=agent_id, display_name=display_name, data=data))

        return AdminMemoryResponse(user_id=user_id, version="v1", buckets=buckets)

    # V2 模式：读 USER.md / MEMORY.md，内容为 {entries: [...]}
    buckets = [
        AdminMemoryBucket(name="default", display_name="默认助手", data=_read_memory_bucket(user_id, None))
    ]
    for agent in owned_agents:
        agent_id = agent["id"]
        display_name = agent.get("name") or agent_id
        try:
            data = _read_memory_bucket(user_id, agent_id)
        except Exception:
            data = {"entries": []}
        buckets.append(AdminMemoryBucket(name=agent_id, display_name=display_name, data=data))

    return AdminMemoryResponse(user_id=user_id, version="v2", buckets=buckets)


@router.post("/{user_id}/memory/entries", response_model=AdminMemoryBucket)
async def admin_add_memory_entry(
    user_id: str,
    body: AdminMemoryEntryAddRequest,
    request: Request,
) -> AdminMemoryBucket:
    """Add a memory entry for a user (admin only)."""
    await _require_admin(request)
    target = "user" if body.agent is None else "memory"
    provider = BuiltinFileProvider()
    provider.initialize(user_id=user_id, agent_id=body.agent)
    try:
        result = provider.add(target, body.content)
        if result is False:
            raise HTTPException(status_code=400, detail="Failed to add entry (quota exceeded or write error)")
    except HTTPException:
        raise
    except Exception as exc:
        logger.error("admin_add_memory_entry failed for user %s: %s", user_id, exc, exc_info=True)
        raise HTTPException(status_code=500, detail="Failed to write memory entry") from exc
    bucket_name = "global" if body.agent is None else body.agent
    return AdminMemoryBucket(name=bucket_name, data=_read_memory_bucket(user_id, body.agent))


@router.put("/{user_id}/memory/entries", response_model=AdminMemoryBucket)
async def admin_replace_memory_entry(
    user_id: str,
    body: AdminMemoryEntryReplaceRequest,
    request: Request,
) -> AdminMemoryBucket:
    """Replace a memory entry for a user (admin only)."""
    await _require_admin(request)
    target = "user" if body.agent is None else "memory"
    provider = BuiltinFileProvider()
    provider.initialize(user_id=user_id, agent_id=body.agent)
    try:
        result = provider.replace(target, body.old_text, body.content)
        if result is False:
            raise HTTPException(status_code=404, detail="Entry not found or write failed")
    except HTTPException:
        raise
    except Exception as exc:
        logger.error("admin_replace_memory_entry failed for user %s: %s", user_id, exc, exc_info=True)
        raise HTTPException(status_code=500, detail="Failed to replace memory entry") from exc
    bucket_name = "global" if body.agent is None else body.agent
    return AdminMemoryBucket(name=bucket_name, data=_read_memory_bucket(user_id, body.agent))


@router.delete("/{user_id}/memory/entries", response_model=AdminMemoryBucket)
async def admin_delete_memory_entry(
    user_id: str,
    request: Request,
    old_text: str = Query(..., min_length=1, description="Short unique substring to locate the entry"),
    agent: str | None = Query(None, description="Agent name; None = global USER.md bucket"),
) -> AdminMemoryBucket:
    """Delete a memory entry for a user (admin only)."""
    await _require_admin(request)
    target = "user" if agent is None else "memory"
    provider = BuiltinFileProvider()
    provider.initialize(user_id=user_id, agent_id=agent)
    try:
        result = provider.remove(target, old_text)
        if result is False:
            raise HTTPException(status_code=404, detail="Entry not found")
    except HTTPException:
        raise
    except Exception as exc:
        logger.error("admin_delete_memory_entry failed for user %s: %s", user_id, exc, exc_info=True)
        raise HTTPException(status_code=500, detail="Failed to delete memory entry") from exc
    bucket_name = "global" if agent is None else agent
    return AdminMemoryBucket(name=bucket_name, data=_read_memory_bucket(user_id, agent))


# ── V1 Memory CRUD (admin) ─────────────────────────────────────────────────

_V1_USER_FIELDS = {"workContext", "personalContext", "topOfMind"}
_V1_HISTORY_FIELDS = {"recentMonths", "earlierContext", "longTermBackground"}


class AdminV1SectionUpdateRequest(BaseModel):
    section_type: str = Field(..., description="'user' 或 'history'")
    field: str = Field(..., description="字段名，如 workContext / recentMonths")
    summary: str = Field(..., description="新的摘要文本")
    agent: str | None = Field(None, description="Agent id；None 表示用户级全局记忆")


class AdminV1FactAddRequest(BaseModel):
    content: str = Field(..., min_length=1)
    category: str = Field(default="context")
    confidence: float = Field(default=0.8, ge=0.0, le=1.0)
    agent: str | None = Field(None)


class AdminV1FactUpdateRequest(BaseModel):
    content: str | None = Field(None)
    category: str | None = Field(None)
    confidence: float | None = Field(None, ge=0.0, le=1.0)
    agent: str | None = Field(None)


@router.put("/{user_id}/memory/v1/sections", response_model=AdminMemoryBucket)
async def admin_update_v1_section(
    user_id: str,
    body: AdminV1SectionUpdateRequest,
    request: Request,
) -> AdminMemoryBucket:
    """更新 V1 记忆的某个摘要字段（用户画像 / 历史）。"""
    await _require_admin(request)
    if body.section_type == "user" and body.field not in _V1_USER_FIELDS:
        raise HTTPException(status_code=400, detail=f"无效 user 字段：{body.field}")
    if body.section_type == "history" and body.field not in _V1_HISTORY_FIELDS:
        raise HTTPException(status_code=400, detail=f"无效 history 字段：{body.field}")
    if body.section_type not in ("user", "history"):
        raise HTTPException(status_code=400, detail="section_type 须为 'user' 或 'history'")

    from deerflow.agents.memory.v1.storage import get_memory_storage, utc_now_iso_z
    storage = get_memory_storage()
    data = storage.load(body.agent, user_id=user_id)
    data.setdefault(body.section_type, {})[body.field] = {
        "summary": body.summary,
        "updatedAt": utc_now_iso_z(),
    }
    if not storage.save(data, body.agent, user_id=user_id):
        raise HTTPException(status_code=500, detail="写入失败")

    bucket_name = "global" if body.agent is None else body.agent
    return AdminMemoryBucket(name=bucket_name, data=_read_v1_memory_bucket(user_id, body.agent))


@router.post("/{user_id}/memory/v1/facts", response_model=AdminMemoryBucket)
async def admin_add_v1_fact(
    user_id: str,
    body: AdminV1FactAddRequest,
    request: Request,
) -> AdminMemoryBucket:
    """新增一条 V1 事实记录。"""
    await _require_admin(request)
    try:
        from deerflow.agents.memory.v1.updater import create_memory_fact
        create_memory_fact(
            body.content,
            category=body.category,
            confidence=body.confidence,
            agent_name=body.agent,
            user_id=user_id,
        )
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc
    except Exception as exc:
        raise HTTPException(status_code=500, detail="新增失败") from exc
    bucket_name = "global" if body.agent is None else body.agent
    return AdminMemoryBucket(name=bucket_name, data=_read_v1_memory_bucket(user_id, body.agent))


@router.put("/{user_id}/memory/v1/facts/{fact_id}", response_model=AdminMemoryBucket)
async def admin_update_v1_fact(
    user_id: str,
    fact_id: str,
    body: AdminV1FactUpdateRequest,
    request: Request,
) -> AdminMemoryBucket:
    """更新一条 V1 事实记录（内容 / 类别 / 置信度）。"""
    await _require_admin(request)
    try:
        from deerflow.agents.memory.v1.updater import update_memory_fact
        update_memory_fact(
            fact_id,
            content=body.content,
            category=body.category,
            confidence=body.confidence,
            agent_name=body.agent,
            user_id=user_id,
        )
    except KeyError:
        raise HTTPException(status_code=404, detail=f"事实 '{fact_id}' 不存在")
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc
    except Exception as exc:
        raise HTTPException(status_code=500, detail="更新失败") from exc
    bucket_name = "global" if body.agent is None else body.agent
    return AdminMemoryBucket(name=bucket_name, data=_read_v1_memory_bucket(user_id, body.agent))


@router.delete("/{user_id}/memory/v1/facts/{fact_id}", response_model=AdminMemoryBucket)
async def admin_delete_v1_fact(
    user_id: str,
    fact_id: str,
    request: Request,
    agent: str | None = Query(None, description="Agent id；None 表示用户级全局记忆"),
) -> AdminMemoryBucket:
    """删除一条 V1 事实记录。"""
    await _require_admin(request)
    try:
        from deerflow.agents.memory.v1.updater import delete_memory_fact
        delete_memory_fact(fact_id, agent_name=agent, user_id=user_id)
    except KeyError:
        raise HTTPException(status_code=404, detail=f"事实 '{fact_id}' 不存在")
    except Exception as exc:
        raise HTTPException(status_code=500, detail="删除失败") from exc
    bucket_name = "global" if agent is None else agent
    return AdminMemoryBucket(name=bucket_name, data=_read_v1_memory_bucket(user_id, agent))