pricing: price Claude's 1-hour cache writes at 2x input, not 1.25x

tests/test_cache_write_1h.py

@@ -0,0 +1,123 @@
+"""A cache write is billed by its lifetime: Anthropic charges 1.25x input for the 5-minute cache
+and 2x for the 1-hour one, and a message's usage says how its cache writes split between them
+(`cache_creation: {ephemeral_5m_input_tokens, ephemeral_1h_input_tokens}`). Claude Code on a
+subscription writes to the 1-hour cache, so pricing every cache write at 1.25x understates such a
+session."""
+import json
+
+from tokeneconomizer import pricing, trace
+from tokeneconomizer.plugin import accounting
+
+
+def _row(cache_write=1000, one_hour=None, model="claude-opus-4-8"):
+    row = {"input_tokens": 100, "cache_read_input_tokens": 10_000,
+           "cache_creation_input_tokens": cache_write, "output_tokens": 10, "model": model}
+    if one_hour is not None:
+        row["cache_creation_1h_input_tokens"] = one_hour
+    return row
+
+
+BASE = 100 * 1.0 + 10_000 * 0.10 + 10 * 5.0          # everything but the cache writes
+
+
+# ---- pricing ---------------------------------------------------------------------------------
+def test_one_hour_cache_writes_are_priced_at_twice_input():
+    assert pricing.effective_tokens(_row(1000, one_hour=1000)) == BASE + 1000 * 2.0
+
+
+def test_a_mixed_row_prices_each_share_at_its_own_rate():
+    assert pricing.effective_tokens(_row(1000, one_hour=400)) == BASE + 600 * 1.25 + 400 * 2.0
+
+
+def test_a_row_without_the_split_is_priced_as_before():
+    assert pricing.effective_tokens(_row(1000)) == BASE + 1000 * 1.25
+    assert pricing.effective_tokens(_row(1000, one_hour=0)) == BASE + 1000 * 1.25
+
+
+def test_a_claude_id_outside_the_table_gets_the_one_hour_rate_too():
+    row = _row(1000, one_hour=1000, model="claude-opus-5-5")
+    assert pricing.effective_tokens(row) == BASE + 1000 * 2.0
+
+
+def test_the_one_hour_share_is_never_more_than_the_cache_writes():
+    assert pricing.cache_write_1h_tokens(_row(1000, one_hour=5000)) == 1000
+    assert pricing.cache_write_1h_tokens(_row(1000, one_hour=-5)) == 0
+    assert pricing.cache_write_1h_tokens(_row(1000, one_hour="x")) == 0
+    assert pricing.cache_write_1h_tokens(None) == 0
+
+
+def test_a_weight_table_without_a_one_hour_rate_uses_its_cache_write_rate():
+    # gpt-5.6-sol has no 1-hour entry: its cache writes stay at 1.25x whatever the row says.
+    row = _row(1000, one_hour=1000, model="gpt-5.6-sol")
+    assert pricing.effective_tokens(row) == 100 + 10_000 * 0.10 + 1000 * 1.25 + 10 * 6.0
+    explicit = dict(input=1.0, cache_read=0.0, cache_write=1.0, output=1.0)
+    assert pricing.effective_tokens(_row(1000, one_hour=1000), weights=explicit) == 100 + 1000 + 10
+
+
+def test_the_split_is_read_from_an_anthropic_usage_object():
+    usage = {"cache_creation_input_tokens": 248,
+             "cache_creation": {"ephemeral_5m_input_tokens": 148, "ephemeral_1h_input_tokens": 100}}
+    assert pricing.cache_write_1h_tokens(usage) == 100
+    assert pricing.cache_write_1h_tokens({"cache_creation_input_tokens": 248,
+                                          "cache_creation": "not an object"}) == 0
+
+
+# ---- Claude Code transcripts -----------------------------------------------------------------
+def _message(mid, cache_write, split=None, blocks=2):
+    usage = {"input_tokens": 10, "cache_read_input_tokens": 1000,
+             "cache_creation_input_tokens": cache_write, "output_tokens": 50}
+    if split is not None:
+        usage["cache_creation"] = split
+    return [{"type": "assistant", "message": {
+        "id": mid, "model": "claude-opus-4-8", "role": "assistant", "usage": usage,
+        "content": [{"type": "text", "text": f"block {b}"}]}} for b in range(blocks)]
+
+
+def _transcript(path, entries):
+    path.write_text("\n".join(json.dumps(e) for e in entries) + "\n")
+    return str(path)
+
+
+def test_a_claude_session_reports_its_one_hour_cache_writes(tmp_path):
+    p = _transcript(tmp_path / "s.jsonl", [
+        *_message("m1", 300, {"ephemeral_5m_input_tokens": 0, "ephemeral_1h_input_tokens": 300}),
+        *_message("m2", 200, {"ephemeral_5m_input_tokens": 150, "ephemeral_1h_input_tokens": 50}),
+        *_message("m3", 100),                                   # an older entry: no split
+    ])
+    u = trace.claude_session_usage(p)
+    assert u["cache_creation_input_tokens"] == 600              # each message once, as before
+    assert u["cache_creation_1h_input_tokens"] == 350
+
+
+def test_a_split_on_a_message_that_reports_no_cache_writes_adds_nothing(tmp_path):
+    # Seen in real transcripts: every count 0, yet a `cache_creation` split that is not.
+    p = _transcript(tmp_path / "s.jsonl", [
+        *_message("m1", 0, {"ephemeral_5m_input_tokens": 0, "ephemeral_1h_input_tokens": 2948}),
+        *_message("m2", 80, {"ephemeral_5m_input_tokens": 0, "ephemeral_1h_input_tokens": "x"}),
+    ])
+    u = trace.claude_session_usage(p)
+    assert u["cache_creation_input_tokens"] == 80
+    assert u["cache_creation_1h_input_tokens"] == 0
+
+
+# ---- the ledger ------------------------------------------------------------------------------
+def test_a_session_is_priced_with_its_one_hour_cache_writes(tmp_path):
+    p = _transcript(tmp_path / "s.jsonl", _message(
+        "m1", 1000, {"ephemeral_5m_input_tokens": 0, "ephemeral_1h_input_tokens": 1000}))
+    eff, priced_by = accounting.effective(trace.claude_session_usage(p))
+    assert priced_by == "model"
+    assert eff == 10 + 1000 * 0.10 + 1000 * 2.0 + 50 * 5.0
+
+
+def test_a_claude_side_call_is_priced_from_the_usage_object_it_returns():
+    usage = {"input_tokens": 10, "cache_read_input_tokens": 0, "output_tokens": 0,
+             "cache_creation_input_tokens": 1000, "model": "claude-haiku-4-5",
+             "cache_creation": {"ephemeral_5m_input_tokens": 0, "ephemeral_1h_input_tokens": 1000}}
+    assert accounting.effective(usage) == (10 + 1000 * 2.0, "model")
+
+
+def test_an_unpriced_model_keeps_the_fallback_cache_write_rate():
+    usage = {"input_tokens": 10, "cache_read_input_tokens": 0, "output_tokens": 0,
+             "cache_creation_input_tokens": 1000, "cache_creation_1h_input_tokens": 1000,
+             "model": "some-new-model"}
+    assert accounting.effective(usage) == (10 + 1000 * 1.25, "fallback")

