# Opt in to Cache diagnostics: send the beta header and thread the previous
# response id as previous_message_id (None on the first turn). The reason comes
# back on resp.diagnostics.cache_miss_reason.type — e.g. "system_changed".
resp = client.beta.messages.create(
    diagnostics={"previous_message_id": prev_id},
    betas=["cache-diagnosis-2026-04-07"],
    **kwargs,
)
prev_id = resp.id

#!/usr/bin/env python3
"""Generate real Claude Console cache-dashboard evidence: a HEALTHY run (write
once, then cheap reads) and a deliberately BROKEN read-side run (every call
writes, none reads), so the Console "Caching" pane
(platform.claude.com/usage/cache) shows an authentic before/after.

Why this exists
---------------
The `claude-cache-dashboard` tutorial is screenshot-driven. It needs a real
healthy dashboard state (present cache-read band, high read ratio, write
amortization > 1x) and a real read-side failure state (fat cache-write bands,
near-zero cache-read band, read ratio collapsing). This script produces both by
driving the live Anthropic API; the same calls feed the Console pane.

It is the evidence-capture companion to the healthy-only cost demo at
../../prompt-caching-explained/experiments/caching_cost_demo.py.

The two scenarios
-----------------
- healthy: one large document is the STABLE system prefix, marked with a cache
  breakpoint. The first call writes it; every later call reads it. Many reads
  over one write -> high read ratio and write amortization ~= (calls - 1).
- broken: the SAME document, same breakpoint, but each call prepends a unique
  per-call value (timestamp + uuid) to the FRONT of the system block. The
  byte-exact prefix match diverges at the very first token, so every call writes
  a fresh entry and none is ever read -> read ratio ~ 0. This is the single most
  common real read-side failure (a volatile value near the prompt front),
  reproduced on purpose.

What it prints
--------------
Per-call cache counters and a run summary (total write/read tokens, read ratio,
reads-per-write). This lets you confirm the API-level pattern immediately, before
the dashboard's hourly refresh catches up. The dashboard remains the source of
truth for the screenshots; these numbers are only a sanity check.

Run it
------
  export ANTHROPIC_API_KEY=sk-...           # workspace whose pane you will view
  pip install anthropic
  python dashboard_evidence.py healthy      # generate the working reference
  python dashboard_evidence.py broken       # generate the read-side failure

Options:
  --calls N          number of calls in the run (default 12)
  --model MODEL      model id (default claude-opus-4-8)
  <document>         trailing arg: a file path or plain-text URL to cache
                     (default: a public-domain Gutenberg text)

Cost: with the defaults (~15k-token doc, 12 calls, Opus 4.8) each run is well
under $1. Set --model claude-haiku-4-5 to run it ~5x cheaper; the dashboard
shape is identical, only the per-token price differs.
"""

from __future__ import annotations

import argparse
import urllib.request
import uuid
from datetime import datetime, timezone

# Keep the document comfortably above the model's minimum cacheable prefix while
# bounding cost. ~60k chars is ~15k tokens, safely above any current minimum.
MAX_DOC_CHARS = 60_000

# A public-domain text (Project Gutenberg). Override with a file path or URL.
DEFAULT_DOC_URL = "https://www.gutenberg.org/cache/epub/1342/pg1342.txt"

# A short rotating question set. The questions are the volatile tail (in
# `messages`, after the cached block); rotating them keeps each call's work
# distinct without touching the cached prefix.
QUESTIONS = [
    "In one sentence, what is the central conflict of this text?",
    "Name two main characters and describe each in one line.",
    "Quote one sentence that reflects the central theme.",
    "What is the tone of the opening, in one sentence?",
    "Summarize the setting in one sentence.",
    "What does the first chapter establish, in one sentence?",
]

MAX_TOKENS = 120  # short answers keep output cost small and focused


def load_document(source: str) -> str:
    """Load the document from a URL or local path; trim boilerplate and size."""
    if source.startswith(("http://", "https://")):
        req = urllib.request.Request(source, headers={"User-Agent": "cc-demo/1.0"})
        with urllib.request.urlopen(req, timeout=30) as resp:
            text = resp.read().decode("utf-8", errors="replace")
    else:
        with open(source, "r", encoding="utf-8", errors="replace") as fh:
            text = fh.read()

    # Strip Project Gutenberg header/footer so the cached prefix is the work,
    # not licensing text.
    start = text.find("*** START OF")
    if start != -1:
        text = text[text.find("\n", start) + 1 :]
    end = text.find("*** END OF")
    if end != -1:
        text = text[:end]

    text = text.strip()
    if len(text) > MAX_DOC_CHARS:
        text = text[:MAX_DOC_CHARS]
    return text


def counters(usage) -> tuple[int, int, int]:
    """(uncached_input, cache_write, cache_read) from a response usage object."""
    return (
        usage.input_tokens,
        getattr(usage, "cache_creation_input_tokens", 0) or 0,
        getattr(usage, "cache_read_input_tokens", 0) or 0,
    )


DIAGNOSTICS_BETA = "cache-diagnosis-2026-04-07"


def miss_reason(resp) -> str | None:
    """The cache_miss_reason type from a diagnosed response, or None.

    None covers both "no divergence" and "comparison still pending"
    (diagnostics or cache_miss_reason is null); a string is a *_changed /
    previous_message_not_found / unavailable type.
    """
    diag = getattr(resp, "diagnostics", None)
    if diag is None:
        return None
    reason = getattr(diag, "cache_miss_reason", None)
    return getattr(reason, "type", None) if reason is not None else None


