prompt caching was never switched on anywhere

"""Report an Anthropic organization that never switched prompt caching on.

Read only. 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: switching caching on is a change to your
own messages.create() call, not something a script should do to you.

Note on what this report can and cannot say: the messages usage report returns
token sums per bucket and carries no request count at all, so nothing here is
expressed per request. Every ratio below is a ratio of tokens.
"""
import argparse
import datetime
import logging
import os
import sys

import requests

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

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

# Published multipliers on base input: a cache read is a tenth of the price of
# processing the same tokens uncached.
READ_MULTIPLIER = 0.10


def accumulate(results, into=None):
    """Sum the token fields that matter across usage-report results. Pure.

    cache_creation is a nested object holding ephemeral_5m_input_tokens and
    ephemeral_1h_input_tokens. A parser that looks for a flat field instead sums
    zero and reports a heavily cached organization as an uncached one, which is
    why this is a function with tests rather than four lines in a loop.
    """
    total = {"uncached": 0, "cache_read": 0, "write_5m": 0, "write_1h": 0}
    if into:
        total.update(into)
    for result in results or []:
        total["uncached"] += int(result.get("uncached_input_tokens") or 0)
        total["cache_read"] += int(result.get("cache_read_input_tokens") or 0)
        creation = result.get("cache_creation") or {}
        total["write_5m"] += int(creation.get("ephemeral_5m_input_tokens") or 0)
        total["write_1h"] += int(creation.get("ephemeral_1h_input_tokens") or 0)
    return total


def cache_saving_ceiling(uncached_tokens, reusable_fraction):
    """Base-rate tokens you could stop paying for, at best. Pure.

    Deliberately a ceiling and not an estimate: it assumes the given fraction of
    uncached input is a stable prefix that would hit the cache every time, and
    prices that fraction at the read rate instead of the base rate. Real
    integrations do worse. Nothing in the API can tell you what the fraction
    actually is, because the API never returns your prompts.
    """
    if not 0.0 <= reusable_fraction <= 1.0:
        raise ValueError("reusable_fraction must be between 0 and 1")
    return int(max(0, uncached_tokens) * reusable_fraction * (1.0 - READ_MULTIPLIER))


def verdict(total, min_input=1_000_000):
    """Classify one workload's 30 day token totals. Pure.

    Returns (state, detail). The three states that matter are kept apart on
    purpose: caching absent, caching present, and not enough traffic to make
    either claim.
    """
    reads = int(total.get("cache_read", 0))
    writes = int(total.get("write_5m", 0)) + int(total.get("write_1h", 0))
    uncached = int(total.get("uncached", 0))

    if reads > 0:
        return ("in-use",
                "%.1fM read token(s) against %.1fM written. Caching is on here; "
                "whether it earns its keep is the write to read ratio, which is "
                "a separate question." % (reads / 1e6, writes / 1e6))
    if writes > 0:
        return ("writes-only",
                "%.1fM cache write token(s) and not one read. Caching is switched "
                "on and paying nothing back, which costs more than leaving it off: "
                "a write is 1.25x (5m) or 2x (1h) base input, an uncached call is "
                "1x." % (writes / 1e6))
    if uncached < min_input:
        return ("too-little-traffic",
                "only %d uncached input token(s) in the window; too little to "
                "conclude anything" % uncached)
    return ("never-used",
            "%.1fM uncached input token(s), zero cache reads and zero cache "
            "writes. Caching has never been switched on for this workload."
            % (uncached / 1e6))


