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_checkpointer,
    get_current_user_from_request,
    get_thread_store,
)
from deerflow.agents.memory import BuiltinFileProvider
from deerflow.config.agents_config import AGENT_NAME_PATTERN
from deerflow.config.paths import get_paths
from deerflow.runtime import serialize_channel_values

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 a custom-agent name")
    data: dict = Field(default_factory=dict, description="Raw memory data for the bucket")


class AdminMemoryResponse(BaseModel):
    user_id: str
    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.

    ``agent_id is None`` -> the user's ``USER.md`` (profile, cross-agent);
    otherwise that agent's ``MEMORY.md`` (per-agent working notes).
    """
    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)}


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

    Memory V2 layout: the ``global`` bucket is ``USER.md`` (user profile);
    each per-agent bucket is that agent's ``MEMORY.md`` working notes.
    """
    await _require_admin(request)

    buckets: list[AdminMemoryBucket] = [
        AdminMemoryBucket(name="global", data=_read_memory_bucket(user_id, None))
    ]

    agents_root = get_paths().user_dir(user_id) / "agents"
    if agents_root.is_dir():
        for entry in sorted(agents_root.iterdir(), key=lambda p: p.name.lower()):
            if not entry.is_dir() or not (entry / "MEMORY.md").exists():
                continue
            if not AGENT_NAME_PATTERN.match(entry.name):
                continue
            try:
                data = _read_memory_bucket(user_id, entry.name)
            except Exception:
                logger.warning("Failed to read memory bucket %s for user", entry.name)
                continue
            buckets.append(AdminMemoryBucket(name=entry.name, data=data))

    return AdminMemoryResponse(user_id=user_id, buckets=buckets)