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

# #region AssistantAdminRoutes [C:5] [TYPE Module] [SEMANTICS assistant, admin, route, audit, conversation]
# @defgroup AssistantApi Module group.
# @BRIEF 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

from datetime import datetime
from typing import Any

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

from src.core.database import get_db
from src.core.logger import belief_scope
from src.dependencies import get_current_user, has_permission
from src.models.assistant import (
    AssistantAuditRecord,
)
from src.schemas.auth import User

from ._history import (
    _cleanup_history_ttl,
    _coerce_query_bool,
    _is_conversation_archived,
    _resolve_or_create_conversation,
)
from ._routes import router
from ._schemas import (
    ASSISTANT_AUDIT,
)


# #region list_conversations [C:2] [TYPE Function]
# @ingroup AssistantApi
# @BRIEF DEPRECATED — replaced by AgentChat.Api.ListConversations.
#         Return empty list. Kept for import compatibility.
# @DEPRECATED Replaced by AgentChat.Api.ListConversations
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: str | None = Query(None),
    current_user: User = Depends(get_current_user),
    db: Session = Depends(get_db),
):
    """DEPRECATED — use AgentChat.Api.ListConversations instead."""
    return {"items": [], "total": 0, "page": page, "page_size": page_size, "has_next": False, "active_total": 0, "archived_total": 0}
# #endregion list_conversations









@router.get("/audit")
# #region get_assistant_audit [TYPE Function]
# @ingroup AssistantApi
# @BRIEF 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.
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),
        }


# #endregion get_assistant_audit


# #endregion AssistantAdminRoutes