"""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 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 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. Agent list is sourced from the database (all agents owned by this user) so agents without any existing MEMORY.md still appear. """ await _require_admin(request) buckets: list[AdminMemoryBucket] = [ AdminMemoryBucket(name="global", display_name="用户画像", data=_read_memory_bucket(user_id, None)) ] # 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(), ) for agent in owned_agents: agent_id: str = agent["id"] display_name: str = 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, 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))