Cache written on every call by a prefix that keeps moving

"""Find Anthropic keys whose cache is rewritten on every call and never read.

Read only. One GET 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.

Totals cannot tell this apart from two neighbouring problems, so the evidence
is spacing. A run of adjacent one-minute buckets that each write a cache entry
and never read one is longer than the entry's own TTL, which means the entry
was alive and unmatched the whole time. Only a prefix that differs on every
call does that. Caching switched off, and caching that is read but not read
enough, are named and handed to their own notes.

The repair is printed, never performed. Moving a timestamp is a deploy.
"""
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_cache_prefix_churn")

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

# cache_creation is a nested object. A parser looking for a flat
# cache_creation_input_tokens sums zero and reports a key that writes on every
# call as one that never caches at all, which is the opposite finding.
CACHE_CREATION_FIELDS = ("ephemeral_5m_input_tokens", "ephemeral_1h_input_tokens")

FINDINGS = ("prefix-churn",)


def _int(value):
    """Read a usage field as an int. Pure. Missing and unreadable both mean 0."""
    try:
        return int(value or 0)
    except (TypeError, ValueError):
        return 0


def minute_key(stamp):
    """Normalise a timestamp to a UTC minute key. Pure. None if unreadable."""
    if isinstance(stamp, bool):
        return None
    if isinstance(stamp, (int, float)):
        try:
            when = dt.datetime.fromtimestamp(int(stamp), dt.timezone.utc)
        except (ValueError, OSError, OverflowError):
            return None
        return when.strftime("%Y-%m-%dT%H:%MZ")
    text = str(stamp or "").strip().replace(" ", "T")
    if len(text) < 16:
        return None
    head = text[:16]
    if head[4] != "-" or head[7] != "-" or head[10] != "T" or head[13] != ":":
        return None
    for part in (head[0:4], head[5:7], head[8:10], head[11:13], head[14:16]):
        if not part.isdigit():
            return None
    return head + "Z"


def minute_index(stamp):
    """Minutes since the epoch. Pure. None if unreadable.

    Adjacency is the entire finding, so it has to be arithmetic on integers.
    String comparison puts 14:59 and 15:00 two apart, which breaks every run
    that crosses an hour boundary and quietly halves the longest one.
    """
    key = minute_key(stamp)
    if key is None:
        return None
    try:
        when = dt.datetime(int(key[0:4]), int(key[5:7]), int(key[8:10]),
                           int(key[11:13]), int(key[14:16]), tzinfo=dt.timezone.utc)
    except ValueError:
        return None
    return int(when.timestamp()) // 60


def rows_by_key(buckets):
    """Per (api_key_id, model), one row per minute, sorted. Pure."""
    merged = {}
    for bucket in buckets or []:
        stamp = bucket.get("starting_at") or bucket.get("start_time")
        key = minute_key(stamp)
        index = minute_index(stamp)
        if key is None or index is None:
            continue
        for result in bucket.get("results") or []:
            if not isinstance(result, dict):
                continue
            ident = (str(result.get("api_key_id") or "unknown"),
                     str(result.get("model") or "unknown"))
            creation = result.get("cache_creation") or {}
            row = merged.setdefault((ident, index),
                                    {"minute": key, "index": index, "uncached": 0,
                                     "write5m": 0, "write1h": 0, "reads": 0})
            row["uncached"] += _int(result.get("uncached_input_tokens"))
            row["write5m"] += _int(creation.get("ephemeral_5m_input_tokens"))
            row["write1h"] += _int(creation.get("ephemeral_1h_input_tokens"))
            row["reads"] += _int(result.get("cache_read_input_tokens"))
    out = {}
    for (ident, _index), row in merged.items():
        out.setdefault(ident, []).append(row)
    for rows in out.values():
        rows.sort(key=lambda r: r["index"])
    return out


def writes(row):
    """Cache creation tokens in one minute, both TTLs. Pure."""
    return _int((row or {}).get("write5m")) + _int((row or {}).get("write1h"))


def write_share(row):
    """Share of a minute's input that was written as a fresh cache entry. Pure.

    None when nothing was sent, which is a different state from zero: an idle
    minute must not be counted as a minute that cached nothing.
    """
    total = _int((row or {}).get("uncached")) + writes(row)
    if total <= 0:
        return None
    return writes(row) / float(total)


def totals(rows):
    """Sum a series, and count the minutes that carried any traffic. Pure."""
    out = {"uncached": 0, "write5m": 0, "write1h": 0, "reads": 0, "active": 0}
    for row in rows or []:
        out["uncached"] += _int(row.get("uncached"))
        out["write5m"] += _int(row.get("write5m"))
        out["write1h"] += _int(row.get("write1h"))
        out["reads"] += _int(row.get("reads"))
        if _int(row.get("uncached")) + writes(row) + _int(row.get("reads")) > 0:
            out["active"] += 1
    out["writes"] = out["write5m"] + out["write1h"]
    return out


