fix: exclude zero-token <synthetic> records from Claude executions

tests/runtimes/test_claude_adapter.py

@@ -234,6 +234,29 @@ def test_split_message_usage_is_deduplicated_and_content_free(self):
                         "private tool output", "argument"):
             self.assertNotIn(private, encoded)
 
+    def test_native_load_ignores_complete_zero_usage_synthetic_marker(self):
+        self._write([*self.rows, {
+            "type": "assistant", "timestamp": "2026-08-11T00:00:07Z",
+            "message": {
+                "id": "msg-synthetic", "model": "<synthetic>",
+                "stop_reason": "stop_sequence", "content": [],
+                "usage": {
+                    "input_tokens": 0, "cache_read_input_tokens": 0,
+                    "cache_creation_input_tokens": 0, "output_tokens": 0,
+                },
+            },
+        }])
+        source = self.adapter.discover(DiscoveryContext(home=str(self.root)))[0]
+
+        result = self.adapter.load(source, DetailLevel.FULL)
+
+        self.assertEqual(len(result.turns), 2)
+        self.assertEqual(result.usage.input_tokens.value, 45)
+        self.assertEqual(result.usage.output_tokens.value, 25)
+        self.assertEqual(result.ended_at, datetime(
+            2026, 8, 11, 0, 0, 6, tzinfo=timezone.utc,
+        ).astimezone())
+
     def test_mcp_trace_views_are_structural_and_content_free(self):
         self.adapter.compatibility = meter._claude_compatibility()
         source = self.adapter.discover_legacy(

tests/test_meter.py

@@ -4097,6 +4097,146 @@ 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])
+        self.assertEqual(row["turns"], 1)
+        self.assertTrue(row["terminal"])
+        from token_meter.domain.aggregates import aggregate_cross_session_rows
+        self.assertEqual(
+            aggregate_cross_session_rows([row])["total_executions"], 1,
+        )
+
+    def test_claude_summary_synthetic_marker_does_not_extend_timing(self):
+        synthetic = self.synthetic_record()
+        synthetic["timestamp"] = "2026-07-02T00:01:00.000Z"
+        row = meter.claude_summary(self.source("claude"), [{
+            "type": "user", "timestamp": "2026-07-02T00:00:00.000Z",
+            "message": {"content": "test request"},
+        }, self.claude_usage_row("2026-07-02T00:00:01.000Z"), synthetic])
+
+        self.assertEqual(row["duration_s"], 1)
+        self.assertEqual(len(row["_wait_samples"]), 1)
+        self.assertEqual(row["_wait_samples"][0]["duration_s"], 1)
+        self.assertEqual(row["_wait_samples"][0]["model"], "claude-sonnet-4-6")
+
+    def synthetic_record(self, usage=None):
+        if usage is None:
+            usage = {
+                "input_tokens": 0, "cache_creation_input_tokens": 0,
+                "cache_read_input_tokens": 0, "output_tokens": 0,
+            }
+        return {
+            "type": "assistant", "timestamp": "2026-07-02T00:01:00.000Z",
+            "message": {
+                "id": "msg-synthetic", "model": "<synthetic>", "content": [],
+                "usage": usage,
+                "stop_reason": "stop_sequence",
+            },
+        }
+
+    def test_claude_synthetic_marker_requires_valid_complete_zero_usage(self):
+        cases = {
+            "malformed": {
+                "input_tokens": "bad", "cache_creation_input_tokens": 0,
+                "cache_read_input_tokens": 0, "output_tokens": "bad",
+            },
+            "missing-output": {
+                "input_tokens": 0, "cache_creation_input_tokens": 0,
+                "cache_read_input_tokens": 0,
+            },
+        }
+        for name, usage in cases.items():
+            with self.subTest(name=name):
+                objs = [self.claude_usage_row(), self.synthetic_record(usage)]
+
+                summary = meter.claude_summary(self.source("claude"), objs)
+                with tempfile.TemporaryDirectory() as tmp:
+                    path = Path(tmp) / "session.jsonl"
+                    path.write_text("".join(json.dumps(row) + "\n" for row in objs))
+                    state = meter.recompute_claude({
+                        **self.source("claude"), "path": str(path),
+                        "session": path.name,
+                    })
+
+                self.assertEqual(summary["turns"], 2)
+                self.assertEqual(summary["primary_model"], "<synthetic>")
+                self.assertFalse(summary["availability"]["cost"])
+                self.assertFalse(state["availability"]["cost"])
+                if name == "malformed":
+                    self.assertFalse(summary["availability"]["tokens"])
+                    self.assertFalse(state["availability"]["tokens"])
+
+    def test_claude_summary_keeps_other_angle_bracket_models(self):
+        other = self.synthetic_record()
+        other["message"]["model"] = "<local-model>"
+
+        row = meter.claude_summary(
+            self.source("claude"), [self.claude_usage_row(), other],
+        )
+
+        self.assertEqual(row["turns"], 2)
+        self.assertEqual(row["primary_model"], "<local-model>")
+        self.assertIn("<local-model>", row["models"])
+
+    def test_claude_only_synthetic_marker_keeps_usage_unavailable(self):
+        objs = [self.synthetic_record()]
+
+        summary = meter.claude_summary(self.source("claude"), objs)
+        with tempfile.TemporaryDirectory() as tmp:
+            path = Path(tmp) / "session.jsonl"
+            path.write_text(json.dumps(objs[0]) + "\n")
+            state = meter.recompute_claude({
+                **self.source("claude"), "path": str(path),
+                "session": path.name,
+            })
+
+        self.assertEqual(summary["turns"], 0)
+        self.assertFalse(summary["availability"]["tokens"])
+        self.assertFalse(summary["availability"]["cost"])
+        self.assertEqual(state["executions"], [])
+        self.assertFalse(state["availability"]["tokens"])
+        self.assertFalse(state["availability"]["cost"])
+
+    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 = {

token_meter/runtimes/claude.py

@@ -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
@@ -222,6 +228,38 @@ def _cost_coverage_complete(usage, priced):
     )
 
 
