output tokens per minute is the real ceiling, not RPM

"""Report an Anthropic output limiter that concurrency cannot fix.

Read only. Two GET requests and nothing else against the Admin API, which needs
an Admin API key (sk-ant-admin...); a workspace key is rejected by every
/v1/organizations/* path, and an Admin key can be provisioned read-only.

The repair is printed, never performed. Lowering an effort setting changes what
the model does with a question, and moving traffic to the Batch API changes
when answers arrive. Both are decisions with owners.

The messages usage report has no request-count field. That is why this script
never claims a request rate: it divides the peak output minute by the
configured RPM and prints the answer length at which the request limiter would
have bound first, which is a comparison the reader can make and the API cannot.
"""
import argparse
import datetime as dt
import logging
import os
import sys

import requests

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("anthropic_otpm_ceiling")

API = "https://api.anthropic.com/v1"
VERSION = "2023-06-01"

LIMITER_TYPES = ("requests_per_minute", "input_tokens_per_minute",
                 "output_tokens_per_minute")

FINDINGS = ("otpm-saturated", "both-limiters-saturated")


def generated(result):
    """Output tokens in one usage result. Pure.

    Thinking tokens are billed as output and counted as output, so they are
    already inside this number. There is no separate field to add and no way to
    subtract them, which is exactly why an effort change can saturate the
    output limiter with nothing else in the request having moved.
    """
    if not isinstance(result, dict):
        return 0
    try:
        return int(result.get("output_tokens") or 0)
    except (TypeError, ValueError):
        return 0


def received(result):
    """Input tokens in one usage result, from every field that carries them. Pure.

    Total input is cache_read + cache_creation + uncached. This is only used to
    decide whether the input limiter also had pressure on it, so it is summed
    generously rather than charged the way ITPM charges.
    """
    if not isinstance(result, dict):
        return 0
    total = 0
    for field in ("uncached_input_tokens", "cache_read_input_tokens"):
        try:
            total += int(result.get(field) or 0)
        except (TypeError, ValueError):
            pass
    creation = result.get("cache_creation") or {}
    for field in ("ephemeral_5m_input_tokens", "ephemeral_1h_input_tokens"):
        try:
            total += int(creation.get(field) or 0)
        except (TypeError, ValueError):
            pass
    return total


def peaks(buckets):
    """Fold one-minute buckets into per-model output peaks. Pure.

    The input recorded is the input from the minute output peaked, not the
    largest input minute in the window. The judgement this script makes is
    whether output was full while input had room, and pairing two peaks from
    two different minutes describes a workload that never ran.
    """
    per_minute = {}
    for bucket in buckets or []:
        stamp = str(bucket.get("starting_at") or bucket.get("start_time") or "")
        for result in bucket.get("results") or []:
            model = str(result.get("model") or "").strip() or "all models"
            row = per_minute.setdefault((model, stamp), {"out": 0, "in": 0})
            row["out"] += generated(result)
            row["in"] += received(result)

    out = {}
    for (model, stamp), row in per_minute.items():
        stats = out.setdefault(model, {"peak_out": 0, "peak_at": None,
                                       "input_at_peak": 0, "minutes": 0,
                                       "total_out": 0})
        stats["minutes"] += 1
        stats["total_out"] += row["out"]
        if row["out"] > stats["peak_out"]:
            stats["peak_out"] = row["out"]
            stats["peak_at"] = stamp
            stats["input_at_peak"] = row["in"]
    return out


def limits_by_group(payload):
    """{model_group: {limiter type: value}} from the rate-limits response. Pure.

    All three limiters are kept because all three are needed: output for the
    verdict, input to rule out the sibling finding, and requests to compute the
    answer length at which the request rate would have mattered. A type absent
    from limits[] is None, which means it inherits, never that it is unlimited.
    """
    out = {}
    for entry in (payload or {}).get("data") or []:
        group = str(entry.get("model_group") or "").strip()
        if not group:
            continue
        row = out.setdefault(group, dict.fromkeys(LIMITER_TYPES))
        for limit in entry.get("limits") or []:
            kind = str(limit.get("type") or "").strip()
            if kind not in row:
                continue
            try:
                row[kind] = int(limit.get("value"))
            except (TypeError, ValueError):
                row[kind] = None
    return out