def churn_runs(rows, share_floor=0.5, read_floor=0.01):
    """Maximal runs of adjacent minutes that wrote and never read. Pure.

    This is the finding and nothing else in the section computes it. A five
    minute entry written in the first minute of a run is still alive in the
    fifth, so a run that long with no read in it means the entry was live and
    unmatched throughout. Neither a cold start nor a TTL expiring between calls
    can produce that; a prefix that differs on every call is the only thing
    that can.
    """
    runs = []
    current = []
    for row in rows or []:
        made = writes(row)
        share = write_share(row)
        churning = (made > 0 and share is not None and share >= share_floor
                    and _int(row.get("reads")) <= made * read_floor)
        if not churning:
            if current:
                runs.append(current)
                current = []
            continue
        if current and _int(row.get("index")) == _int(current[-1].get("index")) + 1:
            current.append(row)
        else:
            if current:
                runs.append(current)
            current = [row]
    if current:
        runs.append(current)
    return runs


def gap_profile(rows):
    """Median gap in minutes between minutes that wrote. Pure. None under two.

    The alternative explanation, stated as a number. Traffic arriving less
    often than the TTL writes an entry that expires before anything can read
    it, and that is a different note with a different repair. Its signature is
    isolated writing minutes; churn's is adjacent ones.
    """
    indices = [_int(r.get("index")) for r in rows or [] if writes(r) > 0]
    indices.sort()
    if len(indices) < 2:
        return None
    gaps = sorted(indices[i + 1] - indices[i] for i in range(len(indices) - 1))
    middle = len(gaps) // 2
    if len(gaps) % 2:
        return float(gaps[middle])
    return (gaps[middle - 1] + gaps[middle]) / 2.0


def ttl_split(sums):
    """Which TTL the writes were bought at. Pure. Returns (state, detail).

    It changes how damning a run is and what it cost. A 5 minute entry has to
    be matched within five minutes and is billed at 1.25x base input; a 1 hour
    entry is alive for sixty and is billed at 2x, so an adjacent run against
    hour-long writes is both stronger evidence and twice the surcharge.
    """
    sums = sums or {}
    five = _int(sums.get("write5m"))
    hour = _int(sums.get("write1h"))
    if five + hour <= 0:
        return ("no-writes", "nothing was written to the cache in this window")
    if hour > five:
        return ("1h-dominant",
                "the writes are mostly 1 hour entries at 2x base input, so each "
                "one was alive for sixty minutes and never matched in any of them")
    if five > hour:
        return ("5m-dominant",
                "the writes are 5 minute entries at 1.25x base input, so any run "
                "longer than five minutes outlived calls that never matched it")
    return ("mixed", "the writes are split evenly between the 5 minute and 1 "
                     "hour TTLs")


def handoff(state):
    """Which note owns this shape, when it is not this one. Pure.

    Three findings read the same two numbers. Naming the other two in the
    output is the difference between a check that classifies and a check that
    claims everything it sees.
    """
    if state == "caching-off":
        return ("no writes and no reads anywhere: caching was never switched "
                "on for this key. Read the prompt-caching-never-used note; the "
                "loss there is a discount not taken rather than a surcharge "
                "paid.")
    if state == "cache-is-read":
        return ("entries are being matched, so the prefix is stable enough to "
                "hit. Whether it hits often enough to pay for the write "
                "premium is the write-to-read ratio, which is the "
                "cache-writes-with-no-reads note.")
    if state == "gap-driven-misses":
        return ("the writing minutes are isolated rather than adjacent, so each "
                "entry plausibly expired before the next call arrived. That is "
                "arrival rate against TTL, and it is the "
                "cache-writes-with-no-reads note rather than this one.")
    return ""


def classify(rows, min_run=5, share_floor=0.5, read_floor=0.01, min_active=10):
    """Classify one key and model series. Pure. Returns (state, detail).

    The first three branches exist to give the finding away. Only a series with
    writes, no reads, a majority write share and adjacent writing minutes
    belongs to this note.
    """
    sums = totals(rows)
    if sums["active"] < min_active:
        return ("too-little-traffic",
                "%d active minute(s), under the floor of %d. Nothing can be "
                "said about spacing with fewer." % (sums["active"], min_active))

    if sums["writes"] == 0 and sums["reads"] == 0:
        return ("caching-off",
                "%d uncached input token(s), no cache writes and no cache reads"
                % sums["uncached"])
    if sums["writes"] == 0:
        return ("reads-only",
                "%d cache read(s) and no writes in this window: the entries "
                "were written before it started" % sums["reads"])
    if sums["reads"] > sums["writes"] * read_floor:
        return ("cache-is-read",
                "%d cache read token(s) against %d written"
                % (sums["reads"], sums["writes"]))

    share = sums["writes"] / float(sums["uncached"] + sums["writes"])
    if share < share_floor:
        return ("small-cached-prefix",
                "writes are %.0f%% of input with reads at 0, under the floor of "
                "%.0f%%. Something is being cached and never matched, and it is "
                "a minority of the prompt rather than the prefix."
                % (share * 100, share_floor * 100))

    runs = churn_runs(rows, share_floor, read_floor)
    longest = max(runs, key=len) if runs else []
    if len(longest) >= min_run:
        return ("prefix-churn",
                "writes are %.0f%% of input with reads at 0; longest run %d "
                "adjacent minute(s) from %s to %s. The entry written at the "
                "start of that run was still alive at the end and was never "
                "matched, so the prefix differs on every call."
                % (share * 100, len(longest), longest[0]["minute"],
                   longest[-1]["minute"]))

    gap = gap_profile(rows)
    if gap is not None and gap > min_run:
        return ("gap-driven-misses",
                "writes are %.0f%% of input with reads at 0, and the writing "
                "minutes sit a median of %.0f minute(s) apart"
                % (share * 100, gap))

    return ("intermittent-misses",
            "writes are %.0f%% of input with reads at 0, and the longest run of "
            "adjacent writing minutes is %d, under the floor of %d. Suggestive "
            "and not conclusive: widen the window."
            % (share * 100, len(longest), min_run))


