Agent Quota Monitoring Workflow

import json
from pathlib import Path
from datetime import datetime

DAILY_BUDGET = 500
MONTHLY_BUDGET = 10000
WARNING_PCT = 0.75
CRITICAL_PCT = 0.90

def load_usage() -> dict:
    path = Path("agent_usage.json")
    if path.exists():
        data = json.loads(path.read_text())
        today = datetime.utcnow().strftime("%Y-%m-%d")
        month = datetime.utcnow().strftime("%Y-%m")
        if data.get("date") != today:
            data["daily_used"] = 0
            data["date"] = today
            data["sessions"] = {}
        if data.get("month") != month:
            data["monthly_used"] = 0
            data["month"] = month
        return data
    return {"date": datetime.utcnow().strftime("%Y-%m-%d"), "month": datetime.utcnow().strftime("%Y-%m"), "daily_used": 0, "monthly_used": 0, "sessions": {}}

def save_usage(data: dict):
    Path("agent_usage.json").write_text(json.dumps(data, indent=2))

def record_usage(session_id: str, credits: int = 1):
    data = load_usage()
    data["daily_used"] += credits
    data["monthly_used"] += credits
    data["sessions"][session_id] = data["sessions"].get(session_id, 0) + credits
    save_usage(data)
    return check_budget(data)

def check_budget(data: dict) -> dict:
    daily_pct = data["daily_used"] / DAILY_BUDGET
    monthly_pct = data["monthly_used"] / MONTHLY_BUDGET
    alerts = []

    if daily_pct >= 1.0:
        alerts.append({"level": "HARD_LIMIT", "scope": "daily", "used": data["daily_used"], "limit": DAILY_BUDGET})
    elif daily_pct >= CRITICAL_PCT:
        alerts.append({"level": "CRITICAL", "scope": "daily", "used": data["daily_used"], "limit": DAILY_BUDGET})
    elif daily_pct >= WARNING_PCT:
        alerts.append({"level": "WARNING", "scope": "daily", "used": data["daily_used"], "limit": DAILY_BUDGET})

    if monthly_pct >= 1.0:
        alerts.append({"level": "HARD_LIMIT", "scope": "monthly", "used": data["monthly_used"], "limit": MONTHLY_BUDGET})
    elif monthly_pct >= CRITICAL_PCT:
        alerts.append({"level": "CRITICAL", "scope": "monthly", "used": data["monthly_used"], "limit": MONTHLY_BUDGET})

    return {
        "daily_used": data["daily_used"],
        "daily_budget": DAILY_BUDGET,
        "daily_pct": round(daily_pct * 100, 1),
        "monthly_used": data["monthly_used"],
        "monthly_budget": MONTHLY_BUDGET,
        "monthly_pct": round(monthly_pct * 100, 1),
        "alerts": alerts,
        "blocked": daily_pct >= 1.0 or monthly_pct >= 1.0,
        "top_sessions": sorted(data["sessions"].items(), key=lambda x: x[1], reverse=True)[:5],
    }

# Example: check current budget status
data = load_usage()
status = check_budget(data)
print(f"Daily: {status['daily_used']}/{status['daily_budget']} ({status['daily_pct']}%)")
print(f"Monthly: {status['monthly_used']}/{status['monthly_budget']} ({status['monthly_pct']}%)")
if status["alerts"]:
    for alert in status["alerts"]:
        print(f"  ALERT [{alert['level']}]: {alert['scope']} budget {alert['used']}/{alert['limit']}")
if status["blocked"]:
    print("  STATUS: API calls BLOCKED - budget exceeded")

const DAILY_BUDGET = 500;
const MONTHLY_BUDGET = 10000;

class QuotaMonitor {
  constructor() {
    this.dailyUsed = 0;
    this.monthlyUsed = 0;
    this.sessions = {};
  }

  record(sessionId, credits = 1) {
    this.dailyUsed += credits;
    this.monthlyUsed += credits;
    this.sessions[sessionId] = (this.sessions[sessionId] ?? 0) + credits;
    return this.check();
  }

  check() {
    const dailyPct = this.dailyUsed / DAILY_BUDGET;
    const monthlyPct = this.monthlyUsed / MONTHLY_BUDGET;
    const alerts = [];
    if (dailyPct >= 1) alerts.push({ level: "HARD_LIMIT", scope: "daily" });
    else if (dailyPct >= 0.9) alerts.push({ level: "CRITICAL", scope: "daily" });
    else if (dailyPct >= 0.75) alerts.push({ level: "WARNING", scope: "daily" });
    if (monthlyPct >= 1) alerts.push({ level: "HARD_LIMIT", scope: "monthly" });
    else if (monthlyPct >= 0.9) alerts.push({ level: "CRITICAL", scope: "monthly" });
    return { dailyUsed: this.dailyUsed, dailyPct: Math.round(dailyPct * 1000) / 10, monthlyUsed: this.monthlyUsed, monthlyPct: Math.round(monthlyPct * 1000) / 10, alerts, blocked: dailyPct >= 1 || monthlyPct >= 1 };
  }
}

const monitor = new QuotaMonitor();
// Simulate usage
for (let i = 0; i < 10; i++) monitor.record("agent-session-1");
const status = monitor.check();
console.log(`Daily: ${status.dailyUsed}/${DAILY_BUDGET} (${status.dailyPct}%)`);
console.log(`Monthly: ${status.monthlyUsed}/${MONTHLY_BUDGET} (${status.monthlyPct}%)`);
if (status.blocked) console.log("BLOCKED: budget exceeded");