ss-tools/backend/src/api/routes/assistant/_admin_routes.py

# [DEF:AssistantAdminRoutes:Module]
# @COMPLEXITY: 5
# @PURPOSE: FastAPI route handlers for assistant admin operations — conversation listing, deletion, history, audit.
# @LAYER: API
# @RELATION: DEPENDS_ON -> [AssistantRoutes]
# @RELATION: DEPENDS_ON -> [AssistantSchemas]
# @RELATION: DEPENDS_ON -> [AssistantHistory]
# @INVARIANT: Audit endpoint requires tasks:READ permission.

from __future__ import annotations

import uuid
from datetime import datetime
from typing import Any, Dict, Optional

from fastapi import Depends, HTTPException, Query
from sqlalchemy.orm import Session
from sqlalchemy import desc

from src.core.logger import belief_scope, logger
from src.dependencies import get_current_user, has_permission
from src.core.database import get_db
from src.models.assistant import (
    AssistantAuditRecord,
    AssistantMessageRecord,
)
from src.schemas.auth import User
from ._schemas import (
    ASSISTANT_AUDIT,
    CONVERSATIONS,
    AssistantAction,
    AssistantMessageResponse,
)
from ._history import (
    _cleanup_history_ttl,
    _coerce_query_bool,
    _is_conversation_archived,
    _persist_audit,
    _resolve_or_create_conversation,
)
from ._routes import router


# [DEF:list_conversations:Function]
# @COMPLEXITY: 2
# @PURPOSE: Return paginated conversation list for current user with archived flag and last message preview.
# @PRE: Authenticated user context and valid pagination params.
# @POST: Conversations are grouped by conversation_id sorted by latest activity descending.
# @RETURN: Dict with items, paging metadata, and archive segmentation counts.
@router.get("/conversations")
async def list_conversations(
    page: int = Query(1, ge=1),
    page_size: int = Query(20, ge=1, le=100),
    include_archived: bool = Query(False),
    archived_only: bool = Query(False),
    search: Optional[str] = Query(None),
    current_user: User = Depends(get_current_user),
    db: Session = Depends(get_db),
):
    with belief_scope("assistant.conversations"):
        user_id = current_user.id
        include_archived = _coerce_query_bool(include_archived)
        archived_only = _coerce_query_bool(archived_only)
        _cleanup_history_ttl(db, user_id)

        rows = (
            db.query(AssistantMessageRecord)
            .filter(AssistantMessageRecord.user_id == user_id)
            .order_by(desc(AssistantMessageRecord.created_at))
            .all()
        )

        summary: Dict[str, Dict[str, Any]] = {}
        for row in rows:
            conv_id = row.conversation_id
            if not conv_id:
                continue
            created_at = row.created_at or datetime.utcnow()
            if conv_id not in summary:
                summary[conv_id] = {
                    "conversation_id": conv_id,
                    "title": "",
                    "updated_at": created_at,
                    "last_message": row.text,
                    "last_role": row.role,
                    "last_state": row.state,
                    "last_task_id": row.task_id,
                    "message_count": 0,
                }
            item = summary[conv_id]
            item["message_count"] += 1
            if row.role == "user" and row.text and not item["title"]:
                item["title"] = row.text.strip()[:80]

        items = []
        search_term = search.lower().strip() if search else ""
        archived_total = sum(
            1
            for c in summary.values()
            if _is_conversation_archived(c.get("updated_at"))
        )
        active_total = len(summary) - archived_total
        for conv in summary.values():
            conv["archived"] = _is_conversation_archived(conv.get("updated_at"))
            if not conv.get("title"):
                conv["title"] = f"Conversation {conv['conversation_id'][:8]}"
            if search_term:
                haystack = (
                    f"{conv.get('title', '')} {conv.get('last_message', '')}".lower()
                )
                if search_term not in haystack:
                    continue
            if archived_only and not conv["archived"]:
                continue
            if not archived_only and not include_archived and conv["archived"]:
                continue
            updated = conv.get("updated_at")
            conv["updated_at"] = (
                updated.isoformat() if isinstance(updated, datetime) else None
            )
            items.append(conv)

        items.sort(key=lambda x: x.get("updated_at") or "", reverse=True)
        total = len(items)
        start = (page - 1) * page_size
        page_items = items[start : start + page_size]

        return {
            "items": page_items,
            "total": total,
            "page": page,
            "page_size": page_size,
            "has_next": start + page_size < total,
            "active_total": active_total,
            "archived_total": archived_total,
        }


# [/DEF:list_conversations:Function]


