perf(state): index assistant tool-call rows for Insights queries

InsightsEngine._get_tool_usage and _get_skill_usage scan messages for
role='assistant' AND tool_calls IS NOT NULL, but no index aligns with
that predicate, so SQLite scans the full messages table on a large
state.db. Add a partial index over exactly those rows.

role and tool_calls are base columns in the messages table, so the index
lives in SCHEMA_SQL (created on both fresh and existing databases via the
executescript on every open) rather than DEFERRED_INDEX_SQL.

Adds schema regression coverage (fresh + reopened DB, plan uses the index)
and an Insights regression test proving tool/skill output is identical with
and without the index present.

Fixes #67341

hermes_state_common.py

@@ -337,6 +337,14 @@ CREATE INDEX IF NOT EXISTS idx_sessions_parent ON sessions(parent_session_id);
 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_messages_session_id ON messages(session_id, id);
+-- Partial index for the Insights assistant tool-call scan
+-- (agent/insights.py _get_tool_usage / _get_skill_usage): those queries filter
+-- messages by role='assistant' AND tool_calls IS NOT NULL, a small fraction of
+-- rows on a large state.db. role and tool_calls are base columns, so this can
+-- live in SCHEMA_SQL rather than DEFERRED_INDEX_SQL.
+CREATE INDEX IF NOT EXISTS idx_messages_assistant_calls_by_session
+    ON messages(session_id)
+    WHERE role = 'assistant' AND tool_calls IS NOT NULL;
 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);

tests/agent/test_insights.py

@@ -368,6 +368,31 @@ class TestInsightsPopulated:
 
 
 
+    def test_tool_and_skill_usage_invariant_to_partial_index(self, populated_db):
+        """The assistant tool-call partial index is a pure optimization: tool
+        and skill usage must be byte-for-byte identical with and without it."""
+        engine = InsightsEngine(populated_db)
+        cutoff = 0.0
+        index = "idx_messages_assistant_calls_by_session"
+
+        with_index_tools = engine._get_tool_usage(cutoff)
+        with_index_skills = engine._get_skill_usage(cutoff)
+        # The index must exist by default (it lives in SCHEMA_SQL).
+        assert populated_db._conn.execute(
+            "SELECT 1 FROM sqlite_master WHERE type = 'index' AND name = ?",
+            (index,),
+        ).fetchone() is not None
+
+        populated_db._conn.execute(f"DROP INDEX IF EXISTS {index}")
+        populated_db._conn.commit()
+        without_index_tools = engine._get_tool_usage(cutoff)
+        without_index_skills = engine._get_skill_usage(cutoff)
+
+        assert with_index_tools == without_index_tools
+        assert with_index_skills == without_index_skills
+        # Sanity: the fixture actually exercises the assistant tool_calls path.
+        assert any(t["tool_name"] == "search_files" for t in with_index_tools)
+
 
 # =========================================================================
 # Formatting

tests/test_hermes_state.py

@@ -3657,3 +3657,79 @@ class TestApplyDatabasePragmas:
             assert conn.execute("PRAGMA wal_autocheckpoint").fetchone()[0] == before
         finally:
             conn.close()
+class TestInsightsToolCallIndex:
+    """The Insights assistant tool-call scan has a predicate-aligned index.
+
+    ``InsightsEngine._get_tool_usage`` / ``_get_skill_usage`` filter messages by
+    ``role = 'assistant' AND tool_calls IS NOT NULL``.  A partial index over that
+    predicate keeps the scan off the full ``messages`` table on a large state.db.
+    """
+
+    _INDEX = "idx_messages_assistant_calls_by_session"
+
+    def _index_defn(self, conn):
+        row = conn.execute(
+            "SELECT sql FROM sqlite_master WHERE type = 'index' AND name = ?",
+            (self._INDEX,),
+        ).fetchone()
+        return row["sql"] if row else None
+
+    def test_index_created_on_fresh_db(self, tmp_path):
+        db = SessionDB(db_path=tmp_path / "fresh.db")
+        try:
+            sql = self._index_defn(db._conn)
+            assert sql is not None, "partial index missing on a fresh database"
+            # Partial predicate must match the queried rows exactly.
+            assert "role = 'assistant'" in sql
+            assert "tool_calls IS NOT NULL" in sql
+        finally:
+            db.close()
+
+    def test_index_created_on_existing_db(self, tmp_path):
+        """Reopening a DB that predates the index must create it (SCHEMA_SQL is
+        re-run on every open; role/tool_calls are original base columns)."""
+        db_path = tmp_path / "legacy.db"
+        db = SessionDB(db_path=db_path)
+        # Simulate a database created before the index shipped.
+        db._conn.execute(f"DROP INDEX IF EXISTS {self._INDEX}")
+        db._conn.commit()
+        assert self._index_defn(db._conn) is None
+        db.close()
+
+        db2 = SessionDB(db_path=db_path)
+        try:
+            assert self._index_defn(db2._conn) is not None, (
+                "index not recreated when reopening an existing database"
+            )
+        finally:
+            db2.close()
+
+    def test_index_serves_assistant_tool_call_scan(self, db):
+        """The planner uses the partial index for the exact Insights predicate.
+
+        Seed enough rows that the optimizer prefers the partial index over a
+        full table scan, then assert the plan names it.
+        """
+        db.create_session(session_id="s1", source="cli")
+        for i in range(200):
+            if i % 5 == 0:
+                db.append_message(
+                    "s1", role="assistant", content=f"m{i}",
+                    tool_calls=[{"function": {"name": "search_files"}}],
+                )
+            else:
+                db.append_message("s1", role="user", content=f"m{i}")
+        db._conn.execute("ANALYZE")
+        db._conn.commit()
+
+        plan = "\n".join(
+            row["detail"]
+            for row in db._conn.execute(
+                "EXPLAIN QUERY PLAN "
+                "SELECT m.tool_calls FROM messages m "
+                "JOIN sessions s ON s.id = m.session_id "
+                "WHERE s.started_at >= 0 "
+                "AND m.role = 'assistant' AND m.tool_calls IS NOT NULL"
+            ).fetchall()
+        )
+        assert self._INDEX in plan, plan