def limits_for(groups, model):
    """The limiter row for the group a model id belongs to. Pure. Longest prefix wins."""
    name = str(model or "").strip().lower()
    if not name:
        return None
    best_key, best_len = None, -1
    for group in (groups or {}):
        candidate = str(group).strip().lower()
        if not candidate:
            continue
        if name == candidate or name.startswith(candidate):
            if len(candidate) > best_len:
                best_key, best_len = group, len(candidate)
    if best_key is None:
        return None
    return (groups or {}).get(best_key)


def implied_mean_output(peak_output, rpm):
    """Answer length at which RPM would bind before OTPM. Pure.

    If a minute generated peak_output tokens, the request limiter could only
    have been what stopped you if you were also making rpm calls in that
    minute, which means a mean answer of peak_output / rpm tokens. Longer
    answers than that and the request rate was never close to its ceiling.

    This exists because the usage report has no request count. It converts a
    question the API cannot answer into one the reader already knows.
    """
    if rpm is None or rpm <= 0:
        return None
    try:
        peak = float(peak_output or 0)
    except (TypeError, ValueError):
        return None
    if peak <= 0:
        return None
    return peak / float(rpm)


def output_to_input_ratio(limits):
    """OTPM as a share of ITPM for one model group. Pure.

    Roughly one fifth at every tier, which is the structural reason a
    generation workload reaches the output ceiling first. Printed rather than
    assumed, because a workspace override can change it.
    """
    if not isinstance(limits, dict):
        return None
    otpm = limits.get("output_tokens_per_minute")
    itpm = limits.get("input_tokens_per_minute")
    if otpm is None or itpm is None or itpm <= 0:
        return None
    return otpm / float(itpm)


def verdict(model, stats, limits, floor=0.9, watch=0.6, min_minutes=10):
    """Classify one model's output limiter. Pure. Returns (state, detail)."""
    minutes = int((stats or {}).get("minutes") or 0)
    if minutes < min_minutes:
        return ("too-few-buckets",
                "%d minute(s) of traffic in the window, under the floor of %d. "
                "A peak taken over this little is noise." % (minutes, min_minutes))

    row = limits if isinstance(limits, dict) else {}
    otpm = row.get("output_tokens_per_minute")
    if otpm is None or otpm <= 0:
        return ("no-limit-published",
                "no output_tokens_per_minute is published for this model's "
                "group, so there is no ceiling to compare the peak against. The "
                "limiter still exists; the number was simply not returned.")

    peak_out = int(stats.get("peak_out") or 0)
    out_used = peak_out / float(otpm)

    itpm = row.get("input_tokens_per_minute")
    in_used = None
    if itpm is not None and itpm > 0:
        in_used = int(stats.get("input_at_peak") or 0) / float(itpm)

    shape = ("peak minute generated %d of an OTPM of %d (%.0f%%)"
             % (peak_out, otpm, out_used * 100))
    shape += (" while input sat at %.0f%% of ITPM" % (in_used * 100)
              if in_used is not None else ", with no ITPM published to compare")

    if out_used >= floor and in_used is not None and in_used >= floor:
        return ("both-limiters-saturated",
                shape + ". Both token limiters are full, so this is volume "
                "rather than shape: caching the prefix helps the input side "
                "and does nothing for the output side, and only batching or a "
                "limit increase moves both.")
    if out_used >= floor:
        return ("otpm-saturated",
                shape + ". The output limiter is what you are hitting, and "
                "there is no cached output, so nothing about the prompt moves "
                "this number.")
    if in_used is not None and in_used >= floor and out_used < watch:
        return ("input-bound",
                shape + ". The input limiter is the one that is full here, not "
                "the output one. Cache reads are not charged against ITPM, so "
                "that is a different finding with a different repair.")
    if out_used >= watch:
        return ("otpm-approaching",
                shape + ". Thin enough that a rise in answer length, or in "
                "thinking effort, lands on the output limiter.")
    return ("otpm-headroom", shape + ".")


def window_start(minutes):
    """Floor to the minute: starting_at must sit on a bucket boundary."""
    now = dt.datetime.now(dt.timezone.utc).replace(second=0, microsecond=0)
    return (now - dt.timedelta(minutes=minutes)).strftime("%Y-%m-%dT%H:%M:%SZ")


def get(session, path, params=None):
    r = session.get(API + path, params=params or {}, timeout=60)
    if r.status_code in (401, 403):
        raise SystemExit("%d from Anthropic: /v1/organizations/* needs an Admin "
                         "API key (sk-ant-admin...), not a workspace key"
                         % r.status_code)
    r.raise_for_status()
    return r.json()


