diff --git a/hermes_state_common.py b/hermes_state_common.py index e62fa6fa7c..19260b42de 100644 --- a/hermes_state_common.py +++ b/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})") diff --git a/hermes_state_portability.py b/hermes_state_portability.py index 96953cb604..f622b15118 100644 --- a/hermes_state_portability.py +++ b/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} diff --git a/hermes_state_sessions.py b/hermes_state_sessions.py index 4fe2349685..b70536ebd6 100644 --- a/hermes_state_sessions.py +++ b/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" diff --git a/hermes_state_telegram.py b/hermes_state_telegram.py index 09e492d2a6..8063cc7cb4 100644 --- a/hermes_state_telegram.py +++ b/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 diff --git a/tests/hermes_state/test_session_list_index_choice.py b/tests/hermes_state/test_session_list_index_choice.py index b6b511b58d..d99e96031f 100644 --- a/tests/hermes_state/test_session_list_index_choice.py +++ b/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