agent-platform/agent_platform/audit.py

"""Audit-Helper: strukturiertes Logging aller Aktionen."""
import logging
import json
from datetime import datetime
from typing import Optional, Dict, Any
import aiosqlite

from config import settings
from db import log_audit


logger = logging.getLogger("audit")


async def record(
    db: aiosqlite.Connection,
    agent_id: Optional[str],
    user_id: str,
    action: str,
    target: Optional[str] = None,
    args: Optional[Dict[str, Any]] = None,
    result: Optional[Dict[str, Any]] = None,
    duration_ms: Optional[int] = None,
) -> None:
    """Schreibt einen Audit-Eintrag in DB + Logger."""
    await log_audit(
        db=db,
        agent_id=agent_id,
        user_id=user_id,
        action=action,
        target=target,
        args=args,
        result=result,
        duration_ms=duration_ms,
    )

    log_entry = {
        "ts": datetime.utcnow().isoformat(),
        "agent_id": agent_id,
        "user_id": user_id,
        "action": action,
        "target": target,
        "args": args,
        "result": result,
        "duration_ms": duration_ms,
    }
    logger.info(json.dumps(log_entry, default=str))


async def get_recent(db: aiosqlite.Connection, limit: int = 100, agent_id: Optional[str] = None):
    """Holt die letzten Audit-Einträge."""
    from db import list_audit
    return await list_audit(db, agent_id=agent_id, limit=limit)