diff --git a/api/agent_sessions.py b/api/agent_sessions.py index 31e32fd9..91664f5e 100644 --- a/api/agent_sessions.py +++ b/api/agent_sessions.py @@ -19,7 +19,6 @@ MESSAGING_SOURCES = { CLI_MIN_UNTITLED_MESSAGE_COUNT = 6 CLI_MIN_UNTITLED_USER_MESSAGE_COUNT = 2 -_CRON_PREAGGREGATE_CANDIDATE_ORDER_MIN_MESSAGES = 100000 SOURCE_LABELS = { 'api_server': 'API', @@ -493,37 +492,28 @@ def read_importable_agent_session_rows( use_messages_join = messages_has_session_id count_col = 'id' if 'id' in message_cols else 'session_id' - # Defensive index prime (#3887). The candidate-ordering path below now - # builds a pre-aggregated latest-message relation over ``messages`` and - # the final grouped projection still joins ``messages`` by ``session_id``. - # Both stay fast only when the agent's standard - # ``idx_messages_session ON messages(session_id, timestamp)`` index exists. - # A normally-migrated hermes-agent state.db has it, but a db that lost its - # migrations (older hermes-agent, or a hand-rebuilt/reimported db) does - # not — and the aggregate scan then degrades into a much more expensive - # walk over ``messages``, stalling ``/api/sessions`` on large stores. - # Priming the index is a no-op (~free) when it already exists, and - # self-heals an affected db in milliseconds. Best-effort: degrade - # silently on a read-only db or any error so the listing never fails - # because of the prime. - messages_index_ready = False - messages_rowid_upper_bound = None + # Defensive index prime (#3887). The normal candidate-ordering shape uses + # the agent's standard ``idx_messages_session ON messages(session_id, + # timestamp)`` index; without it, large cron-only scans degrade badly. + # Writable dbs self-heal by recreating the index. Read-only or locked dbs + # fall back to the pre-aggregated cron-only path below instead of failing. + messages_index_present = False if messages_has_session_id and messages_has_timestamp: try: - cur.execute( - "CREATE INDEX IF NOT EXISTS idx_messages_session " - "ON messages(session_id, timestamp)" - ) - conn.commit() - messages_index_ready = True + cur.execute("PRAGMA index_list(messages)") + messages_index_present = any(str(row[1]) == "idx_messages_session" for row in cur.fetchall()) + except sqlite3.Error: + messages_index_present = False + try: + if not messages_index_present: + cur.execute( + "CREATE INDEX IF NOT EXISTS idx_messages_session " + "ON messages(session_id, timestamp)" + ) + conn.commit() + messages_index_present = True except sqlite3.Error: pass # read-only db / locked / older schema — degrade gracefully - try: - cur.execute("SELECT MAX(rowid) FROM messages") - row = cur.fetchone() - messages_rowid_upper_bound = int(row[0] or 0) if row else 0 - except sqlite3.Error: - messages_rowid_upper_bound = None if use_messages_join: actual_count_expr = f"COUNT(m.{count_col})" @@ -567,10 +557,7 @@ def read_importable_agent_session_rows( use_messages_join and messages_has_timestamp and included == ("cron",) - and ( - not messages_index_ready - or (messages_rowid_upper_bound or 0) >= _CRON_PREAGGREGATE_CANDIDATE_ORDER_MIN_MESSAGES - ) + and not messages_index_present ) if use_preaggregated_candidate_order: order_by_clause = "ORDER BY COALESCE(MAX(m.timestamp), s.started_at) DESC" diff --git a/tests/test_issue2628_cli_sessions_perf.py b/tests/test_issue2628_cli_sessions_perf.py index 608d9242..70f4513f 100644 --- a/tests/test_issue2628_cli_sessions_perf.py +++ b/tests/test_issue2628_cli_sessions_perf.py @@ -91,7 +91,15 @@ def _newest_first_reference_ids(db_path, *, include_sources=None, exclude_source conn.close() -def _execute_candidate_ordering_baseline_sql(db_path, candidate_limit, *, budget_ops, include_sources=None, exclude_sources=("webui",)): +def _execute_candidate_ordering_baseline_sql( + db_path, + candidate_limit, + *, + budget_ops, + interval=1, + include_sources=None, + exclude_sources=("webui",), +): where_clauses = ["s.source IS NOT NULL"] params = [] if include_sources: @@ -110,7 +118,7 @@ def _execute_candidate_ordering_baseline_sql(db_path, candidate_limit, *, budget steps = 0 conn = sqlite3.connect(f"file:{db_path}?mode=ro&immutable=1", uri=True) - conn.set_progress_handler(_on_progress, 1) + conn.set_progress_handler(_on_progress, interval) try: return conn.execute( f""" @@ -200,28 +208,35 @@ def test_importable_agent_rows_push_sidebar_limit_into_sql(tmp_path): assert "WITH candidates AS" in src assert "JOIN candidates c ON c.id = s.id" in src assert "latest_messages AS" in src + assert "LEFT JOIN latest_messages lm ON lm.session_id = s.id" in src assert 'included == ("cron",)' in src - assert "_CRON_PREAGGREGATE_CANDIDATE_ORDER_MIN_MESSAGES" in src + assert "not messages_index_present" in src + assert "PRAGMA index_list(messages)" in src + assert "CREATE INDEX IF NOT EXISTS idx_messages_session" in src + assert "_CRON_PREAGGREGATE_CANDIDATE_ORDER_MIN_MESSAGES" not in src assert "MAX(mx.timestamp) FROM messages mx WHERE mx.session_id = s.id" in src assert "candidate_limit = max(result_limit * 8, result_limit)" in src -def test_importable_agent_rows_candidate_ordering_respects_progress_budget(tmp_path, monkeypatch): - """Cron-only candidate ordering should fail under the old shape budget, then pass after pre-aggregation.""" +def test_importable_agent_rows_candidate_ordering_stays_under_progress_budget(tmp_path, monkeypatch): + """Cron-only missing-index scans should fail under the old shape budget, then pass after pre-aggregation.""" db = tmp_path / "state.db" _make_state_db( db, sessions=120, messages_per_session=900, - create_messages_index=True, + create_messages_index=False, source="cron", session_source="cron", ) reference_ids = _newest_first_reference_ids(db, include_sources=("cron",), exclude_sources=None) candidate_limit = max(20 * 8, 20) + progress_interval = 100 original_connect = agent_sessions.sqlite3.connect - connect_with_progress_counter, progress_counter = _make_connect_with_progress_counter(interval=1) + connect_with_progress_counter, progress_counter = _make_connect_with_progress_counter( + interval=progress_interval + ) monkeypatch.setattr(agent_sessions.sqlite3, "connect", connect_with_progress_counter) try: measured_rows = agent_sessions.read_importable_agent_session_rows( @@ -236,15 +251,16 @@ def test_importable_agent_rows_candidate_ordering_respects_progress_budget(tmp_p # counting handler to validate raw cost differences. monkeypatch.setattr(agent_sessions.sqlite3, "connect", original_connect) - # Give the head path a deterministic margin while preserving a strict cap - # that a single-pass candidate-ordering query can still satisfy. - progress_budget_ops = max(progress_counter["count"] + 2500, 1) + # Give the head path a small deterministic margin, then require the old + # correlated query to exceed the same budget on the missing-index branch. + progress_budget_ops = max(progress_counter["count"] + 200, 1) with pytest.raises(sqlite3.OperationalError, match="interrupted"): _execute_candidate_ordering_baseline_sql( db, candidate_limit, budget_ops=progress_budget_ops, + interval=progress_interval, include_sources=("cron",), exclude_sources=None, ) @@ -254,7 +270,7 @@ def test_importable_agent_rows_candidate_ordering_respects_progress_budget(tmp_p "connect", _make_connect_with_progress_budget( budget_ops=progress_budget_ops, - interval=1, + interval=progress_interval, ), )