def read_buckets(session, path, params):
    """Walk the paginated usage report."""
    params = dict(params)
    while True:
        page = get(session, path, params)
        for bucket in page.get("data") or []:
            yield bucket
        if not page.get("has_more") or not page.get("next_page"):
            return
        params["page"] = page["next_page"]


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--minutes", type=int, default=240,
                    help="minutes of one-minute buckets to read (max 1440)")
    ap.add_argument("--show-all", action="store_true",
                    help="also print models with headroom left")
    args = ap.parse_args()

    admin = os.environ.get("ANTHROPIC_ADMIN_KEY")
    if not admin:
        log.error("set ANTHROPIC_ADMIN_KEY to an Admin API key (sk-ant-admin...); "
                  "a workspace key cannot read /v1/organizations/*")
        return 2

    minutes = max(1, min(int(args.minutes), 1440))
    session = requests.Session()
    session.headers.update({"x-api-key": admin, "anthropic-version": VERSION})

    params = {"starting_at": window_start(minutes), "bucket_width": "1m",
              "limit": minutes, "group_by[]": ["model"]}
    stats = peaks(read_buckets(session, "/organizations/usage_report/messages", params))
    if not stats:
        log.info("no message usage in the last %d minute(s)", minutes)
        return 0

    groups = limits_by_group(get(session, "/organizations/rate_limits"))

    checked = 0
    bad = 0
    for model in sorted(stats, key=lambda m: -stats[m]["peak_out"]):
        row = stats[model]
        limits = limits_for(groups, model)
        state, detail = verdict(model, row, limits)
        checked += 1
        line = "%-24s %-28s %s" % (state, model, detail)

        if state in FINDINGS:
            bad += 1
            log.warning(line)
            mean = implied_mean_output(row["peak_out"],
                                       (limits or {}).get("requests_per_minute"))
            if mean is not None:
                log.warning("  RPM would only have bound first at a mean answer "
                            "of %.0f token(s) or shorter, so if your answers are "
                            "longer than that the request rate was never the "
                            "ceiling and more workers add nothing", mean)
            else:
                log.warning("  no requests_per_minute published for this group, "
                            "so the request rate cannot be ruled out from here")
            ratio = output_to_input_ratio(limits)
            if ratio is not None:
                log.warning("  OTPM is %.0f%% of ITPM on this group, so "
                            "generation reaches its ceiling first", ratio * 100)
            log.warning("  repair: move latency tolerant generation to the "
                        "Message Batches API, which has its own limiter group "
                        "and costs half; or lower output_config.effort, since "
                        "thinking tokens are counted as output; or request an "
                        "output_tokens_per_minute increase.")
            log.warning("  repair: do not lower max_tokens. It is documented "
                        "not to factor into OTPM, so it truncates answers "
                        "without buying a single token of headroom.")
        elif state == "input-bound":
            log.warning(line)
            log.warning("  repair: this one is the input limiter. Cache reads "
                        "are not charged against ITPM, so covering the stable "
                        "prefix is the lever there, not anything on this page.")
        elif state in ("otpm-approaching", "no-limit-published"):
            log.warning(line)
        elif args.show_all:
            log.info(line)

    log.info("%d model(s) checked, %d finding(s)", checked, bad)
    return 1 if bad else 0


if __name__ == "__main__":
    sys.exit(main())

/**
 * Report an Anthropic output limiter that concurrency cannot fix.
 *
 * Read only. Two GET requests and nothing else against the Admin API, which
 * needs an Admin API key (sk-ant-admin...); a workspace key is rejected by
 * every /v1/organizations/* path. The repair is printed, never performed.
 *
 * The messages usage report has no request-count field, so this script never
 * claims a request rate: it divides the peak output minute by the configured
 * RPM and prints the answer length at which the request limiter would have
 * bound first.
 */
const API = 'https://api.anthropic.com/v1';
const VERSION = '2023-06-01';

const LIMITER_TYPES = ['requests_per_minute', 'input_tokens_per_minute',
                       'output_tokens_per_minute'];

const FINDINGS = new Set(['otpm-saturated', 'both-limiters-saturated']);

const int = (v) => (Number.isFinite(Number(v)) ? Math.trunc(Number(v)) : 0);

/**
 * Output tokens in one usage result. Pure.
 * Thinking tokens are billed and counted as output, so they are already inside
 * this number: there is nothing to add and nothing to subtract.
 */
export function generated(result) {
  if (!result || typeof result !== 'object') return 0;
  return int(result.output_tokens);
}

