Handle usage limits

import os
import random
import time
from datetime import datetime, timezone
 
import requests
 
API_URL = "https://app.syntic.ai/v1/chat/completions"
HEADERS = {"Authorization": f"Bearer {os.environ['SYNTIC_API_KEY']}"}
 
PAUSE = {"session", "weekly"}
NEEDS_ADMIN = {"monthly", "spend_limit", "no_credits"}
 
 
class UsagePaused(Exception):
    """The plan window is full; it reopens by itself at `resets_at`."""
 
    def __init__(self, limit, resets_at):
        super().__init__(f"{limit} window full until {resets_at}")
        self.limit, self.resets_at = limit, resets_at
 
 
class UsageBlocked(Exception):
    """An administrator has to act before requests can succeed."""
 
    def __init__(self, limit, message):
        super().__init__(message)
        self.limit = limit
 
 
def backoff(attempt):
    # 1s, 2s, 4s, 8s ... plus up to one second of jitter
    time.sleep(2 ** attempt + random.random())
 
 
def chat(messages, model="Amara S1.2", max_tokens=1024, attempts=5):
    body = {"model": model, "messages": messages, "stream": False, "max_tokens": max_tokens}
 
    for attempt in range(attempts):
        try:
            res = requests.post(API_URL, headers=HEADERS, json=body, timeout=180)
        except requests.ConnectionError:
            backoff(attempt)
            continue
 
        if res.ok:
            source = res.headers.get("x-syntic-billing-source")
            if source == "api_credit":
                print("note: this request was paid from usage credits")
            return res.json()
 
        data = (res.json() if res.content else {}).get("error") or {}
 
        if res.status_code == 429:
            time.sleep(int(res.headers.get("Retry-After", "5")) + random.random())
            continue
 
        if res.status_code == 402 and data.get("code") == "quota_exhausted":
            limit = data.get("limit")
            if limit in PAUSE:
                raise UsagePaused(limit, data.get("resets_at"))
            if limit in NEEDS_ADMIN:
                raise UsageBlocked(limit, data.get("message", ""))
            backoff(attempt)  # "unavailable" or anything new: treat as temporary
            continue
 
        if res.status_code >= 500:
            backoff(attempt)
            continue
 
        # Any other 4xx: the request or the key is wrong. Retrying will not help.
        detail = data.get("message") or ""
        raise RuntimeError(f"{res.status_code} {data.get('error')}: {detail} "
                           f"(request {res.headers.get('x-syntic-request-id')})")
 
    raise RuntimeError("Gave up after repeated temporary failures.")
 
 
if __name__ == "__main__":
    try:
        reply = chat([{"role": "user", "content": "Give me a two-line packing list for a rainy weekend."}])
        print(reply["choices"][0]["message"]["content"])
    except UsagePaused as paused:
        when = datetime.fromisoformat(paused.resets_at.replace("Z", "+00:00"))
        wait = when - datetime.now(timezone.utc)
        print(f"Paused: the {paused.limit} limit reopens in {wait}.")
    except UsageBlocked as blocked:
        print(f"Blocked ({blocked.limit}). An administrator needs to act: {blocked}")

const API_URL = 'https://app.syntic.ai/v1/chat/completions'
const HEADERS = {
  Authorization: `Bearer ${process.env.SYNTIC_API_KEY}`,
  'Content-Type': 'application/json',
}
 
const PAUSE = new Set(['session', 'weekly'])
const NEEDS_ADMIN = new Set(['monthly', 'spend_limit', 'no_credits'])
 
export class UsagePaused extends Error {
  constructor(limit, resetsAt) {
    super(`${limit} window full until ${resetsAt}`)
    this.limit = limit
    this.resetsAt = resetsAt
  }
}
 
export class UsageBlocked extends Error {
  constructor(limit, message) {
    super(message)
    this.limit = limit
  }
}
 
const sleep = (ms) => new Promise((r) => setTimeout(r, ms))
const backoff = (attempt) => sleep(2 ** attempt * 1000 + Math.random() * 1000)
 
export async function chat(messages, { model = 'Amara S1.2', maxTokens = 1024, attempts = 5 } = {}) {
  const body = JSON.stringify({ model, messages, stream: false, max_tokens: maxTokens })
 
  for (let attempt = 0; attempt < attempts; attempt++) {
    let res
    try {
      res = await fetch(API_URL, { method: 'POST', headers: HEADERS, body })
    } catch {
      await backoff(attempt) // network failure
      continue
    }
 
    if (res.ok) return res.json()
 
    const data = ((await res.json().catch(() => ({}))) as { error?: Record<string, any> }).error ?? {}
 
    if (res.status === 429) {
      await sleep(Number(res.headers.get('retry-after') ?? 5) * 1000 + Math.random() * 1000)
      continue
    }
 
    if (res.status === 402 && data.code === 'quota_exhausted') {
      if (PAUSE.has(data.limit)) throw new UsagePaused(data.limit, data.resets_at)
      if (NEEDS_ADMIN.has(data.limit)) throw new UsageBlocked(data.limit, data.message ?? '')
      await backoff(attempt) // "unavailable" or a value added later
      continue
    }
 
    if (res.status >= 500) {
      await backoff(attempt)
      continue
    }
 
    throw new Error(
      `${res.status} ${data.code}: ${data.message ?? ''} ` +
        `(request ${res.headers.get('x-syntic-request-id')})`,
    )
  }
  throw new Error('Gave up after repeated temporary failures.')
}
 
try {
  const reply = await chat([{ role: 'user', content: 'Give me a two-line packing list for a rainy weekend.' }])
  console.log(reply.choices[0].message.content)
} catch (err) {
  if (err instanceof UsagePaused) console.log(`Paused until ${err.resetsAt}`)
  else if (err instanceof UsageBlocked) console.log(`Blocked (${err.limit}): ${err.message}`)
  else throw err
}