feat(agent): track per-model token usage for mid-session model switches

agent/insights.py

@@ -17,6 +17,7 @@
 """
 
 import json
+import sqlite3
 import time
 from collections import Counter, defaultdict
 from datetime import datetime
@@ -142,7 +143,7 @@ def generate(self, days: int = 30, source: str = None) -> Dict[str, Any]:
 
         # Compute insights
         overview = self._compute_overview(sessions, message_stats)
-        models = self._compute_model_breakdown(sessions)
+        models = self._compute_model_breakdown(sessions, cutoff, source)
         platforms = self._compute_platform_breakdown(sessions)
         tools = self._compute_tool_breakdown(tool_usage)
         skills = self._compute_skill_breakdown(skill_usage)
@@ -473,39 +474,140 @@ def _compute_overview(self, sessions: List[Dict], message_stats: Dict) -> Dict:
             "included_cost_sessions": included_cost_sessions,
         }
 
-    def _compute_model_breakdown(self, sessions: List[Dict]) -> List[Dict]:
-        """Break down usage by model."""
+    _GET_MODEL_USAGE_WITH_SOURCE = (
+        "SELECT u.session_id, u.model, u.billing_provider, u.billing_base_url,"
+        " u.api_call_count, u.input_tokens, u.output_tokens,"
+        " u.cache_read_tokens, u.cache_write_tokens, u.reasoning_tokens,"
+        " u.estimated_cost_usd"
+        " FROM session_model_usage u"
+        " JOIN sessions s ON s.id = u.session_id"
+        " WHERE s.started_at >= ? AND s.source = ?"
+    )
+    _GET_MODEL_USAGE_ALL = (
+        "SELECT u.session_id, u.model, u.billing_provider, u.billing_base_url,"
+        " u.api_call_count, u.input_tokens, u.output_tokens,"
+        " u.cache_read_tokens, u.cache_write_tokens, u.reasoning_tokens,"
+        " u.estimated_cost_usd"
+        " FROM session_model_usage u"
+        " JOIN sessions s ON s.id = u.session_id"
+        " WHERE s.started_at >= ?"
+    )
+
+    def _get_model_usage(self, cutoff: float, source: str = None) -> List[Dict]:
+        """Fetch per-model usage rows within the window (issue #51607).
+
+        Returns an empty list when the table is missing (e.g. a DB opened by
+        older code that never created it) so the caller can fall back to the
+        per-session aggregate.
+        """
+        try:
+            if source:
+                cursor = self._conn.execute(
+                    self._GET_MODEL_USAGE_WITH_SOURCE, (cutoff, source)
+                )
+            else:
+                cursor = self._conn.execute(self._GET_MODEL_USAGE_ALL, (cutoff,))
+            return [dict(row) for row in cursor.fetchall()]
+        except sqlite3.OperationalError:
+            return []
+
+    def _compute_model_breakdown(
+        self, sessions: List[Dict], cutoff: float, source: str = None
+    ) -> List[Dict]:
+        """Break down token usage and cost by model.
+
+        Tokens and cost are attributed per model from session_model_usage, so a
+        session that switched models mid-flight (via ``/model``) splits across
+        every model it used instead of dumping everything on the initial model
+        (issue #51607). Sessions without per-model rows — e.g. data written
+        before this table existed and not yet backfilled — fall back to their
+        single recorded (model, billing_provider) aggregate so nothing is lost.
+
+        Tool calls aren't tied to a specific API invocation, so they stay
+        attributed to the session's recorded model.
+        """
         model_data = defaultdict(lambda: {
-            "sessions": 0, "input_tokens": 0, "output_tokens": 0,
+            "sessions": set(), "input_tokens": 0, "output_tokens": 0,
             "cache_read_tokens": 0, "cache_write_tokens": 0,
-            "total_tokens": 0, "tool_calls": 0, "cost": 0.0,
+            "reasoning_tokens": 0, "total_tokens": 0, "api_calls": 0,
+            "tool_calls": 0, "cost": 0.0,
         })
 
-        for s in sessions:
-            model = s.get("model") or "unknown"
+        def _accumulate(model, provider, base_url, session_id, inp, out,
+                        cache_read, cache_write, reasoning):
+            model = model or "unknown"
             # Normalize: strip provider prefix for display
             display_model = model.split("/")[-1] if "/" in model else model
             d = model_data[display_model]
-            d["sessions"] += 1
-            inp = s.get("input_tokens") or 0
-            out = s.get("output_tokens") or 0
-            cache_read = s.get("cache_read_tokens") or 0
-            cache_write = s.get("cache_write_tokens") or 0
+            d["sessions"].add(session_id)
             d["input_tokens"] += inp
             d["output_tokens"] += out
             d["cache_read_tokens"] += cache_read
             d["cache_write_tokens"] += cache_write
+            d["reasoning_tokens"] += reasoning
             d["total_tokens"] += inp + out + cache_read + cache_write
-            d["tool_calls"] += s.get("tool_call_count") or 0
-            estimate, status = _estimate_cost(s)
+            estimate, status = _estimate_cost(
+                model, inp, out,
+                cache_read_tokens=cache_read, cache_write_tokens=cache_write,
+                provider=provider or None, base_url=base_url,
+            )
             d["cost"] += estimate
-            d["has_pricing"] = has_known_pricing(model, s.get("billing_provider"), s.get("billing_base_url"))
             d["cost_status"] = status
+            if has_known_pricing(model, provider or None, base_url):
+                d["has_pricing"] = True
+            else:
+                d.setdefault("has_pricing", False)
+            return display_model
+
+        usage_rows = self._get_model_usage(cutoff, source)
+        covered: set = set()
+        for r in usage_rows:
+            covered.add(r["session_id"])
+            d = _accumulate(
+                r["model"], r["billing_provider"], r.get("billing_base_url"),
+                r["session_id"], r["input_tokens"] or 0, r["output_tokens"] or 0,
+                r["cache_read_tokens"] or 0, r["cache_write_tokens"] or 0,
+                r["reasoning_tokens"] or 0,
+            )
+            model_data[d]["api_calls"] += r["api_call_count"] or 0
 
-        result = [
-            {"model": model, **data}
-            for model, data in model_data.items()
-        ]
+        # Fallback for sessions with token totals but no per-model rows
+        # (legacy data not covered by the v17 backfill). Attribute their
+        # aggregate to the single recorded model so totals never regress.
+        for s in sessions:
+            if s["id"] in covered:
+                continue
+            inp = s.get("input_tokens") or 0
+            out = s.get("output_tokens") or 0
+            cache_read = s.get("cache_read_tokens") or 0
+            cache_write = s.get("cache_write_tokens") or 0
+            if not (inp or out or cache_read or cache_write):
+                continue
+            _accumulate(
+                s.get("model"), s.get("billing_provider"),
+                s.get("billing_base_url"), s["id"],
+                inp, out, cache_read, cache_write, 0,
+            )
+
+        # Tool calls are attributed by the session's recorded model.
+        for s in sessions:
+            tool_calls = s.get("tool_call_count") or 0
+            if not tool_calls:
+                continue
+            model = s.get("model") or "unknown"
+            display_model = model.split("/")[-1] if "/" in model else model
+            model_data[display_model]["tool_calls"] += tool_calls
+
+        result = []
+        for model, data in model_data.items():
+            entry = {"model": model, **data}
+            entry["sessions"] = len(data["sessions"])
+            # Models that surfaced only via tool-call attribution (no token
+            # rows) won't have these set by _accumulate — default them so the
+            # output shape is uniform for downstream/JSON consumers.
+            entry.setdefault("has_pricing", False)
+            entry.setdefault("cost_status", "unknown")
+            result.append(entry)
         # Sort by tokens first, fall back to session count when tokens are 0
         result.sort(key=lambda x: (x["total_tokens"], x["sessions"]), reverse=True)
         return result

hermes_state.py

@@ -122,7 +122,7 @@ def _delete_delegate_children(conn, parent_ids: List[str]) -> List[str]:
 
 DEFAULT_DB_PATH = get_hermes_home() / "state.db"
 
-SCHEMA_VERSION = 19
+SCHEMA_VERSION = 20
 
 # Cap on user-controlled FTS5 query input before regex/sanitizer processing.
 # Search queries do not need to be arbitrarily large, and bounding them keeps
@@ -767,6 +767,23 @@ def repair_state_db_schema(db_path: Path, *, backup: bool = True) -> Dict[str, A
     compacted INTEGER NOT NULL DEFAULT 0
 );
 
+CREATE TABLE IF NOT EXISTS session_model_usage (
+    session_id TEXT NOT NULL REFERENCES sessions(id),
+    model TEXT NOT NULL,
+    billing_provider TEXT NOT NULL DEFAULT '',
+    billing_base_url TEXT,
+    api_call_count INTEGER NOT NULL DEFAULT 0,
+    input_tokens INTEGER NOT NULL DEFAULT 0,
+    output_tokens INTEGER NOT NULL DEFAULT 0,
+    cache_read_tokens INTEGER NOT NULL DEFAULT 0,
+    cache_write_tokens INTEGER NOT NULL DEFAULT 0,
+    reasoning_tokens INTEGER NOT NULL DEFAULT 0,
+    estimated_cost_usd REAL NOT NULL DEFAULT 0,
+    first_seen REAL,
+    last_seen REAL,
+    PRIMARY KEY (session_id, model, billing_provider)
+);
+
 CREATE TABLE IF NOT EXISTS state_meta (
     key TEXT PRIMARY KEY,
     value TEXT
@@ -793,6 +810,8 @@ def repair_state_db_schema(db_path: Path, *, backup: bool = True) -> Dict[str, A
 CREATE INDEX IF NOT EXISTS idx_sessions_started ON sessions(started_at DESC);
 CREATE INDEX IF NOT EXISTS idx_messages_session ON messages(session_id, timestamp);
 CREATE INDEX IF NOT EXISTS idx_compression_locks_expires ON compression_locks(expires_at);
+CREATE INDEX IF NOT EXISTS idx_session_model_usage_session ON session_model_usage(session_id);
+CREATE INDEX IF NOT EXISTS idx_session_model_usage_model ON session_model_usage(model);
 """
 
 # Indexes that reference columns added in later schema versions must be
@@ -1553,6 +1572,45 @@ def _init_schema(self):
                     # means consumers fall back to sessions.json for those
                     # rows until the gateway rewrites them.
                     logger.debug("v18 gateway metadata backfill skipped: %s", exc)
+            if current_version < 20:
+                # v20: per-model usage attribution (issue #51607). Going
+                # forward update_token_counts() records each API call into
+                # session_model_usage keyed by the live model, but existing
+                # sessions only have their aggregate totals on the sessions
+                # row. Seed one usage row per historical session from those
+                # aggregates so insights reads uniformly from the new table.
+                # INSERT OR IGNORE keeps it idempotent: if newer code already
+                # wrote a (session_id, model, provider) row for a session, the
+                # PK conflict skips the stale aggregate rather than doubling it.
+                try:
+                    cursor.execute(
+                        """INSERT OR IGNORE INTO session_model_usage (
+                               session_id, model, billing_provider,
+                               billing_base_url, api_call_count, input_tokens,
+                               output_tokens, cache_read_tokens,
+                               cache_write_tokens, reasoning_tokens,
+                               estimated_cost_usd, first_seen, last_seen
+                           )
+                           SELECT id, COALESCE(model, 'unknown'),
+                                  COALESCE(billing_provider, ''),
+                                  billing_base_url,
+                                  COALESCE(api_call_count, 0),
+                                  COALESCE(input_tokens, 0),
+                                  COALESCE(output_tokens, 0),
+                                  COALESCE(cache_read_tokens, 0),
+                                  COALESCE(cache_write_tokens, 0),
+                                  COALESCE(reasoning_tokens, 0),
+                                  COALESCE(estimated_cost_usd, 0),
+                                  started_at, COALESCE(ended_at, started_at)
+                           FROM sessions
+                           WHERE COALESCE(input_tokens, 0)
+                                 + COALESCE(output_tokens, 0)
+                                 + COALESCE(cache_read_tokens, 0)
+                                 + COALESCE(cache_write_tokens, 0)
+                                 + COALESCE(reasoning_tokens, 0) > 0"""
+                    )
+                except sqlite3.OperationalError:
+                    pass
             if current_version < SCHEMA_VERSION and fts_migrations_complete:
                 cursor.execute(
                     "UPDATE schema_version SET version = ?",
@@ -2499,10 +2557,116 @@ def update_token_counts(
             api_call_count,
             session_id,
         )
+        # Per-model usage attribution.  ``update_token_counts`` is the single
+        # chokepoint every per-API-call delta flows through (CLI, gateway, cron,
+        # delegated runs — see conversation_loop / codex_runtime), and each call
+        # carries the model/provider *active at the time of that call*.  The
+        # ``sessions`` row only keeps one (model, billing_provider) pair, so a
+        # mid-session ``/model`` switch otherwise attributes every token to the
+        # initial model (issue #51607).  Recording the per-call delta into
+        # session_model_usage keyed by the live model preserves an accurate
+        # per-model breakdown regardless of how many times the user switches.
+        #
+        # Only the incremental path records here: the gateway also issues an
+        # ``absolute=True`` call that overwrites the sessions summary totals
+        # with the cached agent's cumulative figures — folding those in would
+        # double-count, and cumulative totals can't be split back per model.
+        record_model_usage = (not absolute) and (
+            input_tokens or output_tokens or cache_read_tokens
+            or cache_write_tokens or reasoning_tokens or api_call_count
+            or estimated_cost_usd
+        )
+
         def _do(conn):
             conn.execute(sql, params)
+            if record_model_usage:
+                self._record_model_usage(
+                    conn,
+                    session_id,
+                    model=model,
+                    billing_provider=billing_provider,
+                    billing_base_url=billing_base_url,
+                    input_tokens=input_tokens,
+                    output_tokens=output_tokens,
+                    cache_read_tokens=cache_read_tokens,
+                    cache_write_tokens=cache_write_tokens,
+                    reasoning_tokens=reasoning_tokens,
+                    estimated_cost_usd=estimated_cost_usd,
+                    api_call_count=api_call_count,
+                )
         self._execute_write(_do)
 
+    def _record_model_usage(
+        self,
+        conn,
+        session_id: str,
+        *,
+        model: Optional[str],
+        billing_provider: Optional[str],
+        billing_base_url: Optional[str],
+        input_tokens: int,
+        output_tokens: int,
+        cache_read_tokens: int,
+        cache_write_tokens: int,
+        reasoning_tokens: int,
+        estimated_cost_usd: Optional[float],
+        api_call_count: int,
+    ) -> None:
+        """Accumulate a per-API-call usage delta into session_model_usage.
+
+        Runs inside the caller's write transaction (after the ``sessions``
+        UPDATE) so the per-model rows stay consistent with the summary row.
+        When the caller omits the model/provider (some paths only pass token
+        deltas), fall back to the values already recorded on the session row —
+        the same COALESCE-from-session behaviour the summary update uses.
+        """
+        row = conn.execute(
+            "SELECT model, billing_provider, billing_base_url "
+            "FROM sessions WHERE id = ?",
+            (session_id,),
+        ).fetchone()
+        sess_model = row["model"] if row is not None else None
+        sess_provider = row["billing_provider"] if row is not None else None
+        sess_base_url = row["billing_base_url"] if row is not None else None
+
+        eff_model = model or sess_model or "unknown"
+        eff_provider = billing_provider or sess_provider or ""
+        eff_base_url = billing_base_url or sess_base_url
+        now = time.time()
+        conn.execute(
+            """INSERT INTO session_model_usage (
+                   session_id, model, billing_provider, billing_base_url,
+                   api_call_count, input_tokens, output_tokens,
+                   cache_read_tokens, cache_write_tokens, reasoning_tokens,
+                   estimated_cost_usd, first_seen, last_seen
+               ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+               ON CONFLICT(session_id, model, billing_provider) DO UPDATE SET
+                   api_call_count = api_call_count + excluded.api_call_count,
+                   input_tokens = input_tokens + excluded.input_tokens,
+                   output_tokens = output_tokens + excluded.output_tokens,
+                   cache_read_tokens = cache_read_tokens + excluded.cache_read_tokens,
+                   cache_write_tokens = cache_write_tokens + excluded.cache_write_tokens,
+                   reasoning_tokens = reasoning_tokens + excluded.reasoning_tokens,
+                   estimated_cost_usd = estimated_cost_usd + excluded.estimated_cost_usd,
+                   billing_base_url = COALESCE(excluded.billing_base_url, billing_base_url),
+                   last_seen = excluded.last_seen""",
+            (
+                session_id,
+                eff_model,
+                eff_provider,
+                eff_base_url,
+                api_call_count or 0,
+                input_tokens or 0,
+                output_tokens or 0,
+                cache_read_tokens or 0,
+                cache_write_tokens or 0,
+                reasoning_tokens or 0,
+                float(estimated_cost_usd or 0.0),
+                now,
+                now,
+            ),
+        )
+
     def ensure_session(
         self,
         session_id: str,