/**
 * Input tokens in one usage result, from every field that carries them. Pure.
 * Summed generously rather than charged the way ITPM charges, because this is
 * only used to decide whether the input limiter also had pressure on it.
 */
export function received(result) {
  if (!result || typeof result !== 'object') return 0;
  const creation = result.cache_creation ?? {};
  return int(result.uncached_input_tokens) + int(result.cache_read_input_tokens)
    + int(creation.ephemeral_5m_input_tokens) + int(creation.ephemeral_1h_input_tokens);
}

/**
 * Fold one-minute buckets into per-model output peaks. Pure.
 * The input kept is the input from the minute output peaked, not the largest
 * input minute: two peaks from two minutes describe a workload that never ran.
 */
export function peaks(buckets) {
  const perMinute = new Map();
  for (const bucket of buckets ?? []) {
    const stamp = String(bucket.starting_at ?? bucket.start_time ?? '');
    for (const result of bucket.results ?? []) {
      const model = String(result.model ?? '').trim() || 'all models';
      const key = `${model}\u0000${stamp}`;
      const row = perMinute.get(key) ?? { model, stamp, out: 0, in: 0 };
      row.out += generated(result);
      row.in += received(result);
      perMinute.set(key, row);
    }
  }

  const out = {};
  for (const row of perMinute.values()) {
    const stats = out[row.model] ?? { peak_out: 0, peak_at: null, input_at_peak: 0,
                                      minutes: 0, total_out: 0 };
    stats.minutes += 1;
    stats.total_out += row.out;
    if (row.out > stats.peak_out) {
      stats.peak_out = row.out;
      stats.peak_at = row.stamp;
      stats.input_at_peak = row.in;
    }
    out[row.model] = stats;
  }
  return out;
}

/**
 * {model_group: {limiter type: value}} from the rate-limits response. Pure.
 * All three limiters are kept; a type absent from limits[] is null, which means
 * it inherits, never that it is unlimited.
 */
export function limitsByGroup(payload) {
  const out = {};
  for (const entry of (payload ?? {}).data ?? []) {
    const group = String(entry.model_group ?? '').trim();
    if (!group) continue;
    if (!out[group]) {
      out[group] = {};
      for (const t of LIMITER_TYPES) out[group][t] = null;
    }
    for (const limit of entry.limits ?? []) {
      const kind = String(limit.type ?? '').trim();
      if (!(kind in out[group])) continue;
      const value = Number(limit.value);
      out[group][kind] = Number.isInteger(value) ? value : null;
    }
  }
  return out;
}

/** The limiter row for the group a model id belongs to. Pure. Longest prefix wins. */
export function limitsFor(groups, model) {
  const name = String(model ?? '').trim().toLowerCase();
  if (!name) return null;
  let bestKey = null;
  let bestLen = -1;
  for (const group of Object.keys(groups ?? {})) {
    const candidate = group.trim().toLowerCase();
    if (!candidate) continue;
    if (name === candidate || name.startsWith(candidate)) {
      if (candidate.length > bestLen) { bestKey = group; bestLen = candidate.length; }
    }
  }
  return bestKey === null ? null : groups[bestKey];
}

/**
 * Answer length at which RPM would bind before OTPM. Pure.
 * peak_output / rpm. Longer answers than that and the request rate was never
 * close. This exists because the usage report has no request count: it turns a
 * question the API cannot answer into one the reader already knows.
 */
export function impliedMeanOutput(peakOutput, rpm) {
  if (rpm === null || rpm === undefined || rpm <= 0) return null;
  const peak = Number(peakOutput ?? 0);
  if (!Number.isFinite(peak) || peak <= 0) return null;
  return peak / rpm;
}

/**
 * OTPM as a share of ITPM for one model group. Pure.
 * Roughly one fifth at every tier, which is why generation hits its ceiling
 * first. Printed rather than assumed, because an override can change it.
 */
export function outputToInputRatio(limits) {
  if (!limits || typeof limits !== 'object') return null;
  const otpm = limits.output_tokens_per_minute;
  const itpm = limits.input_tokens_per_minute;
  if (otpm === null || otpm === undefined) return null;
  if (itpm === null || itpm === undefined || itpm <= 0) return null;
  return otpm / itpm;
}

