fix(pr-cost): price the claude lane by model family and cache TTL, and allow a corrected figure

skills/pr-cost/references/annotate.md

@@ -70,3 +70,11 @@ leaves every later `annotate` in the shell live.
 Privacy: do not paste prompts, diffs, or file contents into the PR comment.
 The collector already wraps a JSON payload. If annotate reports
 `"status": "duplicate"`, stop — this session/PR annotation already exists.
+
+The one reason to override that: the posted figure itself was wrong, and you
+are replacing it. `annotate --allow-duplicate` publishes the corrected
+payload and reports `"status": "corrected"`. It appends a second ledger row
+rather than editing the first, so the ledger keeps both what was published
+and what replaced it. The guard keys on `pr_url` + `session_id`, which is why
+a corrected figure for the same session looks like a replay without the flag.
+`from-hook` has no such flag — a hook that re-fires must stay idempotent.

skills/pr-cost/scripts/claude_session_usage.py

@@ -7,23 +7,63 @@
 import json
 import pathlib
 import sys
-from typing import Any
-
-# USD per million tokens: (input, output, cache_read, cache_write), matched by
-# the longest prefix of the lowercased model name. Public Anthropic list
-# prices; the claude lane reports input_tokens WITHOUT cache tokens, unlike
-# the codex lane.
-MODEL_RATES: dict[str, tuple[float, float, float, float]] = {
-    "claude-opus-4": (15.0, 75.0, 1.5, 18.75),
-    "claude-sonnet-4": (3.0, 15.0, 0.3, 3.75),
-    "claude-haiku-4": (1.0, 5.0, 0.1, 1.25),
-    "claude-3-7-sonnet": (3.0, 15.0, 0.3, 3.75),
-    "claude-3-5-sonnet": (3.0, 15.0, 0.3, 3.75),
-    "claude-3-5-haiku": (0.8, 4.0, 0.08, 1.0),
+from typing import Any, NamedTuple
+
+
+class Rates(NamedTuple):
+    """USD per million tokens for one model family."""
+
+    input: float
+    output: float
+    cache_read: float
+    cache_write_5m: float
+    cache_write_1h: float
+
+
+def _rates(input_rate: float, output_rate: float, *, cache_read: float | None = None) -> Rates:
+    """Derive the cache rates from Anthropic's published multipliers.
+
+    Reads cost 0.1x input; writes cost 1.25x for the 5-minute TTL and 2x for
+    the 1-hour TTL. Stating the multipliers once keeps a hand-typed write rate
+    from drifting away from its input rate -- and the 1-hour rate is the one
+    that matters here, because Claude Code sessions run the 1-hour TTL.
+    `cache_read` is passed only for a model that departs from the 0.1x rule.
+    """
+    return Rates(
+        input=input_rate,
+        output=output_rate,
+        cache_read=input_rate * 0.1 if cache_read is None else cache_read,
+        cache_write_5m=input_rate * 1.25,
+        cache_write_1h=input_rate * 2.0,
+    )
+
+
+# Matched by the longest prefix of the lowercased model name. Public Anthropic
+# list prices; the claude lane reports input_tokens WITHOUT cache tokens,
+# unlike the codex lane.
+#
+# The 4-6/4-7/4-8 rows are not redundant with "claude-opus-4". Opus 4.6 and
+# later are priced like Opus 5 (5/25), not like Opus 4.0 (15/75), so the bare
+# prefix swallowed them and overcharged by 3x -- a plausible number rather
+# than an error. Longest-prefix matching keeps 4.0/4.1 on the 15/75 row.
+MODEL_RATES: dict[str, Rates] = {
+    "claude-fable-5-1": _rates(10.0, 50.0, cache_read=0.25),
+    "claude-fable-5": _rates(10.0, 50.0),
+    "claude-opus-5": _rates(5.0, 25.0),
+    "claude-opus-4-8": _rates(5.0, 25.0),
+    "claude-opus-4-7": _rates(5.0, 25.0),
+    "claude-opus-4-6": _rates(5.0, 25.0),
+    "claude-opus-4": _rates(15.0, 75.0),
+    "claude-sonnet-5": _rates(2.0, 10.0),
+    "claude-sonnet-4": _rates(3.0, 15.0),
+    "claude-haiku-4": _rates(1.0, 5.0),
+    "claude-3-7-sonnet": _rates(3.0, 15.0),
+    "claude-3-5-sonnet": _rates(3.0, 15.0),
+    "claude-3-5-haiku": _rates(0.8, 4.0),
 }
 
 
-def lookup_model_rates(model: str | None) -> tuple[float, float, float, float] | None:
+def lookup_model_rates(model: str | None) -> Rates | None:
     if not model:
         return None
     lowered = model.lower()
@@ -84,11 +124,30 @@ def session_usage(path: pathlib.Path) -> dict[str, Any]:
     tokens_out = 0
     cache_read = 0
     cache_write = 0
+    cache_write_5m = 0
+    cache_write_1h = 0
     for usage in messages.values():
         tokens_in += int(usage.get("input_tokens") or 0)
         tokens_out += int(usage.get("output_tokens") or 0)
         cache_read += int(usage.get("cache_read_input_tokens") or 0)
-        cache_write += int(usage.get("cache_creation_input_tokens") or 0)
+        written = int(usage.get("cache_creation_input_tokens") or 0)
+        cache_write += written
+        # usage.cache_creation says which TTL each write bought; the flat
+        # cache_creation_input_tokens total does not. Claude Code sessions run
+        # the 1-hour TTL, which costs 2x input rather than 1.25x, so pricing
+        # every write at 1.25x undercharges systematically.
+        breakdown = usage.get("cache_creation") or {}
+        write_5m = int(breakdown.get("ephemeral_5m_input_tokens") or 0)
+        write_1h = int(breakdown.get("ephemeral_1h_input_tokens") or 0)
+        if write_5m or write_1h:
+            cache_write_5m += write_5m
+            cache_write_1h += write_1h
+        else:
+            # Transcripts predating cache_creation carry no breakdown. Bill
+            # those at the cheaper 5-minute rate so a missing split
+            # understates rather than inflates, and the two split fields
+            # still sum to cache_creation_input_tokens.
+            cache_write_5m += written
 
     return {
         "session_id": session_id,
@@ -98,6 +157,8 @@ def session_usage(path: pathlib.Path) -> dict[str, Any]:
         "uncached_input_tokens": tokens_in,
         "cache_read_input_tokens": cache_read,
         "cache_creation_input_tokens": cache_write,
+        "cache_write_5m_input_tokens": cache_write_5m,
+        "cache_write_1h_input_tokens": cache_write_1h,
         "assistant_messages_seen": assistant_seen,
         "unique_assistant_messages": len(messages),
         "window_start": window_start,
@@ -110,18 +171,21 @@ def estimate_usd(
     *,
     uncached: int,
     cache_read: int,
-    cache_write: int,
+    cache_write_5m: int,
+    cache_write_1h: int,
     tokens_out: int,
     input_rate: float,
     output_rate: float,
     cache_read_rate: float,
-    cache_write_rate: float,
+    cache_write_5m_rate: float,
+    cache_write_1h_rate: float,
 ) -> float:
     return round(
         (
             uncached * input_rate
             + cache_read * cache_read_rate
-            + cache_write * cache_write_rate
+            + cache_write_5m * cache_write_5m_rate
+            + cache_write_1h * cache_write_1h_rate
             + tokens_out * output_rate
         )
         / 1_000_000,
@@ -152,26 +216,35 @@ def main() -> int:
         cache_read_rate = (
             args.cache_read_usd_per_mtok if args.cache_read_usd_per_mtok is not None else 0.5
         )
-        cache_write_rate = (
+        # An explicit write rate is a blunt override: it applies to both TTLs,
+        # because the caller asked for one number.
+        cache_write_5m_rate = cache_write_1h_rate = (
             args.cache_write_usd_per_mtok if args.cache_write_usd_per_mtok is not None else 6.25
         )
         rate_source, usd_basis = "cli", "default-rates"
     elif table_rates is not None:
-        input_rate, output_rate, cache_read_rate, cache_write_rate = table_rates
+        input_rate = table_rates.input
+        output_rate = table_rates.output
+        cache_read_rate = table_rates.cache_read
+        cache_write_5m_rate = table_rates.cache_write_5m
+        cache_write_1h_rate = table_rates.cache_write_1h
         rate_source, usd_basis = "model-table", "model-rates"
     else:
-        input_rate, output_rate, cache_read_rate, cache_write_rate = 5.0, 25.0, 0.5, 6.25
+        input_rate, output_rate, cache_read_rate = 5.0, 25.0, 0.5
+        cache_write_5m_rate = cache_write_1h_rate = 6.25
         rate_source, usd_basis = "cli-default", "default-rates"
 
     usage["usd_estimated"] = estimate_usd(
         uncached=int(usage["uncached_input_tokens"]),
         cache_read=int(usage["cache_read_input_tokens"]),
-        cache_write=int(usage["cache_creation_input_tokens"]),
+        cache_write_5m=int(usage["cache_write_5m_input_tokens"]),
+        cache_write_1h=int(usage["cache_write_1h_input_tokens"]),
         tokens_out=int(usage["tokens_out"]),
         input_rate=input_rate,
         output_rate=output_rate,
         cache_read_rate=cache_read_rate,
-        cache_write_rate=cache_write_rate,
+        cache_write_5m_rate=cache_write_5m_rate,
+        cache_write_1h_rate=cache_write_1h_rate,
     )
     usage["rate_source"] = rate_source
     usage["usd_basis"] = usd_basis

skills/pr-cost/scripts/pr_cost_collect.py

@@ -11,7 +11,6 @@
 import shlex
 import subprocess
 import sys
-import tempfile
 from datetime import datetime, timezone
 from typing import Any
 
@@ -231,15 +230,25 @@ def same_annotation(existing: dict[str, Any], payload: dict[str, Any]) -> bool:
     )
 
 
-def append_ledger(path: pathlib.Path, payload: dict[str, Any]) -> bool:
+def append_ledger(
+    path: pathlib.Path, payload: dict[str, Any], *, allow_duplicate: bool = False
+) -> str:
+    """Append one row. Returns "annotated", "corrected", or "duplicate".
+
+    The guard keys on pr_url + session_id, so a re-run carrying a corrected
+    figure for the same session looks identical to an accidental replay. With
+    `allow_duplicate` the corrected row is appended rather than replacing the
+    first, so the ledger keeps both what was published and what replaced it.
+    """
     rows = load_ledger(path)
-    if any(same_annotation(row, payload) for row in rows):
-        return False
+    duplicate = any(same_annotation(row, payload) for row in rows)
+    if duplicate and not allow_duplicate:
+        return "duplicate"
     path.parent.mkdir(parents=True, exist_ok=True)
     with path.open("a", encoding="utf-8") as handle:
         handle.write(json.dumps(payload, sort_keys=True))
         handle.write("\n")
-    return True
+    return "corrected" if duplicate else "annotated"
 
 
 def comment_body(payload: dict[str, Any]) -> str:
@@ -279,12 +288,12 @@ def command_emit(args: argparse.Namespace) -> dict[str, Any]:
 def command_annotate(args: argparse.Namespace) -> dict[str, Any]:
     payload = payload_from_args(args)
     target_ledger = ledger_path(args.ledger)
-    wrote_ledger = append_ledger(target_ledger, payload)
+    status = append_ledger(target_ledger, payload, allow_duplicate=args.allow_duplicate)
     commented = False
-    if wrote_ledger:
+    if status != "duplicate":
         commented = maybe_comment_pr(payload, live=os.environ.get("PR_COST_HOOK_LIVE") == "1")
     return {
-        "status": "annotated" if wrote_ledger else "duplicate",
+        "status": status,
         "ledger": str(target_ledger),
         "commented": commented,
         "payload": payload,
@@ -367,14 +376,16 @@ def command_from_hook(args: argparse.Namespace) -> int:
         args.pr_url = pr_url
         payload = payload_from_args(args, default_pr_url=pr_url)
         target_ledger = ledger_path(args.ledger)
-        wrote_ledger = append_ledger(target_ledger, payload)
+        # No --allow-duplicate here on purpose: a hook that re-fires must stay
+        # idempotent, or one retried PR create posts the cost twice.
+        status = append_ledger(target_ledger, payload)
         commented = False
-        if wrote_ledger:
+        if status != "duplicate":
             commented = maybe_comment_pr(payload, live=os.environ.get("PR_COST_HOOK_LIVE") == "1")
         print(
             json.dumps(
                 {
-                    "status": "annotated" if wrote_ledger else "duplicate",
+                    "status": status,
                     "ledger": str(target_ledger),
                     "commented": commented,
                     "payload": payload,
@@ -414,6 +425,12 @@ def build_parser() -> argparse.ArgumentParser:
     annotate = subparsers.add_parser("annotate", help="append payload to local ledger and optionally comment")
     add_payload_arguments(annotate)
     annotate.add_argument("--ledger", type=pathlib.Path)
+    annotate.add_argument(
+        "--allow-duplicate",
+        dest="allow_duplicate",
+        action="store_true",
+        help="publish a corrected figure for a session already in the ledger",
+    )
     annotate.set_defaults(handler=command_annotate)
 
     from_hook = subparsers.add_parser("from-hook", help="parse hook stdin and annotate matching PR creates")

skills/pr-cost/tests/test_cache_aware_usd.py

@@ -254,6 +254,108 @@ def test_codex_reports_defaults_for_an_unknown_model(self) -> None:
         )
         self.assertEqual(data["rate_source"], "cli-default")
 
+    # 3. Model families priced by their own row, and cache writes by TTL.
+
+    def test_opus_4_6_is_not_priced_as_opus_4_0(self) -> None:
+        # Opus 4.6 and later cost 5/25, not Opus 4.0's 15/75. The bare
+        # "claude-opus-4" prefix swallowed them and charged 3x -- a plausible
+        # figure rather than an error, which is why nothing caught it.
+        # 1M uncached input at 5.0 is $5.00; at Opus 4.0's rate it is $15.00.
+        data = self.write_and_run(
+            "opus-4-6.jsonl",
+            claude_event(
+                "msg_1",
+                "claude-opus-4-6-20260101",
+                {"input_tokens": 1_000_000, "output_tokens": 0},
+            ),
+            CLAUDE_SCRIPT,
+        )
+        self.assertAlmostEqual(data["usd_estimated"], 5.0, places=4)
+
+    def test_opus_4_0_keeps_its_own_higher_rate(self) -> None:
+        # The guard for the row above: longest-prefix matching must still put
+        # genuine Opus 4.0 on 15/75. Deleting the 4-6/4-7/4-8 rows makes the
+        # previous test fail; deleting the "claude-opus-4" row makes this one.
+        data = self.write_and_run(
+            "opus-4-0.jsonl",
+            claude_event(
+                "msg_1",
+                "claude-opus-4-20250514",
+                {"input_tokens": 1_000_000, "output_tokens": 0},
+            ),
+            CLAUDE_SCRIPT,
+        )
+        self.assertAlmostEqual(data["usd_estimated"], 15.0, places=4)
+
+    def test_opus_5_is_priced_from_the_table_not_the_defaults(self) -> None:
+        # The dollar figure alone cannot catch this: the flat defaults are
+        # 5/25, which happen to equal Opus 5's real rates, so an unmatched
+        # model produced a correct-looking number under a label saying it was
+        # not priced from the model. Assert the label.
+        data = self.write_and_run(
+            "opus-5.jsonl",
+            claude_event(
+                "msg_1",
+                "claude-opus-5",
+                {"input_tokens": 1_000, "output_tokens": 1_000},
+            ),
+            CLAUDE_SCRIPT,
+        )
+        self.assertEqual(data["rate_source"], "model-table")
+        self.assertEqual(data["usd_basis"], "model-rates")
+
+    def test_one_hour_cache_writes_cost_twice_input(self) -> None:
+        # Sonnet rates 3.0 input: a 5-minute write is 1.25x (3.75) and a
+        # 1-hour write is 2x (6.00). Claude Code runs the 1-hour TTL, so
+        # pricing every write at 1.25x undercharged systematically.
+        # 1M 1-hour write tokens = $6.00, where the flat rate gave $3.75.
+        data = self.write_and_run(
+            "write-1h.jsonl",
+            claude_event(
+                "msg_1",
+                "claude-sonnet-4-20250514",
+                {
+                    "input_tokens": 0,
+                    "output_tokens": 0,
+                    "cache_creation_input_tokens": 1_000_000,
+                    "cache_creation": {
+                        "ephemeral_5m_input_tokens": 0,
+                        "ephemeral_1h_input_tokens": 1_000_000,
+                    },
+                },
+            ),
+            CLAUDE_SCRIPT,
+        )
+        self.assertEqual(data["cache_write_1h_input_tokens"], 1_000_000)
+        self.assertEqual(data["cache_write_5m_input_tokens"], 0)
+        self.assertAlmostEqual(data["usd_estimated"], 6.0, places=4)
+
+    def test_a_missing_cache_creation_breakdown_bills_the_cheaper_rate(self) -> None:
+        # Transcripts predating cache_creation carry no split. Those writes
+        # must land on the 5-minute rate, so a missing breakdown understates
+        # rather than inflates -- and the two split fields must still sum to
+        # cache_creation_input_tokens, or the payload contradicts itself.
+        data = self.write_and_run(
+            "write-no-breakdown.jsonl",
+            claude_event(
+                "msg_1",
+                "claude-sonnet-4-20250514",
+                {
+                    "input_tokens": 0,
+                    "output_tokens": 0,
+                    "cache_creation_input_tokens": 1_000_000,
+                },
+            ),
+            CLAUDE_SCRIPT,
+        )
+        self.assertEqual(data["cache_write_5m_input_tokens"], 1_000_000)
+        self.assertEqual(data["cache_write_1h_input_tokens"], 0)
+        self.assertEqual(
+            data["cache_write_5m_input_tokens"] + data["cache_write_1h_input_tokens"],
+            data["cache_creation_input_tokens"],
+        )
+        self.assertAlmostEqual(data["usd_estimated"], 3.75, places=4)
+
 
 if __name__ == "__main__":
     unittest.main()