Fix session analytics accuracy: dedupe usage, current pricing, active time

api/db/indexer.py

@@ -358,6 +358,7 @@ def _index_session(
                             (differs from encoded_name for remapped worktree sessions)
     """
     from models import Session
+    from models.usage import TokenUsage
     from utils import get_initial_prompt
 
     uuid = jsonl_path.stem
@@ -387,6 +388,9 @@ def _index_session(
     subagents_dir = jsonl_path.parent / uuid / "subagents"
     if subagents_dir.exists():
         subagent_count = len(list(subagents_dir.glob("agent-*.jsonl")))
+    subagents = session.list_subagents() if subagent_count else []
+    subagent_cost = sum(agent.get_total_cost() for agent in subagents)
+    subagent_usage = sum((agent.get_usage_summary() for agent in subagents), TokenUsage.zero())
 
     # Detect project_path from session data (or use override for worktrees)
     project_path = project_path_override or _detect_project_path(session, encoded_name)
@@ -405,7 +409,9 @@ def _index_session(
             uuid, slug, project_encoded_name, project_path,
             start_time, end_time, message_count, duration_seconds,
             input_tokens, output_tokens, cache_creation_tokens, cache_read_tokens,
-            total_cost, initial_prompt, git_branch, models_used,
+            total_cost, subagent_cost, subagent_input_tokens, subagent_output_tokens,
+            subagent_cache_read_tokens, subagent_cache_creation_tokens,
+            initial_prompt, git_branch, models_used,
             session_titles, is_continuation_marker, was_compacted,
             compaction_count, file_snapshot_count, subagent_count,
             jsonl_mtime, jsonl_size, session_source, source_encoded_name,
@@ -415,6 +421,8 @@ def _index_session(
             ?, ?, ?, ?,
             ?, ?, ?, ?,
             ?, ?, ?, ?,
+            ?, ?,
+            ?, ?, ?,
             ?, ?, ?,
             ?, ?, ?,
             ?, ?, ?, ?,
@@ -436,6 +444,11 @@ def _index_session(
             usage.cache_creation_input_tokens,
             usage.cache_read_input_tokens,
             session.get_total_cost(),
+            subagent_cost,
+            subagent_usage.total_input,
+            subagent_usage.output_tokens,
+            subagent_usage.cache_read_input_tokens,
+            subagent_usage.cache_creation_input_tokens,
             initial_prompt,
             git_branches[0] if git_branches else None,
             json.dumps(models_used) if models_used else None,
@@ -518,17 +531,15 @@ def _index_session(
             from services.subagent_types import get_all_subagent_types
 
             subagent_types = get_all_subagent_types(jsonl_path, subagents_dir)
-            for subagent in session.list_subagents():
+            for subagent in subagents:
                 subagent_type = subagent_types.get(subagent.agent_id, "_unknown")
                 if subagent_type == "_unknown":
                     logger.debug(
                         "Skipping unclassified subagent %s in session %s", subagent.agent_id, uuid
                     )
                     continue
                 usage = subagent.get_usage_summary()
-                duration = 0.0
-                if subagent.start_time and subagent.end_time:
-                    duration = (subagent.end_time - subagent.start_time).total_seconds()
+                duration = subagent.duration_seconds or 0.0
                 conn.execute(
                     """INSERT INTO subagent_invocations
                        (session_uuid, agent_id, subagent_type, input_tokens,
@@ -540,7 +551,7 @@ def _index_session(
                         subagent_type,
                         usage.total_input,
                         usage.output_tokens,
-                        usage.calculate_cost(subagent.get_primary_model()),
+                        subagent.get_total_cost(),
                         duration,
                         subagent.start_time.isoformat() if subagent.start_time else None,
                     ),
@@ -739,7 +750,9 @@ def _cleanup_stale_sessions(conn: sqlite3.Connection, projects_dir: Path) -> Non
     for col in (
         "slug project_encoded_name project_path start_time end_time message_count "
         "duration_seconds input_tokens output_tokens cache_creation_tokens cache_read_tokens "
-        "total_cost initial_prompt git_branch models_used session_titles is_continuation_marker "
+        "total_cost subagent_cost subagent_input_tokens subagent_output_tokens "
+        "subagent_cache_read_tokens subagent_cache_creation_tokens "
+        "initial_prompt git_branch models_used session_titles is_continuation_marker "
         "was_compacted compaction_count file_snapshot_count subagent_count jsonl_mtime jsonl_size "
         "session_source source_encoded_name indexed_at"
     ).split()

api/db/queries.py

@@ -397,8 +397,10 @@ def query_dashboard_stats(
             COUNT(*) as session_count,
             COUNT(DISTINCT project_encoded_name) as projects_active,
             COALESCE(SUM(duration_seconds), 0) as total_duration,
-            COALESCE(SUM(input_tokens + output_tokens + cache_read_tokens), 0) as total_tokens,
-            COALESCE(SUM(total_cost), 0) as total_cost
+            COALESCE(SUM(input_tokens + output_tokens + cache_read_tokens
+                + COALESCE(subagent_input_tokens, 0) + COALESCE(subagent_output_tokens, 0)
+                + COALESCE(subagent_cache_read_tokens, 0)), 0) as total_tokens,
+            COALESCE(SUM(total_cost + COALESCE(subagent_cost, 0)), 0) as total_cost
         FROM sessions
         WHERE start_time >= :start AND start_time <= :end""",
         {"start": start_dt.isoformat(), "end": end_dt.isoformat()},
@@ -451,11 +453,13 @@ def query_analytics(
     totals_row = conn.execute(
         f"""SELECT
             COUNT(*) as total_sessions,
-            COALESCE(SUM(input_tokens), 0) as total_input,
-            COALESCE(SUM(output_tokens), 0) as total_output,
-            COALESCE(SUM(cache_read_tokens), 0) as total_cache_read,
-            COALESCE(SUM(cache_creation_tokens), 0) as total_cache_creation,
-            COALESCE(SUM(total_cost), 0) as total_cost,
+            COALESCE(SUM(input_tokens + COALESCE(subagent_input_tokens, 0)), 0) as total_input,
+            COALESCE(SUM(output_tokens + COALESCE(subagent_output_tokens, 0)), 0) as total_output,
+            COALESCE(SUM(cache_read_tokens + COALESCE(subagent_cache_read_tokens, 0)), 0)
+                as total_cache_read,
+            COALESCE(SUM(cache_creation_tokens + COALESCE(subagent_cache_creation_tokens, 0)), 0)
+                as total_cache_creation,
+            COALESCE(SUM(total_cost + COALESCE(subagent_cost, 0)), 0) as total_cost,
             COALESCE(SUM(duration_seconds), 0) as total_duration,
             COUNT(DISTINCT project_encoded_name) as projects_active
         FROM sessions s

api/db/schema.py

@@ -10,8 +10,7 @@
 
 logger = logging.getLogger(__name__)
 
-# v20/v21 are taken by PR #106 (session analytics accuracy); v22 is the orphan repair.
-SCHEMA_VERSION = 22
+SCHEMA_VERSION = 23
 
 SCHEMA_SQL = """
 -- Schema versioning
@@ -35,6 +34,11 @@
     cache_creation_tokens INTEGER DEFAULT 0,
     cache_read_tokens INTEGER DEFAULT 0,
     total_cost REAL DEFAULT 0,
+    subagent_cost REAL DEFAULT 0,
+    subagent_input_tokens INTEGER DEFAULT 0,
+    subagent_output_tokens INTEGER DEFAULT 0,
+    subagent_cache_read_tokens INTEGER DEFAULT 0,
+    subagent_cache_creation_tokens INTEGER DEFAULT 0,
     initial_prompt TEXT,
     git_branch TEXT,
     models_used TEXT,
@@ -426,6 +430,9 @@ def ensure_schema(conn: sqlite3.Connection) -> None:
     # v13 incremental block below (ALTER is not idempotent).
     conn.executescript(_SHELLS_CRON_SCHEMA_SQL)
 
+    # Same cross-branch guard for sessions columns added after v19.
+    columns_added = _ensure_session_columns(conn)
+
     # Check current version
     try:
         row = conn.execute("SELECT MAX(version) FROM schema_version").fetchone()
@@ -799,6 +806,18 @@ def ensure_schema(conn: sqlite3.Connection) -> None:
             logger.info("Migrating → v22: remove orphaned child rows (issue #107)")
             _delete_orphan_rows(conn)
 
+        if current_version < 23:
+            logger.info(
+                "Migrating → v23: re-index costs "
+                "(per-response usage dedup, current pricing, active time)"
+            )
+            has_sessions = conn.execute(
+                "SELECT 1 FROM sqlite_master WHERE type='table' AND name='sessions'"
+            ).fetchone()
+            # _ensure_session_columns already forced the reindex if it added columns
+            if has_sessions and not columns_added:
+                conn.execute("UPDATE sessions SET jsonl_mtime = jsonl_mtime - 1")
+
     # Record version
     conn.execute(
         "INSERT OR REPLACE INTO schema_version (version) VALUES (?)",
@@ -812,6 +831,31 @@ def ensure_schema(conn: sqlite3.Connection) -> None:
     logger.info("Schema version %d applied successfully", SCHEMA_VERSION)
 
 
+# Sessions columns added after v19; also in SCHEMA_SQL for fresh installs.
+_SESSION_COLUMNS = [
+    ("subagent_cost", "REAL DEFAULT 0"),
+    ("subagent_input_tokens", "INTEGER DEFAULT 0"),
+    ("subagent_output_tokens", "INTEGER DEFAULT 0"),
+    ("subagent_cache_read_tokens", "INTEGER DEFAULT 0"),
+    ("subagent_cache_creation_tokens", "INTEGER DEFAULT 0"),
+]
+
+
+def _ensure_session_columns(conn: sqlite3.Connection) -> bool:
+    """Add missing sessions columns regardless of schema version; True if any were added."""
+    existing = {r[1] for r in conn.execute("PRAGMA table_info(sessions)").fetchall()}
+    if not existing:
+        return False
+    missing = [(name, ddl) for name, ddl in _SESSION_COLUMNS if name not in existing]
+    for name, ddl in missing:
+        conn.execute(f"ALTER TABLE sessions ADD COLUMN {name} {ddl}")
+    if missing:
+        logger.info("Added sessions columns %s; forcing a reindex", [n for n, _ in missing])
+        conn.execute("UPDATE sessions SET jsonl_mtime = jsonl_mtime - 1")
+        conn.commit()
+    return bool(missing)
+
+
 # Child rows whose parent is gone; (table, parent-existence subquery).
 _ORPHAN_RULES = [
     (

api/models/active_time.py

@@ -0,0 +1,27 @@
+"""Active time: the sum of gaps between messages, with each idle gap capped."""
+
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Optional
+
+# Idle gaps between messages count toward active duration only up to this cap
+ACTIVE_IDLE_CAP_SECONDS = 30 * 60
+
+
+class ActiveTimeAccumulator:
+    """Feed message timestamps in file order; `seconds` is the active time so far."""
+
+    def __init__(self) -> None:
+        self.seconds = 0.0
+        self._latest: Optional[datetime] = None  # active time only moves forward
+
+    def add(self, timestamp: Optional[datetime]) -> None:
+        if timestamp is None:
+            return
+        if self._latest is None:
+            self._latest = timestamp
+        elif timestamp > self._latest:
+            gap = (timestamp - self._latest).total_seconds()
+            self.seconds += min(gap, ACTIVE_IDLE_CAP_SECONDS)
+            self._latest = timestamp

api/models/agent.py

@@ -48,7 +48,9 @@ class AgentCache(BaseCache):
     __slots__ = (
         "start_time",
         "end_time",
+        "active_duration_seconds",
         "usage_summary",
+        "total_cost",
         "message_count",
         "tasks",
         "skills_used",
@@ -60,7 +62,9 @@ def __init__(self) -> None:
         super().__init__()
         self.start_time: Optional[datetime] = None
         self.end_time: Optional[datetime] = None
+        self.active_duration_seconds: Optional[float] = None
         self.usage_summary: Optional["TokenUsage"] = None
+        self.total_cost: Optional[float] = None
         self.message_count: Optional[int] = None
         self.tasks: Optional[List["Task"]] = None
         self.skills_used: Optional[Dict[tuple, int]] = None  # {(name, source): count}
@@ -71,7 +75,9 @@ def reset(self) -> None:
         """Reset all cache values to initial state."""
         self.start_time = None
         self.end_time = None
+        self.active_duration_seconds = None
         self.usage_summary = None
+        self.total_cost = None
         self.message_count = None
         self.tasks = None
         self.skills_used = None
@@ -210,14 +216,16 @@ def _load_metadata(self) -> None:
 
         from command_helpers import classify_invocation, is_command_category, is_skill_category
 
+        from .active_time import ActiveTimeAccumulator
         from .content import ToolUseBlock
         from .message import AssistantMessage
-        from .usage import TokenUsage
+        from .usage import UsageAccumulator
 
         # Initialize accumulators
         first_ts: Optional[datetime] = None
         last_ts: Optional[datetime] = None
-        usage = TokenUsage.zero()
+        active_time = ActiveTimeAccumulator()
+        usage_acc = UsageAccumulator()
         message_count = 0
         models: set = set()
         # Skills/commands use (name, source) tuple keys for invocation tracking
@@ -230,12 +238,12 @@ def _load_metadata(self) -> None:
             # Timestamps
             if first_ts is None:
                 first_ts = msg.timestamp
+            active_time.add(msg.timestamp)
             last_ts = msg.timestamp
 
             # Usage (assistant messages only)
             if isinstance(msg, AssistantMessage):
-                if msg.usage:
-                    usage = usage + msg.usage
+                usage_acc.add(msg)
                 if msg.model:
                     models.add(msg.model)
 
@@ -253,7 +261,9 @@ def _load_metadata(self) -> None:
         # Store all computed values
         cache.start_time = first_ts
         cache.end_time = last_ts
-        cache.usage_summary = usage
+        cache.active_duration_seconds = active_time.seconds
+        cache.usage_summary = usage_acc.usage
+        cache.total_cost = usage_acc.cost
         cache.message_count = message_count
         cache.models_used = models
         cache.skills_used = dict(skills)
@@ -324,6 +334,11 @@ def get_usage_summary(self) -> "TokenUsage":
         self._load_metadata()
         return self._get_cache().usage_summary or TokenUsage.zero()
 
+    def get_total_cost(self) -> float:
+        """Sum per-response costs, each priced by its own model (cached)."""
+        self._load_metadata()
+        return self._get_cache().total_cost or 0.0
+
     def get_skills_used(self) -> Dict[tuple, int]:
         """Get skill usage counts with invocation source (cached).
 
@@ -345,10 +360,6 @@ def get_models_used(self) -> set:
         self._load_metadata()
         return self._get_cache().models_used or set()
 
-    def get_primary_model(self) -> Optional[str]:
-        """Return the first model encountered, or None if no messages have a model."""
-        return next(iter(self.get_models_used()), None)
-
     @property
     def start_time(self) -> Optional[datetime]:
         """Get timestamp of first message (cached)."""
@@ -361,6 +372,14 @@ def end_time(self) -> Optional[datetime]:
         self._load_metadata()
         return self._get_cache().end_time
 
+    @property
+    def duration_seconds(self) -> Optional[float]:
+        """Active duration in seconds, idle gaps capped like Session.duration_seconds (cached)."""
+        self._load_metadata()
+        if not (self.start_time and self.end_time):
+            return None
+        return self._get_cache().active_duration_seconds
+
     def list_tasks(self) -> List[Task]:
         """
         Load tasks created by this agent from its JSONL file.