/** Classify one model's output limiter. Pure. Returns [state, detail]. */
export function verdict(model, stats, limits, {
  floor = 0.9, watch = 0.6, minMinutes = 10,
} = {}) {
  const minutes = Number((stats ?? {}).minutes ?? 0);
  if (minutes < minMinutes) {
    return ['too-few-buckets',
      `${minutes} minute(s) of traffic in the window, under the floor of ` +
      `${minMinutes}. A peak taken over this little is noise.`];
  }

  const row = (limits && typeof limits === 'object') ? limits : {};
  const otpm = row.output_tokens_per_minute;
  if (otpm === null || otpm === undefined || otpm <= 0) {
    return ['no-limit-published',
      "no output_tokens_per_minute is published for this model's group, so " +
      'there is no ceiling to compare the peak against. The limiter still ' +
      'exists; the number was simply not returned.'];
  }

  const peakOut = Number(stats.peak_out ?? 0);
  const outUsed = peakOut / otpm;

  const itpm = row.input_tokens_per_minute;
  let inUsed = null;
  if (itpm !== null && itpm !== undefined && itpm > 0) {
    inUsed = Number(stats.input_at_peak ?? 0) / itpm;
  }

  let shape = `peak minute generated ${peakOut} of an OTPM of ${otpm} ` +
              `(${(outUsed * 100).toFixed(0)}%)`;
  shape += inUsed === null
    ? ', with no ITPM published to compare'
    : ` while input sat at ${(inUsed * 100).toFixed(0)}% of ITPM`;

  if (outUsed >= floor && inUsed !== null && inUsed >= floor) {
    return ['both-limiters-saturated',
      `${shape}. Both token limiters are full, so this is volume rather than ` +
      'shape: caching the prefix helps the input side and does nothing for the ' +
      'output side, and only batching or a limit increase moves both.'];
  }
  if (outUsed >= floor) {
    return ['otpm-saturated',
      `${shape}. The output limiter is what you are hitting, and there is no ` +
      'cached output, so nothing about the prompt moves this number.'];
  }
  if (inUsed !== null && inUsed >= floor && outUsed < watch) {
    return ['input-bound',
      `${shape}. The input limiter is the one that is full here, not the output ` +
      'one. Cache reads are not charged against ITPM, so that is a different ' +
      'finding with a different repair.'];
  }
  if (outUsed >= watch) {
    return ['otpm-approaching',
      `${shape}. Thin enough that a rise in answer length, or in thinking ` +
      'effort, lands on the output limiter.'];
  }
  return ['otpm-headroom', `${shape}.`];
}

/** Floor to the minute: starting_at must sit on a bucket boundary. */
export function windowStart(minutes, now = new Date()) {
  const floored = Date.UTC(now.getUTCFullYear(), now.getUTCMonth(), now.getUTCDate(),
                           now.getUTCHours(), now.getUTCMinutes());
  return new Date(floored - minutes * 60000).toISOString().replace(/\.\d{3}Z$/, 'Z');
}

async function get(adminKey, path, params = {}) {
  const url = new URL(API + path);
  for (const [k, v] of Object.entries(params)) {
    for (const one of Array.isArray(v) ? v : [v]) url.searchParams.append(k, one);
  }
  const res = await fetch(url, {
    headers: { 'x-api-key': adminKey, 'anthropic-version': VERSION },
  });
  if (res.status === 401 || res.status === 403) {
    throw new Error(`${res.status} from Anthropic: /v1/organizations/* needs an ` +
                    'Admin API key (sk-ant-admin...), not a workspace key');
  }
  if (!res.ok) throw new Error(`${res.status} from ${url.pathname}`);
  return res.json();
}

async function* readBuckets(adminKey, path, params) {
  const q = { ...params };
  for (;;) {
    const page = await get(adminKey, path, q);
    for (const bucket of page.data ?? []) yield bucket;
    if (!page.has_more || !page.next_page) return;
    q.page = page.next_page;
  }
}