+def _zero_usage_synthetic_marker(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. Ignore only the exact marker with complete, valid zero-token evidence
+    so malformed, incomplete, or token-bearing records still fail closed.
+    """
+    if str(model or "") != "<synthetic>" or not isinstance(usage, dict):
+        return False
+    for field in USAGE_TOKEN_FIELDS:
+        if field not in usage:
+            return False
+        count, reported = normalize_reported_token_count(usage.get(field))
+        if not reported or count != 0:
+            return False
+    return True
+
+
+def _without_zero_usage_synthetic_markers(rows):
+    return tuple(
+        row for row in rows
+        if not (
+            row.get("type") == "assistant"
+            and isinstance(row.get("message"), dict)
+            and _zero_usage_synthetic_marker(
+                row["message"].get("model"), row["message"].get("usage"),
+            )
+        )
+    )
+
+
 def _compact(value, limit=90):
     value = " ".join(str(value or "").split())
     return value[:limit - 1] + "…" if len(value) > limit else value
@@ -769,7 +807,7 @@ def logical_messages(self, rows, timestamp_parser=None):
         timestamp_parser = timestamp_parser or _timestamp
         by_id = {}
         order = []
-        for row in rows:
+        for row in _without_zero_usage_synthetic_markers(rows):
             if row.get("type") != "assistant":
                 continue
             message = row.get("message") if isinstance(row.get("message"), dict) else {}
@@ -832,6 +870,7 @@ def load(self, source, detail):
         )
         if not available:
             return self._empty(source, detail, ("source_unavailable",))
+        rows = _without_zero_usage_synthetic_markers(rows)
         messages = self.logical_messages(rows)
         usage_seen = False
         input_complete = True
@@ -977,6 +1016,7 @@ def recompute_legacy(self, source):
         objs, _corrupt, _available = self.load_rows(paths)
         if not objs:
             return None
+        objs = _without_zero_usage_synthetic_markers(objs)
 
         msgs = self.logical_messages(objs, timestamp_parser=parse_iso)
         user_events = claude_user_events(objs)
@@ -1238,15 +1278,17 @@ def recompute_legacy(self, source):
                                   "claude", primary_model, approx_cost, executions)
 
         wait_samples = claude_wait_samples(objs)
+        has_executions = bool(executions)
         state = build_state(source, tot, cost, total_tokens, total_cost, series, executions, trace, semantic,
                             analyses, insights, first_ts, last_ts, idle, biggest, side_turns, approx_cost,
                             primary_model, "exact Claude API-rate estimate", execution_timing("claude", objs),
                             wait_samples, availability=metric_availability(
-                                "claude", cost=price_complete,
-                                tokens=input_complete and output_complete,
-                                input_tokens=input_complete,
-                                output_tokens=output_complete,
-                                cache=input_complete,
+                                "claude", cost=has_executions and price_complete,
+                                tokens=(has_executions and input_complete
+                                        and output_complete),
+                                input_tokens=has_executions and input_complete,
+                                output_tokens=has_executions and output_complete,
+                                cache=has_executions and input_complete,
                             ))
         state["throughput"] = performance_summary(claude_performance_samples(objs), tot["output"])
         return state
@@ -1257,6 +1299,7 @@ def summarize_legacy(self, source, objs=None):
             objs, _corrupt, _available = self.load_rows(
                 source.get("_trace_paths") or (source.get("path") or "",)
             )
+        objs = _without_zero_usage_synthetic_markers(objs)
         CURRENT_SESSION_CONTEXT_SAMPLES = compat["context_sample_limit"]
         add_model_daily = compat["add_model_daily"]
         add_model_summary = compat["add_model_summary"]
@@ -1377,15 +1420,17 @@ def summarize_legacy(self, source, objs=None):
 
         performance = claude_performance_samples(objs)
         wait_samples = claude_wait_samples(objs)
+        has_messages = bool(msgs)
         row = summary_row(source, title, cost, tokens, len(msgs), models, first_ts, last_ts, model_cost, model_tok, day_cost, approx,
                           execution_timing("claude", objs), input_tokens, output_tokens, model_stats,
                           list(model_daily.values()), performance, wait_samples,
                           availability=metric_availability(
-                              "claude", cost=price_complete,
-                              tokens=input_complete and output_complete,
-                              input_tokens=input_complete,
-                              output_tokens=output_complete,
-                              cache=input_complete,
+                              "claude", cost=has_messages and price_complete,
+                              tokens=(has_messages and input_complete
+                                      and output_complete),
+                              input_tokens=has_messages and input_complete,
+                              output_tokens=has_messages and output_complete,
+                              cache=has_messages and input_complete,
                           ),
                           session_name=declared_title)
         row["primary_model"] = primary_model