def run(
    client, model: str, document: str, calls: int, broken: bool, diagnostics: bool
) -> None:
    """Drive `calls` sequential requests over the document.

    healthy (broken=False): the document is a byte-stable cached prefix -> first
    call writes, the rest read.
    broken  (broken=True):  a unique value is prepended to the FRONT of the
    system block each call -> the prefix diverges every call, so every call
    writes and none reads.

    With diagnostics=True, each call carries the cache-diagnosis beta header and
    threads the previous response id as previous_message_id, so the Console's
    Cache diagnostics panel populates. A broken run then reports system_changed
    (the divergence lives in the system block); a healthy run reports no
    divergence.
    """
    label = (
        "BROKEN (read-side failure)" if broken else "HEALTHY (write once, then reads)"
    )
    mode = " + diagnostics" if diagnostics else ""
    print(f"\n{label}{mode} — {calls} calls, model {model}")
    header = f"  {'#':>2}  {'uncached_in':>11}  {'cache_write':>11}  {'cache_read':>10}"
    if diagnostics:
        header += f"  {'miss_reason':>26}"
    print(header)

    tot_uncached = tot_write = tot_read = 0
    reason_counts: dict[str, int] = {}
    prev_id = None
    for i in range(calls):
        if broken:
            # A per-call volatile prefix guarantees a fresh write and zero reads.
            volatile = (
                f"Request {datetime.now(timezone.utc).isoformat()} id={uuid.uuid4()}"
            )
            text = f"{volatile}\n\n{document}"
        else:
            text = document
        system_block = {
            "type": "text",
            "text": text,
            "cache_control": {"type": "ephemeral"},
        }

        kwargs = dict(
            model=model,
            max_tokens=MAX_TOKENS,
            system=[system_block],
            messages=[{"role": "user", "content": QUESTIONS[i % len(QUESTIONS)]}],
        )
        if diagnostics:
            resp = client.beta.messages.create(
                diagnostics={"previous_message_id": prev_id},
                betas=[DIAGNOSTICS_BETA],
                **kwargs,
            )
        else:
            resp = client.messages.create(**kwargs)

        uncached, write, read = counters(resp.usage)
        tot_uncached += uncached
        tot_write += write
        tot_read += read
        line = f"  {i + 1:>2}  {uncached:>11}  {write:>11}  {read:>10}"
        if diagnostics:
            reason = miss_reason(resp) or "-"
            reason_counts[reason] = reason_counts.get(reason, 0) + 1
            line += f"  {reason:>26}"
            prev_id = resp.id
        print(line)

    total_in = tot_uncached + tot_write + tot_read
    read_ratio = (tot_read / total_in * 100) if total_in else 0.0
    amort = (tot_read / tot_write) if tot_write else 0.0
    print("  " + "-" * (66 if diagnostics else 40))
    print(
        f"  totals   write={tot_write:,}  read={tot_read:,}  uncached={tot_uncached:,}"
    )
    print(f"  cache read ratio (approx): {read_ratio:.1f}%")
    print(f"  write amortization (read tokens / write tokens): {amort:.2f}x")
    if diagnostics:
        summary = ", ".join(f"{k}={v}" for k, v in sorted(reason_counts.items()))
        print(f"  cache_miss_reason tally: {summary}")
        if broken and reason_counts.get("system_changed", 0) == 0:
            print(
                "  NOTE: expected system_changed on the broken run — none seen; if the"
            )
            print(
                "        tally is all '-', the comparison may have been pending (fast"
            )
            print("        responses) — re-run or add calls.")
    if broken and tot_read > 0:
        print(
            "  NOTE: expected 0 reads for the broken run — inspect before screenshotting."
        )
    if not broken and tot_read == 0:
        print(
            "  NOTE: expected reads > 0 for the healthy run — check the doc cleared the"
        )
        print("        model minimum and that calls landed inside the cache TTL.")
    print(
        "\n  The Console pane (platform.claude.com/usage/cache) reflects this run "
        "after its\n  hourly refresh. That pane, not these numbers, is the "
        "screenshot source of truth."
    )


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Generate Claude cache-dashboard evidence."
    )
    parser.add_argument("scenario", choices=["healthy", "broken"])
    parser.add_argument(
        "document",
        nargs="?",
        default=DEFAULT_DOC_URL,
        help="file path or plain-text URL to cache",
    )
    parser.add_argument("--calls", type=int, default=12)
    parser.add_argument("--model", default="claude-opus-4-8")
    parser.add_argument(
        "--diagnostics",
        action="store_true",
        help="enable the cache-diagnosis beta so the Console "
        "diagnostics panel populates (broken run -> "
        "system_changed)",
    )
    args = parser.parse_args()

    import anthropic  # deferred so --help works before the SDK is installed

    document = load_document(args.document)
    client = anthropic.Anthropic()
    approx = client.messages.count_tokens(
        model=args.model, messages=[{"role": "user", "content": document}]
    ).input_tokens

    print(f"Model:           {args.model}")
    print(f"Document source: {args.document}")
    print(f"Document size:   {len(document):,} chars (~{approx:,} tokens)")
    if approx < 4096:
        print("WARNING: document is below a safe minimum cacheable prefix — caching")
        print("         may not engage; use a larger document.")

    run(
        client,
        args.model,
        document,
        args.calls,
        broken=(args.scenario == "broken"),
        diagnostics=args.diagnostics,
    )


if __name__ == "__main__":
    main()