📟 Operação: o turno de plantão do seu OS

# app/routes/dashboard.py
from fastapi import APIRouter, Depends
from datetime import datetime, timedelta
import asyncpg

router = APIRouter(prefix="/dashboard")

@router.get("/stats")
async def dashboard_stats(db: asyncpg.Connection = Depends(get_db)):
    today = datetime.utcnow().date()
    yesterday = today - timedelta(days=1)

    row = await db.fetchrow("""
        SELECT
          (SELECT COUNT(DISTINCT user_id)
             FROM requests
            WHERE day = $1)                                    AS dau_today,
          (SELECT COUNT(DISTINCT user_id)
             FROM requests
            WHERE day = $2)                                    AS dau_yesterday,
          (SELECT COALESCE(SUM(cost_usd), 0)
             FROM usage_daily
            WHERE day = $1)                                    AS cost_today,
          (SELECT COUNT(*) FROM errors
            WHERE ts > NOW() - INTERVAL '24 hours')            AS errors_24h,
          (SELECT COUNT(*) FROM errors
            WHERE ts > NOW() - INTERVAL '24 hours'
              AND severity = 'critical')                       AS crit_24h,
          (SELECT percentile_cont(0.95)
                  WITHIN GROUP (ORDER BY latency_ms)
             FROM requests
            WHERE day = $1)                                    AS p95_ms
    """, today, yesterday)

    return {
        "dau":     {"today": row["dau_today"], "delta_pct": _delta(row)},
        "cost":    {"today_brl": round(row["cost_today"] * 5.0, 2)},
        "errors":  {"total": row["errors_24h"], "critical": row["crit_24h"]},
        "latency": {"p95_s": round((row["p95_ms"] or 0) / 1000, 2)},
    }

Audit writer com redact + retention

# app/audit.py
import json, re, time
from pathlib import Path

EMAIL_RE = re.compile(r'[\w.+-]+@[\w-]+\.[\w.-]+')
CPF_RE   = re.compile(r'\d{3}\.\d{3}\.\d{3}-\d{2}')
LOG_DIR  = Path("/var/log/agent/audit")
RETENTION_DAYS = 90

def _redact(s: str) -> str:
    s = EMAIL_RE.sub("[email]", s)
    s = CPF_RE.sub("[cpf]", s)
    return s

def audit(event: dict) -> None:
    # nunca confiar — re-aplica redact em strings
    for k, v in list(event.items()):
        if isinstance(v, str):
            event[k] = _redact(v)
    event["ts"] = event.get("ts") or time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
    day_file = LOG_DIR / f"{event['ts'][:10]}.jsonl"
    with day_file.open("a") as f:
        f.write(json.dumps(event, ensure_ascii=False) + "\n")

def purge_expired() -> int:
    """Cron diário. Remove arquivos > 90 dias."""
    cutoff = time.time() - RETENTION_DAYS * 86400
    removed = 0
    for f in LOG_DIR.glob("*.jsonl"):
        if f.stat().st_mtime < cutoff:
            f.unlink()
            removed += 1
    return removed

-- migrations/004_usage_daily.sql
CREATE TABLE usage_daily (
    user_id      TEXT          NOT NULL,
    day          DATE          NOT NULL,
    model        TEXT          NOT NULL,
    tokens_in    BIGINT        NOT NULL DEFAULT 0,
    tokens_out   BIGINT        NOT NULL DEFAULT 0,
    requests     INT           NOT NULL DEFAULT 0,
    cost_usd     NUMERIC(10,6) NOT NULL DEFAULT 0,
    pricing_ver  TEXT          NOT NULL,
    updated_at   TIMESTAMPTZ   NOT NULL DEFAULT NOW(),
    PRIMARY KEY (user_id, day, model)
);

CREATE INDEX idx_usage_day        ON usage_daily (day);
CREATE INDEX idx_usage_user_month ON usage_daily (user_id, date_trunc('month', day));

# app/pricing.py
from datetime import date

PRICING = {
    "2026-03-01": {
        "claude-opus-4-7":   {"in": 15.00, "out": 75.00},
        "claude-sonnet-4-7": {"in":  3.00, "out": 15.00},
    },
    "2026-04-15": {
        "claude-opus-4-7":   {"in": 15.00, "out": 75.00},
        "claude-sonnet-4-7": {"in":  3.00, "out": 15.00},
        "claude-haiku-4-7":  {"in":  0.80, "out":  4.00},
    },
}