tokeneconomizer/plugin/accounting.py

@@ -20,6 +20,9 @@ def effective(usage):
         return None, "no_usage"
     rec = {k: usage.get(k) for k in ("input_tokens", "cache_read_input_tokens",
                                      "cache_creation_input_tokens", "output_tokens", "model")}
+    # The 1-hour share of the cache writes, from a session reader's flat field or from an
+    # Anthropic `usage` object as a side call returns it.
+    rec["cache_creation_1h_input_tokens"] = pricing.cache_write_1h_tokens(usage)
     try:
         return pricing.effective_tokens(rec), "model"
     except ValueError:

tokeneconomizer/plugin/fold.py

@@ -488,6 +488,8 @@ def fold_locked(ctx, session_id, transcript_path=None, imported=False, span=None
             record.update({k: usage.get(k) for k in (
                 "model", "requests", "input_tokens", "cache_read_input_tokens",
                 "cache_creation_input_tokens", "output_tokens")})
+            if usage.get("cache_creation_1h_input_tokens"):
+                record["cache_creation_1h_input_tokens"] = usage["cache_creation_1h_input_tokens"]
             record["effective_tokens"], record["priced_by"] = eff, priced_by
         turns = sess.get("turns") or []
         record.update({

tokeneconomizer/pricing.py

@@ -25,6 +25,12 @@
     gpt-5.3-codex                    1.00     0.097       0.00        8.00
     gpt-5-codex (OpenAI direct)      1.00     0.10        1.00        8.00
 
+Anthropic bills a cache write by its lifetime: 1.25x for the 5-minute cache, 2x for the 1-hour
+one. A row that says how many of its cache-write tokens went to the 1-hour cache
+(`cache_creation_1h_input_tokens`, see cache_write_1h_tokens) has those priced at the table's
+`cache_write_1h`; a table without that entry, or a row without the field, prices every
+cache-write token at `cache_write` as before.
+
 Callers may override with an explicit effective_tokens(rec, weights=...).
 When the per-task breakdown is absent (older runs), effective_tokens() falls back to
 `tokens_total` at full weight so nothing crashes — call has_breakdown() to detect this.
@@ -46,14 +52,22 @@
 # `codex` run bills sol.
 MODEL_WEIGHTS = {
     # --- per-model price snapshot (2026-07-27) ------------------------------
-    "claude-haiku-4-5":        dict(input=1.0, cache_read=0.10,   cache_write=1.25,  output=5.0),
-    "claude-opus-4-6":         dict(input=1.0, cache_read=0.10,   cache_write=1.25,  output=5.0),
-    "claude-opus-4-7":         dict(input=1.0, cache_read=0.10,   cache_write=1.25,  output=5.0),
-    "claude-opus-4-8":         dict(input=1.0, cache_read=0.10,   cache_write=1.25,  output=5.0),
-    "claude-opus-5":           dict(input=1.0, cache_read=0.10,   cache_write=1.25,  output=5.0),
-    "claude-sonnet-4-5":       dict(input=1.0, cache_read=0.10,   cache_write=1.25,  output=5.0),
-    "claude-sonnet-4-6":       dict(input=1.0, cache_read=0.10,   cache_write=1.25,  output=5.0),
-    "claude-sonnet-5":         dict(input=1.0, cache_read=0.10,   cache_write=1.25,  output=5.0),
+    "claude-haiku-4-5":        dict(input=1.0, cache_read=0.10,   cache_write=1.25,  output=5.0,
+                                    cache_write_1h=2.0),
+    "claude-opus-4-6":         dict(input=1.0, cache_read=0.10,   cache_write=1.25,  output=5.0,
+                                    cache_write_1h=2.0),
+    "claude-opus-4-7":         dict(input=1.0, cache_read=0.10,   cache_write=1.25,  output=5.0,
+                                    cache_write_1h=2.0),
+    "claude-opus-4-8":         dict(input=1.0, cache_read=0.10,   cache_write=1.25,  output=5.0,
+                                    cache_write_1h=2.0),
+    "claude-opus-5":           dict(input=1.0, cache_read=0.10,   cache_write=1.25,  output=5.0,
+                                    cache_write_1h=2.0),
+    "claude-sonnet-4-5":       dict(input=1.0, cache_read=0.10,   cache_write=1.25,  output=5.0,
+                                    cache_write_1h=2.0),
+    "claude-sonnet-4-6":       dict(input=1.0, cache_read=0.10,   cache_write=1.25,  output=5.0,
+                                    cache_write_1h=2.0),
+    "claude-sonnet-5":         dict(input=1.0, cache_read=0.10,   cache_write=1.25,  output=5.0,
+                                    cache_write_1h=2.0),
     "gemini-2.5-pro":          dict(input=1.0, cache_read=0.096,  cache_write=0.0,   output=8.0),
     "gemini-3-flash-preview":  dict(input=1.0, cache_read=0.10,   cache_write=0.0,   output=6.0),
     "gemini-3.1-pro-preview":  dict(input=1.0, cache_read=0.10,   cache_write=0.0,   output=6.0),
@@ -101,7 +115,8 @@
 WEIGHTS_CLAUDE = {
     "input": 1.0,         # input_tokens = fresh/uncached prompt tokens
     "cache_read": 0.10,   # cache_read_input_tokens
-    "cache_write": 1.25,  # cache_creation_input_tokens
+    "cache_write": 1.25,  # cache_creation_input_tokens (the 5-minute cache)
+    "cache_write_1h": 2.0,  # the part of them written to the 1-hour cache
     "output": 5.0,        # output_tokens (completion)
 }
 # Back-compat: modules that read pricing.WEIGHTS directly (figures, costmodel) were
@@ -163,11 +178,32 @@ def has_breakdown(rec):
     return isinstance(rec, dict) and rec.get("input_tokens") is not None
 
 
+def cache_write_1h_tokens(usage):
+    """How many of a usage breakdown's cache-write tokens went to the 1-hour cache.
+
+    Reads the flat `cache_creation_1h_input_tokens` a session reader records, else the
+    `cache_creation.ephemeral_1h_input_tokens` of an Anthropic `usage` object as it arrives.
+    Never more than `cache_creation_input_tokens`, of which it is a part; 0 when neither says."""
+    if not isinstance(usage, dict):
+        return 0
+    n = usage.get("cache_creation_1h_input_tokens")
+    if n is None:
+        split = usage.get("cache_creation")
+        n = split.get("ephemeral_1h_input_tokens") if isinstance(split, dict) else None
+    try:
+        n = int(n or 0)
+        total = int(usage.get("cache_creation_input_tokens") or 0)
+    except (TypeError, ValueError):
+        return 0
+    return max(0, min(n, total))
+
+
 def effective_tokens(rec, weights=None):
     """Cache-aware effective cost of one task, in fresh-input-token equivalents.
 
     Uses the token breakdown (input / cache_read / cache_creation / output) when
-    present; otherwise falls back to `tokens_total` at full weight (legacy rows).
+    present, with the 1-hour share of the cache writes at its own price where the row
+    and the weight table both name it; otherwise falls back to `tokens_total` at full weight (legacy rows).
     The weight table is chosen per row (weights_for) from its agent/model unless an
     explicit `weights=` is passed. Returns 0 when neither breakdown nor total exists.
     """
@@ -177,8 +213,10 @@ def effective_tokens(rec, weights=None):
         cr = rec.get("cache_read_input_tokens") or 0
         cw = rec.get("cache_creation_input_tokens") or 0
         ot = rec.get("output_tokens") or 0
+        cw1h = cache_write_1h_tokens(rec)
         return (it * w["input"] + cr * w["cache_read"]
-                + cw * w["cache_write"] + ot * w["output"])
+                + (cw - cw1h) * w["cache_write"]
+                + cw1h * w.get("cache_write_1h", w["cache_write"]) + ot * w["output"])
     tt = rec.get("tokens_total") if isinstance(rec, dict) else None
     return tt if isinstance(tt, (int, float)) else 0
 

tokeneconomizer/trace/claude.py

@@ -20,6 +20,10 @@
 
 _CLAUDE_KEYS = ("input_tokens", "cache_read_input_tokens", "cache_creation_input_tokens",
                 "output_tokens")
+# The part of `cache_creation_input_tokens` written to the 1-hour cache, which Anthropic bills
+# at 2x input where the 5-minute cache is 1.25x. A message's usage splits its cache writes in
+# `cache_creation: {ephemeral_5m_input_tokens, ephemeral_1h_input_tokens}`.
+_CLAUDE_1H = "cache_creation_1h_input_tokens"
 CONVERSATION = ("assistant", "user")
 SYNTHETIC_MODEL = "<synthetic>"     # the model of an entry Claude Code wrote, not a model's reply
 
@@ -280,12 +284,26 @@ def claude_subagent_files(transcript_path):
     return sorted(glob.glob(os.path.join(stem, "subagents", "*.jsonl")))
 
 
+def _cache_write_1h(usage, counts):
+    """A message's 1-hour cache-write tokens, never more than the cache writes it reports: the
+    split is a breakdown of that total, and a notice with no usage of its own can carry one."""
+    split = usage.get("cache_creation")
+    if not isinstance(split, dict):
+        return 0
+    try:
+        n = int(split.get("ephemeral_1h_input_tokens") or 0)
+    except (TypeError, ValueError):
+        return 0
+    return max(0, min(n, counts["cache_creation_input_tokens"]))
+
+
 def claude_session_usage(transcript_path, include_subagents=True):
     """Token breakdown of a Claude Code session, de-duplicated by assistant message id.
 
-    Returns the four breakdown fields `pricing.effective_tokens` reads, plus `model` (the most
-    used one), `requests` (distinct assistant messages carrying usage, less the `<synthetic>`
-    ones Claude Code writes itself) and `files` read."""
+    Returns the four breakdown fields `pricing.effective_tokens` reads and the 1-hour share of
+    the cache writes (`cache_creation_1h_input_tokens`), plus `model` (the most used one),
+    `requests` (distinct assistant messages carrying usage, less the `<synthetic>` ones Claude
+    Code writes itself) and `files` read."""
     files = [transcript_path] + (claude_subagent_files(transcript_path)
                                  if include_subagents else [])
     by_id, models, anon = {}, {}, 0
@@ -313,10 +331,11 @@ def claude_session_usage(transcript_path, include_subagents=True):
                 mid = f"anon-{path}-{anon}"
             # Later entries of the same message carry the final counts; keep the last.
             by_id[mid] = {k: int(usage.get(k) or 0) for k in _CLAUDE_KEYS}
+            by_id[mid][_CLAUDE_1H] = _cache_write_1h(usage, by_id[mid])
             model = msg.get("model")
             if model:
                 models[model] = models.get(model, 0) + 1
-    out = {k: sum(u[k] for u in by_id.values()) for k in _CLAUDE_KEYS}
+    out = {k: sum(u[k] for u in by_id.values()) for k in (*_CLAUDE_KEYS, _CLAUDE_1H)}
     out["requests"] = len(by_id)
     out["model"] = max(models, key=models.get) if models else None
     out["files"] = len([f for f in files if os.path.exists(f)])