# [DEF:delete_conversation:Function]
# @COMPLEXITY: 2
# @PURPOSE: Soft-delete or hard-delete a conversation and clear its in-memory trace.
# @PRE: conversation_id belongs to current_user.
# @POST: Conversation records are removed from DB and CONVERSATIONS cache.
@router.delete("/conversations/{conversation_id}")
async def delete_conversation(
    conversation_id: str,
    current_user: User = Depends(get_current_user),
    db: Session = Depends(get_db),
):
    with belief_scope("assistant.conversations.delete"):
        user_id = current_user.id

        # 1. Remove from in-memory cache
        key = (user_id, conversation_id)
        if key in CONVERSATIONS:
            del CONVERSATIONS[key]

        # 2. Delete from database
        deleted_count = (
            db.query(AssistantMessageRecord)
            .filter(
                AssistantMessageRecord.user_id == user_id,
                AssistantMessageRecord.conversation_id == conversation_id,
            )
            .delete()
        )

        db.commit()

        if deleted_count == 0:
            raise HTTPException(
                status_code=404, detail="Conversation not found or already deleted"
            )

        return {
            "status": "success",
            "deleted": deleted_count,
            "conversation_id": conversation_id,
        }


# [/DEF:delete_conversation:Function]


@router.get("/history")
# [DEF:get_history:Function]
# @PURPOSE: Retrieve paginated assistant conversation history for current user.
# @PRE: Authenticated user is available and page params are valid.
# @POST: Returns persistent messages and mirrored in-memory snapshot for diagnostics.
# @RETURN: Dict with items, paging metadata, and resolved conversation_id.
async def get_history(
    page: int = Query(1, ge=1),
    page_size: int = Query(20, ge=1, le=100),
    conversation_id: Optional[str] = Query(None),
    from_latest: bool = Query(False),
    current_user: User = Depends(get_current_user),
    db: Session = Depends(get_db),
):
    with belief_scope("assistant.history"):
        user_id = current_user.id
        _cleanup_history_ttl(db, user_id)
        conv_id = _resolve_or_create_conversation(user_id, conversation_id, db)

        base_query = db.query(AssistantMessageRecord).filter(
            AssistantMessageRecord.user_id == user_id,
            AssistantMessageRecord.conversation_id == conv_id,
        )
        total = base_query.count()
        start = (page - 1) * page_size
        if from_latest:
            rows = (
                base_query.order_by(desc(AssistantMessageRecord.created_at))
                .offset(start)
                .limit(page_size)
                .all()
            )
            rows = list(reversed(rows))
        else:
            rows = (
                base_query.order_by(AssistantMessageRecord.created_at.asc())
                .offset(start)
                .limit(page_size)
                .all()
            )

        persistent_items = [
            {
                "message_id": row.id,
                "conversation_id": row.conversation_id,
                "role": row.role,
                "text": row.text,
                "state": row.state,
                "task_id": row.task_id,
                "confirmation_id": row.confirmation_id,
                "created_at": row.created_at.isoformat() if row.created_at else None,
                "metadata": row.payload,
            }
            for row in rows
        ]

        memory_items = CONVERSATIONS.get((user_id, conv_id), [])
        return {
            "items": persistent_items,
            "memory_items": memory_items,
            "total": total,
            "page": page,
            "page_size": page_size,
            "has_next": start + page_size < total,
            "from_latest": from_latest,
            "conversation_id": conv_id,
        }


# [/DEF:get_history:Function]


@router.get("/audit")
# [DEF:get_assistant_audit:Function]
# @PURPOSE: Return assistant audit decisions for current user from persistent and in-memory stores.
# @PRE: User has tasks:READ permission.
# @POST: Audit payload is returned in reverse chronological order from DB.
# @RETURN: Dict with persistent and memory audit slices.
async def get_assistant_audit(
    limit: int = Query(50, ge=1, le=500),
    current_user: User = Depends(get_current_user),
    db: Session = Depends(get_db),
    _=Depends(has_permission("tasks", "READ")),
):
    with belief_scope("assistant.audit"):
        memory_rows = ASSISTANT_AUDIT.get(current_user.id, [])
        db_rows = (
            db.query(AssistantAuditRecord)
            .filter(AssistantAuditRecord.user_id == current_user.id)
            .order_by(AssistantAuditRecord.created_at.desc())
            .limit(limit)
            .all()
        )
        persistent = [
            {
                "id": row.id,
                "user_id": row.user_id,
                "conversation_id": row.conversation_id,
                "decision": row.decision,
                "task_id": row.task_id,
                "message": row.message,
                "payload": row.payload,
                "created_at": row.created_at.isoformat() if row.created_at else None,
            }
            for row in db_rows
        ]
        return {
            "items": persistent,
            "memory_items": memory_rows[-limit:],
            "total": len(persistent),
            "memory_total": len(memory_rows),
        }


# [/DEF:get_assistant_audit:Function]


# [/DEF:AssistantAdminRoutes:Module]