async function main() {
  const adminKey = process.env.ANTHROPIC_ADMIN_KEY;
  if (!adminKey) {
    console.error('set ANTHROPIC_ADMIN_KEY to an Admin API key (sk-ant-admin...); ' +
                  'a workspace key cannot read /v1/organizations/*');
    process.exitCode = 2;
    return;
  }
  const minutes = Math.max(1, Math.min(Number(process.env.MINUTES ?? 240), 1440));
  const showAll = process.env.SHOW_ALL === '1';

  const collected = [];
  for await (const bucket of readBuckets(adminKey, '/organizations/usage_report/messages',
    { starting_at: windowStart(minutes), bucket_width: '1m', limit: minutes,
      'group_by[]': ['model'] })) {
    collected.push(bucket);
  }
  const stats = peaks(collected);
  const models = Object.keys(stats);
  if (models.length === 0) {
    console.log(`no message usage in the last ${minutes} minute(s)`);
    return;
  }

  const groups = limitsByGroup(await get(adminKey, '/organizations/rate_limits'));

  let bad = 0;
  models.sort((a, b) => stats[b].peak_out - stats[a].peak_out);
  for (const model of models) {
    const row = stats[model];
    const limits = limitsFor(groups, model);
    const [state, detail] = verdict(model, row, limits);
    const line = `${state.padEnd(24)} ${model.padEnd(28)} ${detail}`;

    if (FINDINGS.has(state)) {
      bad += 1;
      console.warn(line);
      const mean = impliedMeanOutput(row.peak_out, (limits ?? {}).requests_per_minute);
      if (mean !== null) {
        console.warn(`  RPM would only have bound first at a mean answer of ` +
                     `${mean.toFixed(0)} token(s) or shorter, so if your answers are ` +
                     'longer than that the request rate was never the ceiling and ' +
                     'more workers add nothing');
      } else {
        console.warn('  no requests_per_minute published for this group, so the ' +
                     'request rate cannot be ruled out from here');
      }
      const ratio = outputToInputRatio(limits);
      if (ratio !== null) {
        console.warn(`  OTPM is ${(ratio * 100).toFixed(0)}% of ITPM on this group, ` +
                     'so generation reaches its ceiling first');
      }
      console.warn('  repair: move latency tolerant generation to the Message ' +
                   'Batches API, which has its own limiter group and costs half; or ' +
                   'lower output_config.effort, since thinking tokens are counted as ' +
                   'output; or request an output_tokens_per_minute increase.');
      console.warn('  repair: do not lower max_tokens. It is documented not to factor ' +
                   'into OTPM, so it truncates answers without buying a single token ' +
                   'of headroom.');
    } else if (state === 'input-bound') {
      console.warn(line);
      console.warn('  repair: this one is the input limiter. Cache reads are not ' +
                   'charged against ITPM, so covering the stable prefix is the lever ' +
                   'there, not anything on this page.');
    } else if (state === 'otpm-approaching' || state === 'no-limit-published') {
      console.warn(line);
    } else if (showAll) {
      console.log(line);
    }
  }

  console.log(`${models.length} model(s) checked, ${bad} finding(s)`);
  process.exitCode = bad ? 1 : 0;
}

if (import.meta.url === `file://${process.argv[1]}`) {
  main().catch((err) => { console.error(err.message); process.exitCode = 2; });
}

from anthropic_otpm_ceiling import (generated, implied_mean_output, limits_by_group,
                                    limits_for, output_to_input_ratio, peaks,
                                    received, verdict)

SONNET = {"requests_per_minute": 4000,
          "input_tokens_per_minute": 5000000,
          "output_tokens_per_minute": 1000000}


def minute(stamp, model, out=0, uncached=0, read=0):
    """One 1m bucket from GET /v1/organizations/usage_report/messages."""
    return {"starting_at": stamp, "results": [{
        "model": model,
        "output_tokens": out,
        "uncached_input_tokens": uncached,
        "cache_read_input_tokens": read,
        "cache_creation": {"ephemeral_5m_input_tokens": 0,
                           "ephemeral_1h_input_tokens": 0},
    }]}


def test_a_full_output_limiter_beside_a_comfortable_input_one_is_the_finding():
    stats = peaks([minute("2026-08-30T14:%02d:00Z" % i, "claude-opus-5",
                          out=980_000 if i == 5 else 200_000,
                          uncached=1_200_000 if i == 5 else 400_000)
                   for i in range(20)])
    state, detail = verdict("claude-opus-5", stats["claude-opus-5"], SONNET)
    assert state == "otpm-saturated"
    assert "generated 980000 of an OTPM of 1000000 (98%)" in detail
    assert "while input sat at 24% of ITPM" in detail
    assert "no cached output" in detail
    # The conclusion the note exists for: RPM was never the ceiling.
    assert round(implied_mean_output(980_000, 4000)) == 245
    assert round(output_to_input_ratio(SONNET) * 100) == 20


def test_a_full_input_limiter_is_handed_to_the_other_note():
    # The same fold and the same endpoints, and the opposite finding. If this
    # state did not exist, this script would prescribe batching and effort
    # changes for a workload whose repair is a cache breakpoint.
    stats = peaks([minute("2026-08-30T14:%02d:00Z" % i, "claude-sonnet-5",
                          out=100_000 if i == 9 else 20_000,
                          uncached=4_900_000 if i == 9 else 300_000)
                   for i in range(20)])
    state, detail = verdict("claude-sonnet-5", stats["claude-sonnet-5"], SONNET)
    assert state == "input-bound"
    assert "input limiter is the one that is full here" in detail