def pricing_for(day: date) -> tuple[str, dict]:
    """Retorna (version, table) vigente em 'day'. Nunca interpola."""
    versions = sorted(PRICING.keys())
    active = max(v for v in versions if date.fromisoformat(v) <= day)
    return active, PRICING[active]

def cost_usd(model: str, tin: int, tout: int, day: date) -> tuple[float, str]:
    ver, table = pricing_for(day)
    p = table[model]
    return (tin * p["in"] + tout * p["out"]) / 1_000_000, ver

# app/alerts.py
from datetime import datetime, timedelta

DEDUP_WINDOW = timedelta(hours=1)

async def check_alerts(db) -> list[dict]:
    fired = []
    now = datetime.utcnow()

    rules = [
        ("errors_critical_24h",
         "SELECT COUNT(*) FROM errors WHERE severity='critical' "
         "AND ts > NOW() - INTERVAL '24 hours'",
         lambda n: n > 0,
         "{n} erros críticos nas últimas 24h"),

        ("cost_over_budget",
         "SELECT COALESCE(SUM(cost_usd),0) FROM usage_daily WHERE day=CURRENT_DATE",
         lambda n: n > 25.0,
         "Custo do dia: ${n:.2f} (budget diário: $25)"),

        ("p95_latency",
         "SELECT percentile_cont(0.95) WITHIN GROUP (ORDER BY latency_ms) "
         "FROM requests WHERE ts > NOW() - INTERVAL '15 minutes'",
         lambda n: (n or 0) > 3000,
         "p95 latência {n:.0f}ms (SLO 3000ms)"),
    ]

    for key, sql, predicate, msg in rules:
        # snooze ativo? pula.
        snoozed = await db.fetchval(
            "SELECT until FROM alert_snooze WHERE key=$1 AND until > NOW()", key)
        if snoozed:
            continue

        n = await db.fetchval(sql)
        if not predicate(n):
            continue

        # dedup: não disparou nesta janela?
        last = await db.fetchval(
            "SELECT MAX(fired_at) FROM alert_log WHERE key=$1", key)
        if last and (now - last) < DEDUP_WINDOW:
            continue

        await db.execute(
            "INSERT INTO alert_log (key, fired_at, payload) VALUES ($1, $2, $3)",
            key, now, msg.format(n=n))
        fired.append({"key": key, "message": msg.format(n=n)})

    return fired

async def snooze(db, key: str, hours: int = 4):
    until = datetime.utcnow() + timedelta(hours=hours)
    await db.execute("""
        INSERT INTO alert_snooze (key, until) VALUES ($1, $2)
        ON CONFLICT (key) DO UPDATE SET until = EXCLUDED.until
    """, key, until)

-- migrations/005_system_flags.sql
CREATE TABLE system_flags (
    key      TEXT PRIMARY KEY,
    value    JSONB       NOT NULL,
    reason   TEXT,
    set_by   TEXT        NOT NULL,
    set_at   TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

-- killall:      key='killall',     value='true'
-- kill_user_X:  key='kill:u_8a3f', value='true'
-- read_only:    key='read_only',   value='true'

# app/killswitch.py
import time

_CACHE_TTL = 30  # segundos
_cache = {"ts": 0, "flags": {}}

async def _refresh(db):
    rows = await db.fetch("SELECT key, value FROM system_flags")
    _cache["flags"] = {r["key"]: r["value"] for r in rows}
    _cache["ts"] = time.time()

async def is_killed(db, user_id: str) -> tuple[bool, str | None]:
    if time.time() - _cache["ts"] > _CACHE_TTL:
        await _refresh(db)
    flags = _cache["flags"]
    if flags.get("killall") is True:
        return True, "system halted (killall)"
    if flags.get(f"kill:{user_id}") is True:
        return True, "user temporarily blocked"
    return False, None

# uso em CADA request handler
@router.post("/chat")
async def chat(req: ChatRequest, db = Depends(get_db)):
    killed, reason = await is_killed(db, req.user_id)
    if killed:
        audit({"action": "killed", "user_id": req.user_id, "reason": reason})
        raise HTTPException(503, detail=reason)
    return await process(req, db)