perf(dashboard): use GROUP BY for session stats instead of fetching 10k rows

Replaces the O(N) list_sessions_rich histogram in /api/sessions/stats
with a single GROUP BY query, reducing response time from ~575ms to
<1ms on large databases.

Original PR #48921 by @liuhao1024. Salvage fixes based on review
feedback from teknium1 and @wernerhp:

1. Preserve try/except guard — a DB error still degrades to empty
   by_source instead of failing the whole stats response.
2. GROUP BY COALESCE(source, 'cli') — the original GROUP BY source
   could emit duplicate 'cli' keys (NULL group + literal 'cli' group)
   that the dict comprehension silently dropped.
3. Add exclude_children=True — list_sessions_rich excludes subagent
   runs, delegates, and compression continuations by default; the
   bare GROUP BY counted all rows, inflating source counts.

Aggregate shape (exclude_children/include_archived/limit params)
adapted from closed duplicate #61120 by @mijanx.

Closes #48914
Co-authored-by: mijanx <mijanx@users.noreply.github.com>

hermes_cli/web_server.py

@@ -11695,9 +11695,10 @@ async def get_session_stats(profile: Optional[str] = None):
         messages = db.message_count()
         by_source: Dict[str, int] = {}
         try:
-            for s in db.list_sessions_rich(limit=10000, include_archived=True, compact_rows=True):
-                src = str(s.get("source") or "cli")
-                by_source[src] = by_source.get(src, 0) + 1
+            by_source = db.session_count_by_source(
+                include_archived=True,
+                exclude_children=True,
+            )
         except Exception:
             pass
         return {

hermes_state.py

@@ -9175,6 +9175,51 @@ class SessionDB:
             cursor = self._conn.execute(f"SELECT COUNT(*) FROM sessions s{where_sql}", params)
             return cursor.fetchone()[0]
 
+    def session_count_by_source(
+        self,
+        *,
+        include_archived: bool = False,
+        archived_only: bool = False,
+        exclude_children: bool = False,
+    ) -> Dict[str, int]:
+        """Return a ``{source: count}`` dict via a single ``GROUP BY`` query.
+
+        Replaces the O(N) ``list_sessions_rich`` histogram loop with an
+        aggregate query. When ``exclude_children`` is False the query uses
+        ``idx_sessions_source``; when True, the child-exclusion predicates
+        require a full table scan (same as ``session_count`` and
+        ``list_sessions_rich``).
+
+        ``exclude_children=True`` mirrors ``list_sessions_rich`` visibility
+        (roots + branch sessions, excluding sub-agent runs, delegates, and
+        compression continuations) so the source counts match what the
+        Sessions page actually lists.
+        """
+        where_clauses = []
+        params: list = []
+
+        if exclude_children:
+            where_clauses.append(_LISTABLE_CHILD_SQL)
+            where_clauses.append(f"{_delegate_from_json('s.model_config')} IS NULL")
+        if archived_only:
+            where_clauses.append("s.archived = 1")
+        elif not include_archived:
+            where_clauses.append("s.archived = 0")
+
+        where_sql = f" WHERE {' AND '.join(where_clauses)}" if where_clauses else ""
+
+        with self._lock:
+            if self._conn is None:
+                raise RuntimeError("SessionDB connection is closed")
+            rows = self._conn.execute(
+                "SELECT COALESCE(NULLIF(s.source, ''), 'cli') AS source, COUNT(*) AS count "
+                f"FROM sessions s{where_sql} "
+                "GROUP BY COALESCE(NULLIF(s.source, ''), 'cli') "
+                "ORDER BY count DESC",
+                params,
+            ).fetchall()
+        return {str(row["source"]): int(row["count"] or 0) for row in rows}
+
     def message_count(self, session_id: str = None) -> int:
         """Count messages, optionally for a specific session."""
         with self._lock:

tests/hermes_cli/test_dashboard_admin_endpoints.py

@@ -704,6 +704,26 @@ class TestSessionManagementEndpoints:
         assert {"total", "active_store", "archived", "messages", "by_source"} <= set(body)
         assert body["total"] >= 1
 
+    def test_stats_source_counts_use_direct_aggregate(self, monkeypatch):
+        """Source badges must not materialise rich session rows.
+
+        Large stores can have thousands of sessions. The stats endpoint only
+        needs grouped counts, so it should call ``session_count_by_source``
+        instead of ``list_sessions_rich`` and build preview/last-active rows
+        just to count source labels.
+        """
+        from hermes_state import SessionDB
+
+        def fail_list_sessions_rich(self, *args, **kwargs):
+            raise AssertionError("stats should use grouped source counts, not list_sessions_rich")
+
+        monkeypatch.setattr(SessionDB, "list_sessions_rich", fail_list_sessions_rich)
+
+        r = self.client.get("/api/sessions/stats")
+        assert r.status_code == 200
+        body = r.json()
+        assert body["by_source"]["cli"] >= 1
+
     def test_rename(self):
         r = self.client.patch("/api/sessions/sess-x", json={"title": "Renamed"})
         assert r.status_code == 200 and r.json()["title"] == "Renamed"

tests/test_hermes_state.py

@@ -2667,6 +2667,54 @@ class TestCounts:
         assert db.session_count(source="cli") == 2
         assert db.session_count(source="telegram") == 1
 
+    def test_session_count_by_source_group_by(self, db):
+        """session_count_by_source() returns grouped counts via a single query."""
+        db.create_session(session_id="s1", source="cli")
+        db.create_session(session_id="s2", source="telegram")
+        db.create_session(session_id="s3", source="cli")
+        db.create_session(session_id="s4", source="discord")
+        result = db.session_count_by_source(include_archived=True)
+        assert result == {"cli": 2, "telegram": 1, "discord": 1}
+
+    def test_session_count_by_source_empty(self, db):
+        """Empty database returns empty dict."""
+        assert db.session_count_by_source() == {}
+
+    def test_session_count_by_source_excludes_children(self, db):
+        """exclude_children=True hides subagent runs and compression continuations.
+
+        Mirrors list_sessions_rich visibility so the source histogram matches
+        what the Sessions page actually lists, not raw row counts.
+        """
+        db.create_session(session_id="root", source="cli")
+        # Child session (subagent run) — should be excluded.
+        db.create_session(
+            session_id="child", source="cli", parent_session_id="root"
+        )
+        # End the parent so the child looks like a subagent run (not a branch).
+        db.end_session("root", "ended")
+
+        without_children = db.session_count_by_source(exclude_children=True)
+        assert without_children == {"cli": 1}
+
+        with_children = db.session_count_by_source(include_archived=True)
+        assert with_children == {"cli": 2}
+
+    def test_session_count_by_source_coalesce_groups_null(self, db):
+        """GROUP BY COALESCE(source, 'cli') avoids duplicate-key data loss.
+
+        If NULL source rows existed (schema is NOT NULL, but defence-in-depth),
+        NULL and literal 'cli' must collapse to a single 'cli' key — not two
+        separate groups that the dict comprehension silently drops.
+        """
+        db.create_session(session_id="s1", source="cli")
+        # Inject a NULL source directly (bypassing the NOT NULL constraint
+        # would fail on a real db, so just verify the COALESCE works on
+        # normal data — the aliasing is the structural invariant).
+        result = db.session_count_by_source(include_archived=True)
+        assert "cli" in result
+        assert result["cli"] == 1
+
     def test_session_count_by_cwd_prefix(self, db):
         db.create_session("s1", "cli", cwd="/repo")
         db.create_session("s2", "cli", cwd="/repo-wt-feature")