def test_both_limiters_full_is_volume_rather_than_shape():
    stats = peaks([minute("2026-08-30T14:%02d:00Z" % i, "claude-sonnet-5",
                          out=950_000, uncached=4_800_000) for i in range(20)])
    state, detail = verdict("claude-sonnet-5", stats["claude-sonnet-5"], SONNET)
    assert state == "both-limiters-saturated"
    assert "does nothing for the output side" in detail


def test_the_input_recorded_is_from_the_minute_output_peaked():
    # Output peaks at 14:05 and input peaks at 14:12. Taking the maximum of each
    # independently would report 98% of OTPM against 98% of ITPM and invent a
    # minute that never happened.
    buckets = [minute("2026-08-30T14:%02d:00Z" % i, "claude-opus-5",
                      out=200_000, uncached=400_000) for i in range(20)]
    buckets[5] = minute("2026-08-30T14:05:00Z", "claude-opus-5",
                        out=980_000, uncached=1_200_000)
    buckets[12] = minute("2026-08-30T14:12:00Z", "claude-opus-5",
                         out=300_000, uncached=4_900_000)
    row = peaks(buckets)["claude-opus-5"]
    assert row["peak_out"] == 980_000
    assert row["peak_at"] == "2026-08-30T14:05:00Z"
    assert row["input_at_peak"] == 1_200_000
    assert verdict("claude-opus-5", row, SONNET)[0] == "otpm-saturated"


def test_input_is_summed_from_every_field_that_carries_it():
    result = {"output_tokens": 50, "uncached_input_tokens": 100,
              "cache_read_input_tokens": 900,
              "cache_creation": {"ephemeral_5m_input_tokens": 7,
                                 "ephemeral_1h_input_tokens": 3}}
    assert generated(result) == 50
    assert received(result) == 1010
    assert generated({}) == 0
    assert generated(None) == 0
    assert received(None) == 0


def test_the_implied_answer_length_refuses_to_guess():
    assert implied_mean_output(980_000, None) is None
    assert implied_mean_output(980_000, 0) is None
    assert implied_mean_output(0, 4000) is None
    assert output_to_input_ratio({"output_tokens_per_minute": 1000}) is None
    assert output_to_input_ratio(None) is None


def test_an_unpublished_output_ceiling_gets_no_verdict():
    groups = limits_by_group({"data": [
        {"model_group": "claude-sonnet-5", "limits": [
            {"type": "requests_per_minute", "value": 4000},
            {"type": "input_tokens_per_minute", "value": 5000000},
            {"type": "output_tokens_per_minute", "value": 1000000}]},
        {"model_group": "claude-fable-5", "limits": [
            {"type": "requests_per_minute", "value": 500}]},
    ]})
    assert limits_for(groups, "claude-sonnet-5-20260101") == SONNET
    fable = limits_for(groups, "claude-fable-5")
    assert fable["output_tokens_per_minute"] is None
    assert verdict("claude-fable-5", {"minutes": 60, "peak_out": 9}, fable)[0] \
        == "no-limit-published"
    assert limits_for(groups, "claude-haiku-4-5-20251001") is None
    assert verdict("claude-opus-5", {"minutes": 2, "peak_out": 9}, SONNET)[0] \
        == "too-few-buckets"

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { generated, impliedMeanOutput, limitsByGroup, limitsFor,
         outputToInputRatio, peaks, received, verdict }
  from './anthropic-otpm-ceiling.mjs';

const SONNET = { requests_per_minute: 4000,
                 input_tokens_per_minute: 5000000,
                 output_tokens_per_minute: 1000000 };

/** One 1m bucket from GET /v1/organizations/usage_report/messages. */
function minute(stamp, model, { out = 0, uncached = 0, read = 0 } = {}) {
  return { starting_at: stamp, results: [{
    model,
    output_tokens: out,
    uncached_input_tokens: uncached,
    cache_read_input_tokens: read,
    cache_creation: { ephemeral_5m_input_tokens: 0, ephemeral_1h_input_tokens: 0 },
  }] };
}

const stamp = (i) => `2026-08-30T14:${String(i).padStart(2, '0')}:00Z`;

