import uuid import logging from datetime import datetime from typing import List, Optional, Dict, Any from fastapi import APIRouter, Depends, HTTPException from sqlalchemy import select, desc from sqlalchemy.ext.asyncio import AsyncSession from app.database.postgres import get_db, AgentRun, Email, User from app.database.mongo import mongo_db from app.auth.security import get_current_user from app.tasks.email_tasks import fetch_emails_task from pydantic import BaseModel logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/agent", tags=["agent"]) class AgentRunListItem(BaseModel): id: str action_type: str status: str started_at: str completed_at: Optional[str] = None email_subject: Optional[str] = None email_id: Optional[str] = None class AgentLogItem(BaseModel): step: int thought: str action: str result: str timestamp: str @router.post("/trigger-sync") async def trigger_sync(current_user: User = Depends(get_current_user)): """Manually trigger the email synchronization task in the background.""" task = fetch_emails_task.delay() return {"status": "triggered", "task_id": task.id} @router.get("/runs", response_model=List[AgentRunListItem]) async def list_agent_runs( db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user) ): """Retrieve history of agent executions, joining email subjects where applicable.""" # Custom join to fetch email subject query = ( select(AgentRun, Email.subject) .outerjoin(Email, AgentRun.email_id == Email.id) .where(AgentRun.user_id == current_user.id) .order_by(desc(AgentRun.started_at)) .limit(30) ) result = await db.execute(query) rows = result.all() runs = [] for row in rows: run_obj, email_subj = row runs.append( AgentRunListItem( id=str(run_obj.id), action_type=run_obj.action_type, status=run_obj.status, started_at=run_obj.started_at.isoformat(), completed_at=run_obj.completed_at.isoformat() if run_obj.completed_at else None, email_subject=email_subj, email_id=str(run_obj.email_id) if run_obj.email_id else None ) ) return runs @router.get("/runs/{run_id}/logs", response_model=List[AgentLogItem]) async def get_run_logs( run_id: str, current_user: User = Depends(get_current_user) ): """Retrieve step-by-step logic and thoughts of the AI agent from MongoDB.""" # Ensure MongoDB client connection if mongo_db.db is None: mongo_db.connect() try: # Fetch logs for the run, ordered by step cursor = mongo_db.agent_logs.find({"run_id": run_id}).sort("step", 1) logs = await cursor.to_list(length=100) return [ AgentLogItem( step=log.get("step", 0), thought=log.get("thought", ""), action=log.get("action", ""), result=log.get("result", ""), timestamp=log.get("timestamp", datetime.utcnow()).isoformat() ) for log in logs ] except Exception as e: logger.error(f"Error fetching agent logs: {e}") raise HTTPException(status_code=500, detail="Failed to fetch execution logs")