def get(session, path, params):
    r = session.get(API + path, params=params, 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 buckets(session, path, params):
    """Walk the paginated usage or cost 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 window_start(days):
    """Floor to midnight UTC, because starting_at must sit on a bucket boundary."""
    now = datetime.datetime.now(datetime.timezone.utc)
    midnight = now.replace(hour=0, minute=0, second=0, microsecond=0)
    return (midnight - datetime.timedelta(days=days)).strftime("%Y-%m-%dT%H:%M:%SZ")


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--days", type=int, default=30, help="days of daily buckets to read")
    ap.add_argument("--min-input", type=int, default=1_000_000,
                    help="uncached input tokens below which no claim is made")
    ap.add_argument("--reusable", type=float, default=0.5,
                    help="fraction of input you believe is a stable prefix, used "
                         "only for the printed ceiling")
    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

    s = requests.Session()
    s.headers.update({"x-api-key": admin, "anthropic-version": VERSION})

    params = {"starting_at": window_start(args.days), "bucket_width": "1d",
              "limit": min(args.days + 1, 31),
              "group_by[]": ["model", "workspace_id"]}

    workloads = {}
    for bucket in buckets(s, "/organizations/usage_report/messages", params):
        for result in bucket.get("results") or []:
            name = (result.get("model") or "all models",
                    result.get("workspace_id") or "default workspace")
            workloads[name] = accumulate([result], workloads.get(name))

    if not workloads:
        log.info("no message usage in the last %d day(s)", args.days)
        return 0

    off = 0
    for name, total in sorted(workloads.items(), key=lambda kv: -kv[1]["uncached"]):
        state, detail = verdict(total, args.min_input)
        line = "%-18s %s / %s  %s" % (state, name[0], name[1], detail)
        if state in ("in-use", "too-little-traffic"):
            log.info(line)
            continue
        off += 1
        log.warning(line)
        if state == "never-used":
            ceiling = cache_saving_ceiling(total["uncached"], args.reusable)
            log.warning("  at %.0f%% reusable prefix that is up to %.1fM base rate "
                        "input token(s) a window you would stop paying for",
                        args.reusable * 100, ceiling / 1e6)
            log.warning("  repair: add cache_control {\"type\": \"ephemeral\"} at the "
                        "end of the stable prefix, keep everything variable after "
                        "it, redeploy, then re-read this window tomorrow")
        else:
            log.warning("  repair: caching is already on here. Move the breakpoint "
                        "to the end of the stable prefix so entries get read back, "
                        "or remove it: paying to write and never read is worse "
                        "than not caching")

    log.info("%d workload(s), %d with caching switched off", len(workloads), off)
    return 1 if off else 0


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

/**
 * Report an Anthropic organization that never switched prompt caching on.
 *
 * Read only. 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.
 *
 * The messages usage report carries token sums and no request count, so every
 * ratio here is a ratio of tokens, never of calls.
 */
const API = 'https://api.anthropic.com/v1';
const VERSION = '2023-06-01';
const READ_MULTIPLIER = 0.10;

/**
 * Sum the token fields that matter across usage-report results. Pure.
 * cache_creation is a nested object; reading it flat sums zero and reports a
 * heavily cached organization as an uncached one.
 */
export function accumulate(results, into = null) {
  const total = { uncached: 0, cache_read: 0, write_5m: 0, write_1h: 0, ...(into ?? {}) };
  for (const result of results ?? []) {
    total.uncached += Number(result.uncached_input_tokens ?? 0);
    total.cache_read += Number(result.cache_read_input_tokens ?? 0);
    const creation = result.cache_creation ?? {};
    total.write_5m += Number(creation.ephemeral_5m_input_tokens ?? 0);
    total.write_1h += Number(creation.ephemeral_1h_input_tokens ?? 0);
  }
  return total;
}

/**
 * Base-rate tokens you could stop paying for, at best. Pure and deliberately a
 * ceiling: nothing in the API can tell you what fraction of your input is
 * really a stable prefix, because the API never returns your prompts.
 */
export function cacheSavingCeiling(uncachedTokens, reusableFraction) {
  if (!(reusableFraction >= 0 && reusableFraction <= 1)) {
    throw new RangeError('reusableFraction must be between 0 and 1');
  }
  return Math.floor(Math.max(0, uncachedTokens) * reusableFraction * (1 - READ_MULTIPLIER));
}

/** Classify one workload's token totals. Pure. */
export function verdict(total, minInput = 1_000_000) {
  const reads = Number(total.cache_read ?? 0);
  const writes = Number(total.write_5m ?? 0) + Number(total.write_1h ?? 0);
  const uncached = Number(total.uncached ?? 0);

  if (reads > 0) {
    return ['in-use',
      `${(reads / 1e6).toFixed(1)}M read token(s) against ${(writes / 1e6).toFixed(1)}M ` +
      'written. Caching is on here; whether it earns its keep is the write to ' +
      'read ratio, which is a separate question.'];
  }
  if (writes > 0) {
    return ['writes-only',
      `${(writes / 1e6).toFixed(1)}M cache write token(s) and not one read. Caching ` +
      'is switched on and paying nothing back, which costs more than leaving it ' +
      'off: a write is 1.25x (5m) or 2x (1h) base input, an uncached call is 1x.'];
  }
  if (uncached < minInput) {
    return ['too-little-traffic',
      `only ${uncached} uncached input token(s) in the window; too little to ` +
      'conclude anything'];
  }
  return ['never-used',
    `${(uncached / 1e6).toFixed(1)}M uncached input token(s), zero cache reads and ` +
    'zero cache writes. Caching has never been switched on for this workload.'];
}

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* buckets(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;
  }
}

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

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 days = Number(process.env.DAYS ?? 30);
  const minInput = Number(process.env.MIN_INPUT ?? 1_000_000);
  const reusable = Number(process.env.REUSABLE ?? 0.5);

  const params = {
    starting_at: windowStart(days),
    bucket_width: '1d',
    limit: Math.min(days + 1, 31),
    'group_by[]': ['model', 'workspace_id'],
  };

  const workloads = new Map();
  for await (const bucket of buckets(adminKey, '/organizations/usage_report/messages',
                                     params)) {
    for (const result of bucket.results ?? []) {
      const name = `${result.model ?? 'all models'} / ` +
                   `${result.workspace_id ?? 'default workspace'}`;
      workloads.set(name, accumulate([result], workloads.get(name)));
    }
  }

  if (workloads.size === 0) {
    console.log(`no message usage in the last ${days} day(s)`);
    return;
  }

  let off = 0;
  const ordered = [...workloads.entries()].sort((a, b) => b[1].uncached - a[1].uncached);
  for (const [name, total] of ordered) {
    const [state, detail] = verdict(total, minInput);
    const line = `${state.padEnd(18)} ${name}  ${detail}`;
    if (state === 'in-use' || state === 'too-little-traffic') { console.log(line); continue; }
    off += 1;
    console.warn(line);
    if (state === 'never-used') {
      const ceiling = cacheSavingCeiling(total.uncached, reusable);
      console.warn(`  at ${(reusable * 100).toFixed(0)}% reusable prefix that is up to ` +
                   `${(ceiling / 1e6).toFixed(1)}M base rate input token(s) a window ` +
                   'you would stop paying for');
      console.warn('  repair: add cache_control {"type": "ephemeral"} at the end of ' +
                   'the stable prefix, keep everything variable after it, redeploy, ' +
                   'then re-read this window tomorrow');
    } else {
      console.warn('  repair: caching is already on here. Move the breakpoint to the ' +
                   'end of the stable prefix so entries get read back, or remove it: ' +
                   'paying to write and never read is worse than not caching');
    }
  }

  console.log(`${workloads.size} workload(s), ${off} with caching switched off`);
  process.exitCode = off ? 1 : 0;
}

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

import pytest

from anthropic_prompt_cache_off import accumulate, cache_saving_ceiling, verdict


def test_accumulate_reads_the_nested_cache_creation_object():
    # The trap: these two fields live inside cache_creation, not at the top.
    total = accumulate([{
        "uncached_input_tokens": 100,
        "cache_read_input_tokens": 40,
        "cache_creation": {"ephemeral_5m_input_tokens": 7,
                           "ephemeral_1h_input_tokens": 3},
    }])
    assert total == {"uncached": 100, "cache_read": 40, "write_5m": 7, "write_1h": 3}


def test_accumulate_treats_absent_and_null_fields_as_zero():
    assert accumulate([{"uncached_input_tokens": None}])["uncached"] == 0
    assert accumulate([{}])["write_5m"] == 0
    assert accumulate(None)["cache_read"] == 0


def test_accumulate_adds_into_a_running_total():
    first = accumulate([{"uncached_input_tokens": 10}])
    second = accumulate([{"uncached_input_tokens": 5}], first)
    assert second["uncached"] == 15


def test_zero_reads_and_zero_writes_on_real_traffic_is_the_finding():
    state, detail = verdict({"uncached": 50_000_000, "cache_read": 0,
                             "write_5m": 0, "write_1h": 0})
    assert state == "never-used"
    assert "never been switched on" in detail


def test_writes_without_reads_is_the_other_note_not_this_one():
    state, detail = verdict({"uncached": 50_000_000, "cache_read": 0,
                             "write_5m": 4_000_000, "write_1h": 0})
    assert state == "writes-only"
    assert "worse" in detail or "more than leaving it off" in detail


def test_any_read_at_all_means_caching_is_on():
    assert verdict({"uncached": 5_000_000, "cache_read": 1, "write_5m": 0,
                    "write_1h": 0})[0] == "in-use"


def test_a_quiet_workload_makes_no_claim_either_way():
    state, _ = verdict({"uncached": 900, "cache_read": 0, "write_5m": 0, "write_1h": 0})
    assert state == "too-little-traffic"


def test_the_saving_ceiling_prices_the_reusable_share_at_the_read_rate():
    # 0.1x read rate, so 90% of the reusable share stops being paid for.
    assert cache_saving_ceiling(1_000_000, 1.0) == 900_000
    assert cache_saving_ceiling(1_000_000, 0.5) == 450_000
    assert cache_saving_ceiling(1_000_000, 0.0) == 0


def test_the_ceiling_refuses_a_fraction_that_is_not_a_fraction():
    with pytest.raises(ValueError):
        cache_saving_ceiling(1_000_000, 1.4)

import { test } from 'node:test';
import assert from 'node:assert/strict';
import {
  accumulate, cacheSavingCeiling, verdict, windowStart,
} from './anthropic-prompt-cache-off.mjs';

test('accumulate reads the nested cache_creation object', () => {
  const total = accumulate([{
    uncached_input_tokens: 100,
    cache_read_input_tokens: 40,
    cache_creation: { ephemeral_5m_input_tokens: 7, ephemeral_1h_input_tokens: 3 },
  }]);
  assert.deepEqual(total, { uncached: 100, cache_read: 40, write_5m: 7, write_1h: 3 });
});

test('accumulate treats absent and null fields as zero', () => {
  assert.equal(accumulate([{ uncached_input_tokens: null }]).uncached, 0);
  assert.equal(accumulate([{}]).write_5m, 0);
  assert.equal(accumulate(null).cache_read, 0);
});

test('accumulate adds into a running total', () => {
  const first = accumulate([{ uncached_input_tokens: 10 }]);
  assert.equal(accumulate([{ uncached_input_tokens: 5 }], first).uncached, 15);
});

test('zero reads and zero writes on real traffic is the finding', () => {
  const [state, detail] = verdict({
    uncached: 50_000_000, cache_read: 0, write_5m: 0, write_1h: 0,
  });
  assert.equal(state, 'never-used');
  assert.match(detail, /never been switched on/);
});

test('writes without reads is the other note not this one', () => {
  const [state, detail] = verdict({
    uncached: 50_000_000, cache_read: 0, write_5m: 4_000_000, write_1h: 0,
  });
  assert.equal(state, 'writes-only');
  assert.match(detail, /more than leaving it off/);
});

test('any read at all means caching is on', () => {
  assert.equal(verdict({ uncached: 5_000_000, cache_read: 1, write_5m: 0, write_1h: 0 })[0],
               'in-use');
});

test('a quiet workload makes no claim either way', () => {
  assert.equal(verdict({ uncached: 900, cache_read: 0, write_5m: 0, write_1h: 0 })[0],
               'too-little-traffic');
});

test('the saving ceiling prices the reusable share at the read rate', () => {
  assert.equal(cacheSavingCeiling(1_000_000, 1.0), 900_000);
  assert.equal(cacheSavingCeiling(1_000_000, 0.5), 450_000);
  assert.equal(cacheSavingCeiling(1_000_000, 0.0), 0);
});

test('the ceiling refuses a fraction that is not a fraction', () => {
  assert.throws(() => cacheSavingCeiling(1_000_000, 1.4), RangeError);
});

test('the window start is floored to midnight UTC', () => {
  assert.equal(windowStart(7, new Date('2026-08-30T13:45:12Z')), '2026-08-23T00:00:00Z');
});