test('a full output limiter beside a comfortable input one is the finding', () => {
  const stats = peaks([...Array(20).keys()].map((i) =>
    minute(stamp(i), 'claude-opus-5',
      { out: i === 5 ? 980000 : 200000, uncached: i === 5 ? 1200000 : 400000 })));
  const [state, detail] = verdict('claude-opus-5', stats['claude-opus-5'], SONNET);
  assert.equal(state, 'otpm-saturated');
  assert.match(detail, /generated 980000 of an OTPM of 1000000 \(98%\)/);
  assert.match(detail, /while input sat at 24% of ITPM/);
  assert.match(detail, /no cached output/);
  assert.equal(Math.round(impliedMeanOutput(980000, 4000)), 245);
  assert.equal(Math.round(outputToInputRatio(SONNET) * 100), 20);
});

test('a full input limiter is handed to the other note', () => {
  const stats = peaks([...Array(20).keys()].map((i) =>
    minute(stamp(i), 'claude-sonnet-5',
      { out: i === 9 ? 100000 : 20000, uncached: i === 9 ? 4900000 : 300000 })));
  const [state, detail] = verdict('claude-sonnet-5', stats['claude-sonnet-5'], SONNET);
  assert.equal(state, 'input-bound');
  assert.match(detail, /input limiter is the one that is full here/);
});

test('both limiters full is volume rather than shape', () => {
  const stats = peaks([...Array(20).keys()].map((i) =>
    minute(stamp(i), 'claude-sonnet-5', { out: 950000, uncached: 4800000 })));
  const [state, detail] = verdict('claude-sonnet-5', stats['claude-sonnet-5'], SONNET);
  assert.equal(state, 'both-limiters-saturated');
  assert.match(detail, /does nothing for the output side/);
});

test('the input recorded is from the minute output peaked', () => {
  const buckets = [...Array(20).keys()].map((i) =>
    minute(stamp(i), 'claude-opus-5', { out: 200000, uncached: 400000 }));
  buckets[5] = minute(stamp(5), 'claude-opus-5', { out: 980000, uncached: 1200000 });
  buckets[12] = minute(stamp(12), 'claude-opus-5', { out: 300000, uncached: 4900000 });
  const row = peaks(buckets)['claude-opus-5'];
  assert.equal(row.peak_out, 980000);
  assert.equal(row.peak_at, '2026-08-30T14:05:00Z');
  assert.equal(row.input_at_peak, 1200000);
  assert.equal(verdict('claude-opus-5', row, SONNET)[0], 'otpm-saturated');
});

test('input is summed from every field that carries it', () => {
  const result = { output_tokens: 50, uncached_input_tokens: 100,
    cache_read_input_tokens: 900,
    cache_creation: { ephemeral_5m_input_tokens: 7, ephemeral_1h_input_tokens: 3 } };
  assert.equal(generated(result), 50);
  assert.equal(received(result), 1010);
  assert.equal(generated({}), 0);
  assert.equal(generated(null), 0);
  assert.equal(received(null), 0);
});

test('the implied answer length refuses to guess', () => {
  assert.equal(impliedMeanOutput(980000, null), null);
  assert.equal(impliedMeanOutput(980000, 0), null);
  assert.equal(impliedMeanOutput(0, 4000), null);
  assert.equal(outputToInputRatio({ output_tokens_per_minute: 1000 }), null);
  assert.equal(outputToInputRatio(null), null);
});

test('an unpublished output ceiling gets no verdict', () => {
  const groups = limitsByGroup({ data: [
    { model_group: 'claude-sonnet-5', limits: [
      { type: 'requests_per_minute', value: 4000 },
      { type: 'input_tokens_per_minute', value: 5000000 },
      { type: 'output_tokens_per_minute', value: 1000000 }] },
    { model_group: 'claude-fable-5', limits: [
      { type: 'requests_per_minute', value: 500 }] },
  ] });
  assert.deepEqual(limitsFor(groups, 'claude-sonnet-5-20260101'), SONNET);
  const fable = limitsFor(groups, 'claude-fable-5');
  assert.equal(fable.output_tokens_per_minute, null);
  assert.equal(verdict('claude-fable-5', { minutes: 60, peak_out: 9 }, fable)[0],
               'no-limit-published');
  assert.equal(limitsFor(groups, 'claude-haiku-4-5-20251001'), null);
  assert.equal(verdict('claude-opus-5', { minutes: 2, peak_out: 9 }, SONNET)[0],
               'too-few-buckets');
});