def repair_lines():
    """The invalidator hunt, in cache order. Pure."""
    return [
        "hunt the invalidator in cache order: tools, then system, then "
        "messages. A change to the tools invalidates all three.",
        "the usual suspects are a clock (datetime.now in a system prompt), a "
        "tool list built from an unordered dict, a per-request id, a per-user "
        "preamble placed before the breakpoint, and an option toggled per call "
        "such as tool_choice, citations, web search or reasoning effort.",
        "move each one strictly after the last cache_control breakpoint, then "
        "re-read these same minute buckets. The runs should break up before "
        "the totals move.",
    ]


def window_start(minutes):
    """Floor to the minute: starting_at has to 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("--min-run", type=int, default=5,
                    help="adjacent writing minutes with no read that make a "
                         "finding (default 5, the 5m TTL)")
    ap.add_argument("--share-floor", type=float, default=0.5,
                    help="write share of input above which the prefix, rather "
                         "than a fragment of it, is being rewritten")
    ap.add_argument("--show-all", action="store_true",
                    help="also print series that are behaving")
    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(30, min(int(args.minutes), 1440))
    session = requests.Session()
    session.headers.update({"x-api-key": admin, "anthropic-version": VERSION})

    buckets = read_buckets(session, "/organizations/usage_report/messages", {
        "starting_at": window_start(minutes),
        "bucket_width": "1m",
        "limit": minutes,
        "group_by[]": ["api_key_id", "model"],
    })

    series = rows_by_key(buckets)
    if not series:
        log.info("no messages usage in the last %d minute(s)", minutes)
        return 0

    checked = 0
    bad = 0
    for ident in sorted(series):
        rows = series[ident]
        state, detail = classify(rows, args.min_run, args.share_floor)
        checked += 1
        line = "%-20s %s / %s  %s" % (state, ident[0], ident[1], detail)

        if state in FINDINGS:
            bad += 1
            log.warning(line)
            _, ttl = ttl_split(totals(rows))
            log.warning("  %s", ttl)
            log.warning("  note: grouped by key and model. A key serving many "
                        "tenants with a per tenant prefix writes constantly and "
                        "correctly; this finding is strongest on a key with one "
                        "workload.")
            for repair in repair_lines():
                log.warning("  repair: %s", repair)
        else:
            note = handoff(state)
            if note:
                log.info(line)
                log.info("  %s", note)
            elif args.show_all or state == "intermittent-misses":
                log.info(line)

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


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

/**
 * Find Anthropic keys whose cache is rewritten on every call and never read.
 *
 * Read only. One GET against the Admin API, which needs an Admin API key
 * (sk-ant-admin...). A workspace key is rejected by /v1/organizations/.
 *
 * Totals cannot separate this from two neighbouring problems, so the evidence
 * is spacing: a run of adjacent one-minute buckets that each write and never
 * read is longer than the entry's TTL, so the entry was alive and unmatched.
 * Caching switched off, and caching read but not read enough, are named and
 * handed to their own notes.
 */
const API = 'https://api.anthropic.com/v1';
const VERSION = '2023-06-01';

const FINDINGS = new Set(['prefix-churn']);

/** Read a usage field as an integer. Pure. Missing and unreadable both mean 0. */
export function readInt(value) {
  const n = Number(value ?? 0);
  return Number.isFinite(n) ? Math.trunc(n) : 0;
}

/** Normalise a timestamp to a UTC minute key. Pure. Null if unreadable. */
export function minuteKey(stamp) {
  if (typeof stamp === 'boolean') return null;
  if (typeof stamp === 'number' && Number.isFinite(stamp)) {
    const when = new Date(Math.trunc(stamp) * 1000);
    if (Number.isNaN(when.getTime())) return null;
    return `${when.toISOString().slice(0, 16)}Z`;
  }
  const text = String(stamp ?? '').trim().replace(' ', 'T');
  if (text.length < 16) return null;
  const head = text.slice(0, 16);
  if (!/^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}$/.test(head)) return null;
  return `${head}Z`;
}

/**
 * Minutes since the epoch. Pure. Null if unreadable.
 * Adjacency has to be integer arithmetic: string comparison puts 14:59 and
 * 15:00 two apart and quietly halves every run that crosses an hour.
 */
export function minuteIndex(stamp) {
  const key = minuteKey(stamp);
  if (key === null) return null;
  const when = Date.parse(`${key.slice(0, 16)}:00Z`);
  if (Number.isNaN(when)) return null;
  return Math.floor(when / 60000);
}

/** Per api_key_id and model, one row per minute, sorted. Pure. */
export function rowsByKey(buckets) {
  const merged = new Map();
  for (const bucket of buckets ?? []) {
    const stamp = bucket?.starting_at ?? bucket?.start_time;
    const key = minuteKey(stamp);
    const index = minuteIndex(stamp);
    if (key === null || index === null) continue;
    for (const result of bucket?.results ?? []) {
      if (!result || typeof result !== 'object') continue;
      const ident = `${result.api_key_id ?? 'unknown'}\t${result.model ?? 'unknown'}`;
      const cell = `${ident}\t${index}`;
      if (!merged.has(cell)) {
        merged.set(cell, { ident, minute: key, index, uncached: 0,
                           write5m: 0, write1h: 0, reads: 0 });
      }
      const row = merged.get(cell);
      const creation = result.cache_creation ?? {};
      row.uncached += readInt(result.uncached_input_tokens);
      row.write5m += readInt(creation.ephemeral_5m_input_tokens);
      row.write1h += readInt(creation.ephemeral_1h_input_tokens);
      row.reads += readInt(result.cache_read_input_tokens);
    }
  }
  const out = new Map();
  for (const row of merged.values()) {
    if (!out.has(row.ident)) out.set(row.ident, []);
    out.get(row.ident).push(row);
  }
  for (const rows of out.values()) rows.sort((a, b) => a.index - b.index);
  return out;
}

/** Cache creation tokens in one minute, both TTLs. Pure. */
export function writes(row) {
  return readInt(row?.write5m) + readInt(row?.write1h);
}

/** Share of a minute's input written as a fresh entry. Pure. Null when idle. */
export function writeShare(row) {
  const total = readInt(row?.uncached) + writes(row);
  if (total <= 0) return null;
  return writes(row) / total;
}

/** Sum a series, and count the minutes that carried any traffic. Pure. */
export function totals(rows) {
  const out = { uncached: 0, write5m: 0, write1h: 0, reads: 0, active: 0 };
  for (const row of rows ?? []) {
    out.uncached += readInt(row?.uncached);
    out.write5m += readInt(row?.write5m);
    out.write1h += readInt(row?.write1h);
    out.reads += readInt(row?.reads);
    if (readInt(row?.uncached) + writes(row) + readInt(row?.reads) > 0) out.active += 1;
  }
  out.writes = out.write5m + out.write1h;
  return out;
}

/**
 * Maximal runs of adjacent minutes that wrote and never read. Pure.
 * The finding. A 5 minute entry written at the start of a five minute run was
 * still alive at the end of it, so nothing but a moving prefix explains a run.
 */
export function churnRuns(rows, shareFloor = 0.5, readFloor = 0.01) {
  const runs = [];
  let current = [];
  for (const row of rows ?? []) {
    const made = writes(row);
    const share = writeShare(row);
    const churning = made > 0 && share !== null && share >= shareFloor
      && readInt(row?.reads) <= made * readFloor;
    if (!churning) {
      if (current.length > 0) { runs.push(current); current = []; }
      continue;
    }
    if (current.length > 0 && readInt(row?.index) === readInt(current[current.length - 1]?.index) + 1) {
      current.push(row);
    } else {
      if (current.length > 0) runs.push(current);
      current = [row];
    }
  }
  if (current.length > 0) runs.push(current);
  return runs;
}

/**
 * Median gap in minutes between minutes that wrote. Pure. Null under two.
 * The alternative explanation as a number: traffic slower than the TTL writes
 * isolated entries that expire before anything can read them.
 */
export function gapProfile(rows) {
  const indices = (rows ?? []).filter((r) => writes(r) > 0)
    .map((r) => readInt(r?.index)).sort((a, b) => a - b);
  if (indices.length < 2) return null;
  const gaps = [];
  for (let i = 0; i < indices.length - 1; i += 1) gaps.push(indices[i + 1] - indices[i]);
  gaps.sort((a, b) => a - b);
  const middle = Math.floor(gaps.length / 2);
  if (gaps.length % 2) return gaps[middle];
  return (gaps[middle - 1] + gaps[middle]) / 2;
}

/** Which TTL the writes were bought at. Pure. Returns [state, detail]. */
export function ttlSplit(sums) {
  const five = readInt(sums?.write5m);
  const hour = readInt(sums?.write1h);
  if (five + hour <= 0) {
    return ['no-writes', 'nothing was written to the cache in this window'];
  }
  if (hour > five) {
    return ['1h-dominant',
      'the writes are mostly 1 hour entries at 2x base input, so each one was ' +
      'alive for sixty minutes and never matched in any of them'];
  }
  if (five > hour) {
    return ['5m-dominant',
      'the writes are 5 minute entries at 1.25x base input, so any run longer ' +
      'than five minutes outlived calls that never matched it'];
  }
  return ['mixed', 'the writes are split evenly between the 5 minute and 1 hour TTLs'];
}

/** Which note owns this shape, when it is not this one. Pure. */
export function handoff(state) {
  if (state === 'caching-off') {
    return 'no writes and no reads anywhere: caching was never switched on for ' +
      'this key. Read the prompt-caching-never-used note; the loss there is a ' +
      'discount not taken rather than a surcharge paid.';
  }
  if (state === 'cache-is-read') {
    return 'entries are being matched, so the prefix is stable enough to hit. ' +
      'Whether it hits often enough to pay for the write premium is the ' +
      'write-to-read ratio, which is the cache-writes-with-no-reads note.';
  }
  if (state === 'gap-driven-misses') {
    return 'the writing minutes are isolated rather than adjacent, so each ' +
      'entry plausibly expired before the next call arrived. That is arrival ' +
      'rate against TTL, and it is the cache-writes-with-no-reads note rather ' +
      'than this one.';
  }
  return '';
}

/** Classify one key and model series. Pure. Returns [state, detail]. */
export function classify(rows, minRun = 5, shareFloor = 0.5, readFloor = 0.01,
                         minActive = 10) {
  const sums = totals(rows);
  if (sums.active < minActive) {
    return ['too-little-traffic',
      `${sums.active} active minute(s), under the floor of ${minActive}. ` +
      'Nothing can be said about spacing with fewer.'];
  }

  if (sums.writes === 0 && sums.reads === 0) {
    return ['caching-off',
      `${sums.uncached} uncached input token(s), no cache writes and no cache reads`];
  }
  if (sums.writes === 0) {
    return ['reads-only',
      `${sums.reads} cache read(s) and no writes in this window: the entries ` +
      'were written before it started'];
  }
  if (sums.reads > sums.writes * readFloor) {
    return ['cache-is-read',
      `${sums.reads} cache read token(s) against ${sums.writes} written`];
  }

  const share = sums.writes / (sums.uncached + sums.writes);
  if (share < shareFloor) {
    return ['small-cached-prefix',
      `writes are ${(share * 100).toFixed(0)}% of input with reads at 0, under ` +
      `the floor of ${(shareFloor * 100).toFixed(0)}%. Something is being ` +
      'cached and never matched, and it is a minority of the prompt rather ' +
      'than the prefix.'];
  }

  const runs = churnRuns(rows, shareFloor, readFloor);
  let longest = [];
  for (const run of runs) if (run.length > longest.length) longest = run;
  if (longest.length >= minRun) {
    return ['prefix-churn',
      `writes are ${(share * 100).toFixed(0)}% of input with reads at 0; ` +
      `longest run ${longest.length} adjacent minute(s) from ` +
      `${longest[0].minute} to ${longest[longest.length - 1].minute}. The ` +
      'entry written at the start of that run was still alive at the end and ' +
      'was never matched, so the prefix differs on every call.'];
  }

  const gap = gapProfile(rows);
  if (gap !== null && gap > minRun) {
    return ['gap-driven-misses',
      `writes are ${(share * 100).toFixed(0)}% of input with reads at 0, and ` +
      `the writing minutes sit a median of ${gap.toFixed(0)} minute(s) apart`];
  }

  return ['intermittent-misses',
    `writes are ${(share * 100).toFixed(0)}% of input with reads at 0, and the ` +
    `longest run of adjacent writing minutes is ${longest.length}, under the ` +
    `floor of ${minRun}. Suggestive and not conclusive: widen the window.`];
}

/** The invalidator hunt, in cache order. Pure. */
export function repairLines() {
  return [
    'hunt the invalidator in cache order: tools, then system, then messages. ' +
    'A change to the tools invalidates all three.',
    'the usual suspects are a clock (datetime.now in a system prompt), a tool ' +
    'list built from an unordered dict, a per-request id, a per-user preamble ' +
    'placed before the breakpoint, and an option toggled per call such as ' +
    'tool_choice, citations, web search or reasoning effort.',
    'move each one strictly after the last cache_control breakpoint, then ' +
    're-read these same minute buckets. The runs should break up before the ' +
    'totals move.',
  ];
}

function windowStart(minutes) {
  const now = new Date();
  now.setUTCSeconds(0, 0);
  return `${new Date(now.getTime() - minutes * 60000).toISOString().slice(0, 19)}Z`;
}

async function get(key, path, params) {
  const url = new URL(API + path);
  for (const [k, v] of Object.entries(params ?? {})) {
    if (Array.isArray(v)) v.forEach((item) => url.searchParams.append(k, item));
    else url.searchParams.set(k, String(v));
  }
  const res = await fetch(url, {
    headers: { 'x-api-key': key, '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 ${path}`);
  return res.json();
}

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

async function main() {
  const admin = process.env.ANTHROPIC_ADMIN_KEY;
  if (!admin) {
    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(30, Math.min(Number(process.env.MINUTES ?? 240), 1440));
  const minRun = Number(process.env.MIN_RUN ?? 5);
  const shareFloor = Number(process.env.SHARE_FLOOR ?? 0.5);
  const showAll = process.env.SHOW_ALL === '1';

  const buckets = [];
  for await (const bucket of readBuckets(admin, '/organizations/usage_report/messages', {
    starting_at: windowStart(minutes),
    bucket_width: '1m',
    limit: minutes,
    'group_by[]': ['api_key_id', 'model'],
  })) buckets.push(bucket);

  const series = rowsByKey(buckets);
  if (series.size === 0) {
    console.log(`no messages usage in the last ${minutes} minute(s)`);
    return;
  }

  let checked = 0;
  let bad = 0;
  for (const ident of [...series.keys()].sort()) {
    const rows = series.get(ident);
    const [state, detail] = classify(rows, minRun, shareFloor);
    checked += 1;
    const line = `${state.padEnd(20)} ${ident.replace('\t', ' / ')}  ${detail}`;

    if (FINDINGS.has(state)) {
      bad += 1;
      console.warn(line);
      const [, ttl] = ttlSplit(totals(rows));
      console.warn(`  ${ttl}`);
      console.warn('  note: grouped by key and model. A key serving many tenants ' +
                   'with a per tenant prefix writes constantly and correctly; ' +
                   'this finding is strongest on a key with one workload.');
      for (const repair of repairLines()) console.warn(`  repair: ${repair}`);
    } else {
      const note = handoff(state);
      if (note) {
        console.log(line);
        console.log(`  ${note}`);
      } else if (showAll || state === 'intermittent-misses') {
        console.log(line);
      }
    }
  }

  console.log(`${checked} key/model series 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_cache_prefix_churn import (churn_runs, classify, gap_profile,
                                         handoff, minute_index, minute_key,
                                         rows_by_key, totals, ttl_split,
                                         write_share, writes)

BASE = minute_index("2026-08-31T10:00Z")


def minute(offset, uncached=100_000, write5m=0, write1h=0, reads=0):
    index = BASE + offset
    hour, rest = divmod(offset, 60)
    return {"minute": "2026-08-31T%02d:%02dZ" % (10 + hour, rest), "index": index,
            "uncached": uncached, "write5m": write5m, "write1h": write1h,
            "reads": reads}


# Every call writes: a hundred and twenty adjacent minutes, never a read.
CHURN = [minute(i, write5m=500_000) for i in range(120)]
# Byte-identical totals, six writing minutes twenty apart. Traffic slower than
# the TTL, which is a different note.
SLOW = [minute(i, write5m=10_000_000 if i % 20 == 0 else 0) for i in range(120)]


def test_a_write_in_every_adjacent_minute_and_never_a_read():
    # The note in one assertion. The run is longer than the TTL, so the entry
    # written at 10:00 was alive at 10:04 and the call at 10:04 wrote another.
    sums = totals(CHURN)
    assert sums["writes"] == 60_000_000 and sums["uncached"] == 12_000_000
    assert sums["reads"] == 0 and sums["active"] == 120
    assert round(write_share(CHURN[0]), 4) == 0.8333

    runs = churn_runs(CHURN)
    assert len(runs) == 1 and len(runs[0]) == 120

    state, detail = classify(CHURN)
    assert state == "prefix-churn"
    assert "longest run 120 adjacent minute(s)" in detail
    assert "from 2026-08-31T10:00Z to 2026-08-31T11:59Z" in detail
    assert ttl_split(sums)[0] == "5m-dominant"


def test_identical_totals_spaced_out_are_a_different_note():
    # The pair. Same writes, same uncached input, same zero reads, and the
    # opposite conclusion. Nothing an hourly bucket can see separates these.
    assert totals(SLOW)["writes"] == totals(CHURN)["writes"]
    assert totals(SLOW)["uncached"] == totals(CHURN)["uncached"]
    assert totals(SLOW)["reads"] == totals(CHURN)["reads"] == 0

    assert max(len(r) for r in churn_runs(SLOW)) == 1
    assert gap_profile(SLOW) == 20.0
    assert gap_profile(CHURN) == 1.0

    state, detail = classify(SLOW)
    assert state == "gap-driven-misses"
    assert "median of 20 minute(s) apart" in detail
    assert "cache-writes-with-no-reads" in handoff(state)


def test_reads_anywhere_hand_the_finding_to_the_ratio_note():
    warm = [minute(i, write5m=500_000 if i == 0 else 0,
                   reads=400_000 if i else 0) for i in range(120)]
    state, detail = classify(warm)
    assert state == "cache-is-read"
    assert "against 500000 written" in detail
    assert "write-to-read ratio" in handoff(state)


def test_no_writes_and_no_reads_is_the_never_switched_on_note():
    off = [minute(i) for i in range(120)]
    state, detail = classify(off)
    assert state == "caching-off"
    assert "no cache writes and no cache reads" in detail
    assert "prompt-caching-never-used" in handoff(state)
    assert ttl_split(totals(off))[0] == "no-writes"
    reads_only = [minute(i, reads=400_000) for i in range(120)]
    assert classify(reads_only)[0] == "reads-only"


def test_a_minority_cached_fragment_is_not_the_prefix():
    small = [minute(i, uncached=900_000, write5m=100_000) for i in range(120)]
    state, detail = classify(small)
    assert state == "small-cached-prefix"
    assert "writes are 10% of input" in detail
    assert handoff(state) == ""


def test_an_hour_long_ttl_makes_the_same_run_worse():
    hourly = [minute(i, write1h=500_000) for i in range(120)]
    state, _ = classify(hourly)
    assert state == "prefix-churn"
    ttl_state, ttl_detail = ttl_split(totals(hourly))
    assert ttl_state == "1h-dominant"
    assert "2x base input" in ttl_detail
    assert ttl_split({"write5m": 10, "write1h": 10})[0] == "mixed"


def test_a_run_crossing_an_hour_boundary_is_not_broken_in_half():
    # 10:57 through 11:02. Comparing the minute strings puts 10:59 and 11:00
    # sixty apart and reports two runs of three.
    crossing = [minute(i, write5m=500_000) for i in range(57, 63)]
    assert [r["minute"] for r in crossing][:4] == [
        "2026-08-31T10:57Z", "2026-08-31T10:58Z", "2026-08-31T10:59Z",
        "2026-08-31T11:00Z"]
    runs = churn_runs(crossing)
    assert len(runs) == 1 and len(runs[0]) == 6
    assert minute_index("2026-08-31T11:00Z") - minute_index("2026-08-31T10:59Z") == 1


def test_the_nested_cache_creation_object_is_actually_read():
    buckets = [{"starting_at": "2026-08-31T10:0%dZ" % i,
                "results": [{"api_key_id": "apikey_01Ab", "model": "claude-opus-5",
                             "uncached_input_tokens": 100_000,
                             "cache_read_input_tokens": 0,
                             "cache_creation": {"ephemeral_5m_input_tokens": 500_000,
                                                "ephemeral_1h_input_tokens": 0}}]}
               for i in range(6)]
    series = rows_by_key(buckets)
    rows = series[("apikey_01Ab", "claude-opus-5")]
    assert len(rows) == 6
    assert writes(rows[0]) == 500_000
    assert [r["index"] for r in rows] == sorted(r["index"] for r in rows)
    state, _ = classify(rows, min_active=6)
    assert state == "prefix-churn"


def test_thin_and_unreadable_windows_produce_no_verdict():
    assert classify([minute(i, write5m=500_000) for i in range(4)])[0] == "too-little-traffic"
    assert classify([])[0] == "too-little-traffic"
    assert classify(None)[0] == "too-little-traffic"
    assert write_share({"uncached": 0, "write5m": 0, "write1h": 0}) is None
    assert gap_profile([]) is None
    assert minute_key("nonsense") is None
    assert minute_index(None) is None
    assert rows_by_key([{"starting_at": "bad", "results": []}]) == {}

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { churnRuns, classify, gapProfile, handoff, minuteIndex, minuteKey,
         rowsByKey, totals, ttlSplit, writeShare, writes }
  from './anthropic-cache-prefix-churn.mjs';

const BASE = minuteIndex('2026-08-31T10:00Z');

const minute = (offset, { uncached = 100000, write5m = 0, write1h = 0,
                          reads = 0 } = {}) => {
  const hour = Math.floor(offset / 60);
  const rest = offset % 60;
  const pad = (n) => String(n).padStart(2, '0');
  return { minute: `2026-08-31T${pad(10 + hour)}:${pad(rest)}Z`,
           index: BASE + offset, uncached, write5m, write1h, reads };
};

const CHURN = Array.from({ length: 120 }, (_, i) => minute(i, { write5m: 500000 }));
const SLOW = Array.from({ length: 120 },
  (_, i) => minute(i, { write5m: i % 20 === 0 ? 10000000 : 0 }));

test('a write in every adjacent minute and never a read', () => {
  const sums = totals(CHURN);
  assert.equal(sums.writes, 60000000);
  assert.equal(sums.uncached, 12000000);
  assert.equal(sums.reads, 0);
  assert.equal(sums.active, 120);
  assert.equal(Number(writeShare(CHURN[0]).toFixed(4)), 0.8333);

  const runs = churnRuns(CHURN);
  assert.equal(runs.length, 1);
  assert.equal(runs[0].length, 120);

  const [state, detail] = classify(CHURN);
  assert.equal(state, 'prefix-churn');
  assert.match(detail, /longest run 120 adjacent minute/);
  assert.match(detail, /from 2026-08-31T10:00Z to 2026-08-31T11:59Z/);
  assert.equal(ttlSplit(sums)[0], '5m-dominant');
});

test('identical totals spaced out are a different note', () => {
  assert.equal(totals(SLOW).writes, totals(CHURN).writes);
  assert.equal(totals(SLOW).uncached, totals(CHURN).uncached);
  assert.equal(totals(SLOW).reads, 0);
  assert.equal(totals(CHURN).reads, 0);

  assert.equal(Math.max(...churnRuns(SLOW).map((r) => r.length)), 1);
  assert.equal(gapProfile(SLOW), 20);
  assert.equal(gapProfile(CHURN), 1);

  const [state, detail] = classify(SLOW);
  assert.equal(state, 'gap-driven-misses');
  assert.match(detail, /median of 20 minute\(s\) apart/);
  assert.match(handoff(state), /cache-writes-with-no-reads/);
});

test('reads anywhere hand the finding to the ratio note', () => {
  const warm = Array.from({ length: 120 }, (_, i) => minute(i, {
    write5m: i === 0 ? 500000 : 0, reads: i ? 400000 : 0 }));
  const [state, detail] = classify(warm);
  assert.equal(state, 'cache-is-read');
  assert.match(detail, /against 500000 written/);
  assert.match(handoff(state), /write-to-read ratio/);
});

test('no writes and no reads is the never switched on note', () => {
  const off = Array.from({ length: 120 }, (_, i) => minute(i));
  const [state, detail] = classify(off);
  assert.equal(state, 'caching-off');
  assert.match(detail, /no cache writes and no cache reads/);
  assert.match(handoff(state), /prompt-caching-never-used/);
  assert.equal(ttlSplit(totals(off))[0], 'no-writes');
  const readsOnly = Array.from({ length: 120 }, (_, i) => minute(i, { reads: 400000 }));
  assert.equal(classify(readsOnly)[0], 'reads-only');
});

test('a minority cached fragment is not the prefix', () => {
  const small = Array.from({ length: 120 },
    (_, i) => minute(i, { uncached: 900000, write5m: 100000 }));
  const [state, detail] = classify(small);
  assert.equal(state, 'small-cached-prefix');
  assert.match(detail, /writes are 10% of input/);
  assert.equal(handoff(state), '');
});

test('an hour long ttl makes the same run worse', () => {
  const hourly = Array.from({ length: 120 }, (_, i) => minute(i, { write1h: 500000 }));
  assert.equal(classify(hourly)[0], 'prefix-churn');
  const [ttlState, ttlDetail] = ttlSplit(totals(hourly));
  assert.equal(ttlState, '1h-dominant');
  assert.match(ttlDetail, /2x base input/);
  assert.equal(ttlSplit({ write5m: 10, write1h: 10 })[0], 'mixed');
});

test('a run crossing an hour boundary is not broken in half', () => {
  const crossing = Array.from({ length: 6 },
    (_, i) => minute(57 + i, { write5m: 500000 }));
  assert.deepEqual(crossing.slice(0, 4).map((r) => r.minute),
    ['2026-08-31T10:57Z', '2026-08-31T10:58Z', '2026-08-31T10:59Z',
     '2026-08-31T11:00Z']);
  const runs = churnRuns(crossing);
  assert.equal(runs.length, 1);
  assert.equal(runs[0].length, 6);
  assert.equal(minuteIndex('2026-08-31T11:00Z') - minuteIndex('2026-08-31T10:59Z'), 1);
});

test('the nested cache creation object is actually read', () => {
  const buckets = Array.from({ length: 6 }, (_, i) => ({
    starting_at: `2026-08-31T10:0${i}Z`,
    results: [{ api_key_id: 'apikey_01Ab', model: 'claude-opus-5',
                uncached_input_tokens: 100000,
                cache_read_input_tokens: 0,
                cache_creation: { ephemeral_5m_input_tokens: 500000,
                                  ephemeral_1h_input_tokens: 0 } }],
  }));
  const series = rowsByKey(buckets);
  const rows = [...series.values()][0];
  assert.equal(rows.length, 6);
  assert.equal(writes(rows[0]), 500000);
  assert.equal(classify(rows, 5, 0.5, 0.01, 6)[0], 'prefix-churn');
});

test('thin and unreadable windows produce no verdict', () => {
  const thin = Array.from({ length: 4 }, (_, i) => minute(i, { write5m: 500000 }));
  assert.equal(classify(thin)[0], 'too-little-traffic');
  assert.equal(classify([])[0], 'too-little-traffic');
  assert.equal(classify(null)[0], 'too-little-traffic');
  assert.equal(writeShare({ uncached: 0, write5m: 0, write1h: 0 }), null);
  assert.equal(gapProfile([]), null);
  assert.equal(minuteKey('nonsense'), null);
  assert.equal(minuteIndex(null), null);
  assert.equal(rowsByKey([{ starting_at: 'bad', results: [] }]).size, 0);
});