Skip to content
Open
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
56 changes: 56 additions & 0 deletions tests/test_meter.py
Original file line number Diff line number Diff line change
Expand Up @@ -4097,6 +4097,62 @@ def message(message_id, model, input_tokens, output_tokens):
self.assertTrue(row["availability"]["cost"])
self.assertFalse(row["cost_approx"])
self.assertGreater(row["cost"], 0)
self.assertEqual(row["primary_model"], "claude-sonnet-4-6")
self.assertEqual(row["models"], ["sonnet-4-6"])
self.assertEqual(row["context"]["latest"], 100)
self.assertEqual(row["_context_samples"], [100])

def synthetic_record(self, usage=None):
return {
"type": "assistant", "timestamp": "2026-07-02T00:01:00.000Z",
"message": {
"id": "msg-synthetic", "model": "<synthetic>", "content": [],
"usage": usage or {
"input_tokens": 0, "cache_creation_input_tokens": 0,
"cache_read_input_tokens": 0, "output_tokens": 0,
},
"stop_reason": "stop_sequence",
},
}

def test_claude_summary_prices_pseudo_model_records_that_report_tokens(self):
objs = [
self.claude_usage_row(),
self.synthetic_record({"input_tokens": 40, "output_tokens": 5}),
]

row = meter.claude_summary(self.source("claude"), objs)

self.assertFalse(row["availability"]["cost"])
self.assertEqual(row["primary_model"], "<synthetic>")

def test_claude_recompute_keeps_cost_available_with_synthetic_records(self):
records = [{
"type": "assistant", "timestamp": "2026-07-02T00:00:00.000Z",
"message": {
"id": "msg-1", "model": "claude-sonnet-4-6", "content": [],
"usage": {"input_tokens": 10, "cache_creation_input_tokens": 5_000,
"cache_read_input_tokens": 5_000, "output_tokens": 10,
"cache_creation": {
"ephemeral_5m_input_tokens": 5_000,
"ephemeral_1h_input_tokens": 0,
}},
"stop_reason": "end_turn",
},
}, self.synthetic_record()]
with tempfile.TemporaryDirectory() as tmp:
path = Path(tmp) / "session.jsonl"
path.write_text("".join(json.dumps(row) + "\n" for row in records))
source = {
**self.source("claude"), "path": str(path),
"session": path.name,
}
state = meter.recompute_claude(source)

self.assertTrue(state["availability"]["cost"])
self.assertFalse(state["cost_approx"])
self.assertGreater(state["total_cost"], 0)
self.assertEqual(len(state["executions"]), 1)

def test_unknown_model_keeps_cache_money_unavailable(self):
record = {
Expand Down
24 changes: 24 additions & 0 deletions token_meter/runtimes/claude.py
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,12 @@


DEFAULT_MODEL = "claude-sonnet-4-6"
USAGE_TOKEN_FIELDS = (
"input_tokens",
"output_tokens",
"cache_read_input_tokens",
"cache_creation_input_tokens",
)
MAX_DETAIL_TURNS = 2_000
MAX_TOOL_EVENTS = 2_000
ACTIVITY_TAIL_BYTES = 1024 * 1024
Expand Down Expand Up @@ -222,6 +228,20 @@ def _cost_coverage_complete(usage, priced):
)


def _unbilled_pseudo_model(model, usage):
"""Claude Code records locally generated messages as `<synthetic>`.

They are not model executions, so counting them misreports the session's
model identity, collapses latest context to zero, and inflates the execution
count. Require both the pseudo-model name shape and absent token evidence so
a record that does carry billable tokens is still priced.
"""
name = str(model or "")
if not (name.startswith("<") and name.endswith(">")):
return False
return not any(_safe_int(usage.get(field)) for field in USAGE_TOKEN_FIELDS)


def _compact(value, limit=90):
value = " ".join(str(value or "").split())
return value[:limit - 1] + "…" if len(value) > limit else value
Expand Down Expand Up @@ -1016,6 +1036,8 @@ def recompute_legacy(self, source):
)
if not usage:
continue
if _unbilled_pseudo_model(rec["model"], usage):
continue
input_complete = input_complete and usage["input_available"]
output_complete = output_complete and usage["output_available"]
idx = len(series) + 1
Expand Down Expand Up @@ -1299,6 +1321,8 @@ def summarize_legacy(self, source, objs=None):
)
if not usage:
continue
if _unbilled_pseudo_model(rec["model"], usage):
continue
input_complete = input_complete and usage["input_available"]
output_complete = output_complete and usage["output_available"]
primary_model = rec["model"] or primary_model
Expand Down