Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
46 changes: 46 additions & 0 deletions tests/test_claude_cost_correctness.py
Original file line number Diff line number Diff line change
Expand Up @@ -342,6 +342,52 @@ def test_inconsistent_cache_detail_and_unknown_dimensions_are_unpriceable(self):
unknown_geo, "claude-opus-4-8", "claude",
).values()), 0.0)

def test_zeroed_copy_with_stale_cache_split_is_free_and_keeps_cost_covered(self):
copied = claude_usage(cache_write_5m=953_342)
copied["cache_creation_input_tokens"] = 0
source = {
"provider": "claude", "client": "claude_code",
"label": "Claude Code", "id": "zeroed-copy",
"session": "zeroed-copy.jsonl", "path": "/private/not-read",
"project": "private", "mtime": 1, "title": None, "model": None,
}

normalized = _normalized_usage(copied)
summary = meter.claude_summary(source, [
assistant("copied-from-deleted-parent", copied),
assistant(
"own", claude_usage(input_tokens=1_000_000),
"2026-09-07T00:00:01.000Z",
),
])

self.assertTrue(normalized["billing_available"])
self.assertEqual(normalized["cache_creation_5m_input_tokens"], 0)
self.assertEqual(sum(meter.cost_of(
copied, "claude-opus-4-8", "claude",
).values()), 0.0)
self.assertEqual(summary["cost"], 5.0)
self.assertTrue(summary["availability"]["cost"])
self.assertFalse(summary["cost_approx"])

def test_zero_cache_total_beside_billed_tokens_still_fails_closed(self):
for field in ("output_tokens", "cache_read_input_tokens"):
with self.subTest(field=field):
usage = claude_usage(cache_write_5m=500)
usage["cache_creation_input_tokens"] = 0
usage[field] = 10

self.assertFalse(_normalized_usage(usage)["billing_available"])

def test_zero_cache_split_row_missing_a_total_still_fails_closed(self):
for field in ("cache_read_input_tokens", "cache_creation_input_tokens"):
with self.subTest(field=field):
usage = claude_usage(cache_write_5m=500)
usage["cache_creation_input_tokens"] = 0
del usage[field]

self.assertFalse(_normalized_usage(usage)["billing_available"])

def test_aggregate_only_cache_write_remains_a_labeled_compatibility_estimate(self):
normalized = _normalized_usage({
"input_tokens": 0,
Expand Down
15 changes: 14 additions & 1 deletion token_meter/runtimes/claude.py
Original file line number Diff line number Diff line change
Expand Up @@ -154,7 +154,20 @@ def _normalized_usage(usage):
cache_write_unspecified = 0
cache_duration_valid = True
cache_duration_complete = cache_write == 0
if isinstance(cache_creation, dict):
output_count, output_count_reported = normalize_reported_token_count(
usage.get("output_tokens")
)
# Resumed and forked transcripts copy history with the four totals zeroed
# but the original cache_creation split left in place; the copy is not billed.
zeroed_copy = (
all(field in usage for field in USAGE_TOKEN_FIELDS)
and input_reported and cache_read_reported and cache_write_reported
and output_count_reported
and not (input_tokens or cache_read or cache_write or output_count)
)
if isinstance(cache_creation, dict) and zeroed_copy:
cache_duration_complete = True
elif isinstance(cache_creation, dict):
cache_write_5m, cache_write_5m_reported = normalize_reported_token_count(
cache_creation.get("ephemeral_5m_input_tokens")
)
Expand Down
Loading