Spaces:
Running
Running
File size: 2,197 Bytes
6a2c89b e0f8f74 6a2c89b e0f8f74 6a2c89b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 | from __future__ import annotations
import logging
from typing import Optional
from fastapi import APIRouter, BackgroundTasks, HTTPException, Request
from pydantic import BaseModel
from app.deps import limiter
from app.services.supabase import get_supabase
from app.services.audit_engine import run_audit
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/audit", tags=["audit"])
_SESSION_EVENT_COUNTS: dict[str, int] = {}
_AUDIT_INTERVAL = 5
_MAX_SESSIONS = 1000 # cap to prevent unbounded memory growth
class AuditEventIn(BaseModel):
session_id: str
user_id: Optional[str] = None
step: str
tool: str
status: str
input_summary: str = ""
output_summary: str = ""
duration_ms: int = 0
metadata: Optional[dict] = None
timestamp: Optional[str] = None
def _should_trigger_audit(session_id: str) -> bool:
count = _SESSION_EVENT_COUNTS.get(session_id, 0) + 1
_SESSION_EVENT_COUNTS[session_id] = count
if len(_SESSION_EVENT_COUNTS) > _MAX_SESSIONS:
oldest = list(_SESSION_EVENT_COUNTS.keys())[:_MAX_SESSIONS // 2]
for k in oldest:
_SESSION_EVENT_COUNTS.pop(k, None)
return count % _AUDIT_INTERVAL == 0
@router.post("/event")
@limiter.exempt
async def receive_event(event: AuditEventIn, request: Request, background: BackgroundTasks):
sb = get_supabase()
try:
sb.table("audit_events").insert(event.model_dump(exclude_none=True)).execute()
except Exception as e:
logger.warning(f"Failed to store audit event: {e}")
should_audit = event.status == "failed" or _should_trigger_audit(event.session_id)
if should_audit:
background.add_task(run_audit, event.session_id, event.step)
return {"ok": True}
@router.get("/insights")
@limiter.exempt
async def get_insights(session: str):
if not session:
raise HTTPException(400, detail="session query parameter is required")
sb = get_supabase()
resp = sb.table("audit_insights") \
.select("*") \
.eq("session_id", session) \
.order("created_at", desc=True) \
.limit(1) \
.execute()
return {"latest": resp.data[0] if resp.data else None}
|