💰 Camada 5b — Governança: Custo e Audit

class CostAlertSystem:
    """Sistema de alertas de custo multicamada."""

    def __init__(self, slack_webhook: str, email_recipients: list[str]):
        self.slack_webhook = slack_webhook
        self.emails = email_recipients

        # Thresholds configuráveis
        self.thresholds = {
            "per_execution_warning_usd": 4.0,    # 80% de budget $5
            "per_execution_critical_usd": 5.0,   # budget máximo
            "per_hour_usd": 20.0,                # custo/hora máximo
            "per_day_usd": 100.0,                # custo/dia máximo
            "anomaly_multiplier": 2.5,           # X vezes o baseline histórico
        }

        self._hourly_cost = 0.0
        self._daily_cost = 0.0
        self._baseline_hourly = None  # calculado do histórico

    async def check_and_alert(self, cost_event: dict):
        """Verifica todas as camadas e dispara alertas se necessário."""
        cost = cost_event["call_cost"]
        accumulated = cost_event["accumulated"]

        self._hourly_cost += cost
        self._daily_cost += cost

        alerts = []

        # Camada 1: por execução
        if accumulated >= self.thresholds["per_execution_critical_usd"]:
            alerts.append({
                "level": "CRITICAL",
                "message": f"Budget crítico: ${accumulated:.2f} atingido",
                "action": "Execução pausada automaticamente"
            })
        elif accumulated >= self.thresholds["per_execution_warning_usd"]:
            alerts.append({
                "level": "WARNING",
                "message": f"Execução em ${accumulated:.2f} (80% do budget)",
                "action": "Monitorar de perto"
            })

        # Camada 2: por hora
        if self._hourly_cost > self.thresholds["per_hour_usd"]:
            alerts.append({
                "level": "WARNING",
                "message": f"Custo horário: ${self._hourly_cost:.2f}/h (limite: ${self.thresholds['per_hour_usd']})"
            })

        # Anomaly detection simples
        if self._baseline_hourly and self._hourly_cost > self._baseline_hourly * self.thresholds["anomaly_multiplier"]:
            alerts.append({
                "level": "WARNING",
                "message": f"Anomalia: custo {self._hourly_cost/self._baseline_hourly:.1f}x acima do baseline"
            })

        for alert in alerts:
            await self._send_alert(alert, cost_event)

    async def _send_alert(self, alert: dict, context: dict):
        """Envia alerta para Slack."""
        emoji = "🚨" if alert["level"] == "CRITICAL" else "⚠️"
        message = {
            "text": f"{emoji} *Alerta de Custo de Agente*",
            "attachments": [{
                "color": "#FF0000" if alert["level"] == "CRITICAL" else "#FFA500",
                "fields": [
                    {"title": "Nível", "value": alert["level"], "short": True},
                    {"title": "Mensagem", "value": alert["message"], "short": False},
                    {"title": "Session", "value": context.get("session_id", "N/A"), "short": True},
                ]
            }]
        }
        async with aiohttp.ClientSession() as session:
            await session.post(self.slack_webhook, json=message)

Query SQL para cálculo de ROI

-- ROI por tipo de tarefa (baseado em custo humano configurado)
WITH human_cost_reference AS (
    SELECT
        task_type,
        human_hourly_rate_usd,
        estimated_human_hours
    FROM task_type_config  -- tabela de configuração com custo humano estimado
),
agent_cost AS (
    SELECT
        JSON_EXTRACT(input_data, '$.agente_tipo') AS task_type,
        AVG(cost_usd) AS avg_agent_cost,
        AVG(EXTRACT(EPOCH FROM (completed_at - started_at))) AS avg_duration_sec,
        COUNT(*) AS total_executions
    FROM task_executions
    WHERE status = 'done'
      AND started_at >= NOW() - INTERVAL '30 days'
    GROUP BY task_type
)
SELECT
    a.task_type,
    a.avg_agent_cost,
    a.avg_duration_sec / 60 AS avg_duration_min,
    (h.human_hourly_rate_usd * h.estimated_human_hours) AS human_cost_equivalent,
    (h.human_hourly_rate_usd * h.estimated_human_hours) / a.avg_agent_cost AS roi_multiplier,
    a.total_executions,
    a.total_executions * a.avg_agent_cost AS total_ai_cost,
    a.total_executions * (h.human_hourly_rate_usd * h.estimated_human_hours) AS total_human_equivalent
FROM agent_cost a
JOIN human_cost_reference h ON a.task_type = h.task_type
ORDER BY roi_multiplier DESC;