refactor(state): one preview-column builder, pinned at every session listing (#119403)

Review of the previous commit found a fourth copy of the preview
subquery, in list_recent_sessions_bounded (session_search's recent
browse, correlated on tip.id). It still had a bare FROM messages m, so it
kept the stale-stats plan. The three copies the previous commit patched
differed from _PREVIEW_RAW_SUBQUERY_SQL only in whitespace, and no test
pins their SQL text.

_sql_preview_raw(session_id_expr) now builds the column for the sidebar,
the bounded recent browse, the telegram unlinked listing and the export
select, so the INDEXED BY lives in one place. The regression test now
covers the sidebar, session_search and telegram readers. It asserts that
both the preview (m) and last-active (_act_m) searches run, and that each
one goes only through idx_messages_session. All three cases fail on
origin/main and when the hint is dropped from the helper.

hermes_state_common.py

@@ -155,11 +155,13 @@ def _shape_preview(raw: Any) -> str:
     return text[:_PREVIEW_MAX_CHARS] + "..." if len(text) > _PREVIEW_MAX_CHARS else text
 
 
-# Correlated ``_preview_raw`` column for a ``sessions s`` row. ``INDEXED BY`` walks the session in timestamp
-# order and stops at the first eligible row; the planner otherwise may sort the whole session (#119403).
-_PREVIEW_RAW_SUBQUERY_SQL = (f"COALESCE((SELECT {_PREVIEW_RAW_SELECT} FROM messages m INDEXED BY idx_messages_session"
-    f" WHERE m.session_id = s.id AND m.role = 'user' AND m.content IS NOT NULL AND {_PREVIEW_ELIGIBLE_SQL}"
-    f" ORDER BY m.timestamp, m.id LIMIT 1), '') AS _preview_raw")
+def _sql_preview_raw(session_id_expr: str = "s.id") -> str:
+    """Correlated ``_preview_raw`` column: the first eligible user message of *session_id_expr*.
+    ``INDEXED BY`` walks the session in timestamp order and stops at the first eligible row; with planner
+    stats older than idx_messages_session_id, SQLite otherwise reads and sorts the whole session (#119403)."""
+    return (f"COALESCE((SELECT {_PREVIEW_RAW_SELECT} FROM messages m INDEXED BY idx_messages_session"
+        f" WHERE m.session_id = {session_id_expr} AND m.role = 'user' AND m.content IS NOT NULL"
+        f" AND {_PREVIEW_ELIGIBLE_SQL} ORDER BY m.timestamp, m.id LIMIT 1), '') AS _preview_raw")
 
 # ── Session lineage predicates ({a} = sessions alias) ───────────────────────
 
@@ -257,8 +259,7 @@ def _sql_freshest_of(activity: str, session_id_expr: str, started: str) -> str:
     Cells outside the ``coerce_epoch`` window (garbage doubles salvaged from a damaged page, TEXT) are
     skipped, fallback included, or one bad row pins the session's recency to ``5e+246`` (#91536); a
     session with no trusted cell at all is NULL."""
-    # Pinned to the (session_id, timestamp) index: with planner stats older than idx_messages_session_id,
-    # SQLite picks that non-covering index and reads every message row of every listed session (#119403).
+    # Pinned for the same reason as _sql_preview_raw (#119403).
     msg_max = (f"(SELECT MAX(_act_m.timestamp) FROM messages _act_m INDEXED BY idx_messages_session"
         f" WHERE _act_m.session_id = {session_id_expr}"
         f" AND _act_m.timestamp {_SQL_IN_WINDOW})")

hermes_state_portability.py

@@ -15,7 +15,7 @@ from agent.skill_commands import AUTO_LOAD_SCAFFOLD_SQL_LIKE, SKILL_SCAFFOLD_SQL
 from utils import safe_json_loads
 from hermes_cli.timefmt import coerce_epoch
 from hermes_state_ids import new_session_id
-from hermes_state_common import SCHEMA_SQL, _PREVIEW_RAW_SUBQUERY_SQL, _shape_preview, _sql_session_last_active
+from hermes_state_common import SCHEMA_SQL, _shape_preview, _sql_preview_raw, _sql_session_last_active
 from hermes_state_messages import _parse_tool_calls, _tool_calls_count
 
 # Pre-split logger identity so log filtering/capture is unchanged.
@@ -73,7 +73,7 @@ def _rich_select(select_cols: str, where: str, tail: str = "", prompt_select: Op
     prompt_join = "" if prompt_select is None else "LEFT JOIN system_prompts sp ON sp.hash = s.system_prompt_hash"
     return f"""
             SELECT {select_cols}{prompt_select or ""},
-                {_PREVIEW_RAW_SUBQUERY_SQL},
+                {_sql_preview_raw()},
                 {_sql_session_last_active("s")} AS last_active
             FROM sessions s
             {prompt_join}

hermes_state_sessions.py

@@ -17,9 +17,9 @@ from agent.session_activity import (
 from hermes_startup_watchdog import report_startup_progress
 from hermes_state_errors import SessionActiveWriteGuardError
 from hermes_state_common import (
-    _LISTABLE_CHILD_SQL, _PREVIEW_ELIGIBLE_SQL, _PREVIEW_RAW_SELECT, _RECOVERABLE_END_REASONS,
+    _LISTABLE_CHILD_SQL, _RECOVERABLE_END_REASONS,
     _RECOVERABLE_END_REASONS_SQL, _RESET_CHILD_SQL, _RESET_END_REASONS, _legacy_reset_child_sql, _non_continuation_child_sql,
-    _shape_preview, QUEUED_PROMPT_METADATA_KEY,
+    _shape_preview, _sql_preview_raw, QUEUED_PROMPT_METADATA_KEY,
     _sql_in_window, _sql_json_extract, _sql_session_last_active, _sql_session_last_active_by_id,
     escape_like as _escape_like, _SQL_IN_CHUNK, _id_chunks, _placeholders as _session_ids_placeholders,
 )
@@ -103,18 +103,6 @@ def _workspace_key_clause(key: str) -> Tuple[str, List[str]]:
     )
 
 
-# First user message of a session, shaped by _shape_preview() in Python.
-# The indentation is part of the list_sessions_rich SQL text.
-_PREVIEW_COL_SQL = f"""COALESCE(
-                        (SELECT {_PREVIEW_RAW_SELECT}
-                         FROM messages m INDEXED BY idx_messages_session
-                         WHERE m.session_id = s.id AND m.role = 'user' AND m.content IS NOT NULL
-                           AND {_PREVIEW_ELIGIBLE_SQL}
-                         ORDER BY m.timestamp, m.id LIMIT 1),
-                        ''
-                    ) AS _preview_raw"""
-
-
 def _where_sql(clauses: List[str], lead: str = "") -> str:
     """``WHERE a AND b`` (with *lead* prefix) or "" when there are no clauses."""
     return f"{lead}WHERE {' AND '.join(clauses)}" if clauses else ""
@@ -1245,16 +1233,7 @@ class SessionSessionsMixin:
                 tip.message_count,
                 tip.tool_call_count,
                 rt.activity AS last_active,
-                COALESCE(
-                    (SELECT {_PREVIEW_RAW_SELECT}
-                     FROM messages m
-                     WHERE m.session_id = tip.id
-                       AND m.role = 'user'
-                       AND m.content IS NOT NULL
-                       AND {_PREVIEW_ELIGIBLE_SQL}
-                     ORDER BY m.timestamp, m.id LIMIT 1),
-                    ''
-                ) AS _preview_raw,
+                {_sql_preview_raw('tip.id')},
                 CASE WHEN s.id != tip.id THEN s.id ELSE NULL END
                     AS _lineage_root_id
             FROM ranked_tips rt
@@ -1345,7 +1324,7 @@ class SessionSessionsMixin:
         select_head = (
             f"SELECT {self._compact_session_cols() if compact_rows else 's.*'}"
             + ("" if compact_rows else ", COALESCE(sp.prompt, s.system_prompt) AS _system_prompt_resolved")
-            + f",\n                    {_PREVIEW_COL_SQL},\n                    "
+            + f",\n                    {_sql_preview_raw()},\n                    "
         )
         prompt_join = (
             "" if compact_rows else "LEFT JOIN system_prompts sp ON sp.hash = s.system_prompt_hash"

hermes_state_telegram.py

@@ -8,7 +8,7 @@ import sqlite3
 import time
 from typing import Any, Dict, List, Optional
 
-from hermes_state_common import _PREVIEW_ELIGIBLE_SQL, _PREVIEW_RAW_SELECT, _sql_session_last_active
+from hermes_state_common import _sql_preview_raw, _sql_session_last_active
 from hermes_state_errors import StateDbReplacedError
 
 # caplog tests pin the "hermes_state" logger name.
@@ -69,14 +69,7 @@ _UNLINKED_SELECT_HEAD = f"""
                     SELECT s.*,
                         COALESCE(sp.prompt, s.system_prompt)
                             AS _system_prompt_resolved,
-                        COALESCE(
-                            (SELECT {_PREVIEW_RAW_SELECT}
-                             FROM messages m INDEXED BY idx_messages_session
-                             WHERE m.session_id = s.id AND m.role = 'user' AND m.content IS NOT NULL
-                               AND {_PREVIEW_ELIGIBLE_SQL}
-                             ORDER BY m.timestamp, m.id LIMIT 1),
-                            ''
-                        ) AS _preview_raw,
+                        {_sql_preview_raw()},
                         {_sql_session_last_active("s")} AS last_active
                     FROM sessions s
                     LEFT JOIN system_prompts sp

tests/hermes_state/test_session_list_index_choice.py

@@ -2,19 +2,29 @@
 
 A ``sqlite_stat1`` written before ``idx_messages_session_id`` existed has no row for that index, so the
 planner guesses it is selective and routes the per-session last-active and preview subqueries through
-it: every listed session then reads all of its message rows (≈1 GB per list on a 125k-message store).
+it: every listed session then reads all of its message rows (≈0.5 GB per sidebar list on a 125k-message
+store before this fix).
 """
 
+import re
 import sqlite3
 
+import pytest
+
 from hermes_state import SessionDB
 
+_READERS = {
+    "sidebar": lambda db: db.list_sessions_rich(limit=2, order_by_last_active=True, include_pinned=True),
+    "session_search": lambda db: db.list_recent_sessions_bounded(limit=2),
+    "telegram_unlinked": lambda db: db.list_unlinked_telegram_sessions_for_user(chat_id="c", user_id="u", limit=2),
+}
+
 
 def _stale_stats_db(path):
     db = SessionDB(db_path=path)
     for i in range(4):
         sid = f"s{i}"
-        db.create_session(sid, "cli")
+        db.create_session(sid, "telegram", user_id="u")
         db.append_messages_batch(sid, [
             {"role": "user" if j % 2 == 0 else "assistant", "content": f"{sid} message {j}", "timestamp": 1_000 + j}
             for j in range(40)
@@ -28,7 +38,8 @@ def _stale_stats_db(path):
     raw.close()
 
 
-def test_session_list_never_routes_messages_through_the_id_index(tmp_path, monkeypatch):
+@pytest.mark.parametrize("reader", sorted(_READERS))
+def test_session_listings_search_messages_only_through_the_timestamp_index(tmp_path, monkeypatch, reader):
     path = tmp_path / "state.db"
     _stale_stats_db(path)
     statements = []
@@ -42,15 +53,16 @@ def test_session_list_never_routes_messages_through_the_id_index(tmp_path, monke
     monkeypatch.setattr(sqlite3, "connect", traced_connect)
     db = SessionDB(db_path=path)
     statements.clear()
-    rows = db.list_sessions_rich(limit=2, order_by_last_active=True, include_pinned=True)
+    rows = _READERS[reader](db)
     db.close()
     monkeypatch.undo()
 
-    assert {r["id"] for r in rows} >= {"s0"} and all(r["preview"] for r in rows)
+    assert rows and all(r["preview"] for r in rows)
     reads = [s for s in statements if s.lstrip().split(None, 1)[0].upper() in ("SELECT", "WITH")]
     plan_conn = sqlite3.connect(f"file:{path}?mode=ro", uri=True)
     plans = [row[3] for sql in reads for row in plan_conn.execute("EXPLAIN QUERY PLAN " + sql)]
     plan_conn.close()
-    message_searches = [p for p in plans if "idx_messages_session" in p]
-    assert message_searches, plans
-    assert not [p for p in message_searches if "idx_messages_session_id" in p], message_searches
+    message_searches = [p for p in plans if re.match(r"(SEARCH|SCAN) (m|_act_m)\b", p)]
+    # Preview (m) and last-active (_act_m) both run, and both only through idx_messages_session.
+    assert {p.split()[1] for p in message_searches} == {"m", "_act_m"}, plans
+    assert all(re.search(r"INDEX idx_messages_session \(", p) for p in message_searches), message_searches