From 0e2214b6bdcd5b2fcbcee496282c409ea0f6e591 Mon Sep 17 00:00:00 2001 From: Frank Song Date: Wed, 24 Jun 2026 16:59:46 +0800 Subject: [PATCH 1/3] Fix WebUI state.db session reconciliation --- api/models.py | 101 +++++++++++++++++- api/routes.py | 11 +- tests/test_session_sidebar_cache.py | 52 ++++++--- tests/test_streaming_session_sidebar.py | 38 +++++++ ...t_webui_state_db_context_reconciliation.py | 14 ++- tests/test_webui_state_db_reconciliation.py | 99 +++++++++++++++++ 6 files changed, 296 insertions(+), 19 deletions(-) diff --git a/api/models.py b/api/models.py index 3b0220b580a..8015809b9bc 100644 --- a/api/models.py +++ b/api/models.py @@ -3298,6 +3298,18 @@ def _read_state_db_sidebar_overrides(db_path: Path, session_ids: set[str]) -> di source_expr = 's.source' if 'source' in session_cols else 'NULL AS source' session_source_expr = 's.session_source' if 'session_source' in session_cols else 'NULL AS session_source' title_expr = 's.title' if 'title' in session_cols else 'NULL AS title' + message_count_expr = 's.message_count' if 'message_count' in session_cols else 'NULL AS message_count' + + cur.execute("SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'messages'") + has_messages_table = cur.fetchone() is not None + messages_has_session_id = False + messages_has_timestamp = False + if has_messages_table: + cur.execute("PRAGMA table_info(messages)") + message_cols = {str(row[1]) for row in cur.fetchall()} + messages_has_session_id = 'session_id' in message_cols + messages_has_timestamp = 'timestamp' in message_cols + overrides: dict[str, dict] = {} ids = list(wanted) chunk_size = 500 @@ -3306,7 +3318,7 @@ def _read_state_db_sidebar_overrides(db_path: Path, session_ids: set[str]) -> di placeholders = ','.join('?' * len(chunk)) cur.execute( f""" - SELECT s.id, {source_expr}, {session_source_expr}, {title_expr} + SELECT s.id, {source_expr}, {session_source_expr}, {title_expr}, {message_count_expr} FROM sessions s WHERE s.id IN ({placeholders}) """, @@ -3326,8 +3338,39 @@ def _read_state_db_sidebar_overrides(db_path: Path, session_ids: set[str]) -> di entry['_state_db_raw_source'] = source_meta.get('raw_source') entry['_state_db_session_source'] = source_meta.get('session_source') entry['_state_db_source_label'] = source_meta.get('source_label') + if row['message_count'] is not None: + try: + entry['_state_db_message_count'] = max(0, int(row['message_count'] or 0)) + except (TypeError, ValueError): + pass if entry: overrides[sid] = entry + if has_messages_table and messages_has_session_id: + last_at_expr = "MAX(timestamp) AS last_message_at" if messages_has_timestamp else "NULL AS last_message_at" + cur.execute( + f""" + SELECT session_id, COUNT(*) AS actual_message_count, {last_at_expr} + FROM messages + WHERE session_id IN ({placeholders}) + GROUP BY session_id + """, + chunk, + ) + for row in cur.fetchall(): + sid = str(row['session_id']) + entry = overrides.setdefault(sid, {}) + try: + entry['_state_db_message_count'] = max( + int(entry.get('_state_db_message_count') or 0), + int(row['actual_message_count'] or 0), + ) + except (TypeError, ValueError): + pass + if row['last_message_at'] is not None: + try: + entry['_state_db_last_message_at'] = float(row['last_message_at'] or 0) + except (TypeError, ValueError): + pass return overrides except Exception: return {} @@ -3357,12 +3400,40 @@ def _apply_sidebar_state_db_override_metadata(sessions: list[dict], metadata: di state_db_raw_source = entry.pop('_state_db_raw_source', None) state_db_session_source = entry.pop('_state_db_session_source', None) state_db_source_label = entry.pop('_state_db_source_label', None) + state_db_message_count = entry.pop('_state_db_message_count', None) + state_db_last_message_at = entry.pop('_state_db_last_message_at', None) if state_db_source == 'webui': session['source_tag'] = state_db_source_tag session['raw_source'] = state_db_raw_source session['session_source'] = state_db_session_source session['source_label'] = state_db_source_label session['is_cli_session'] = False + try: + current_count = max(0, int(session.get('message_count') or 0)) + state_count = max(0, int(state_db_message_count or 0)) + except (TypeError, ValueError): + current_count = 0 + state_count = 0 + try: + current_last = max( + float(session.get('last_message_at') or 0), + float(session.get('updated_at') or 0), + ) + except (TypeError, ValueError): + current_last = 0.0 + try: + state_last = float(state_db_last_message_at or 0) + except (TypeError, ValueError): + state_last = 0.0 + if state_count > current_count and (state_last <= 0 or state_last > current_last): + session['message_count'] = state_count + session['actual_message_count'] = max( + state_count, + int(session.get('actual_message_count') or 0) if str(session.get('actual_message_count') or '').isdigit() else 0, + ) + if state_last > 0: + session['last_message_at'] = max(float(session.get('last_message_at') or 0), state_last) + session['updated_at'] = max(float(session.get('updated_at') or 0), state_last) title = session.get('title') if ( state_db_title @@ -3489,6 +3560,34 @@ def all_sessions(diag=None, *, include_lineage_metadata: bool = True): include_runtime=True, active_stream_ids=active_stream_ids, ) + missing_persisted_ids = [] + if persisted_ids is not None: + indexed_ids = {str(sid) for sid in index_map.keys() if sid} + missing_persisted_ids = sorted( + str(sid) for sid in persisted_ids + if sid and str(sid) not in indexed_ids + ) + recovered_sidecars = [] + if missing_persisted_ids: + _diag_stage(diag, "all_sessions.recover_missing_index_sidecars") + for sid in missing_persisted_ids: + try: + sidecar = Session.load_metadata_only(sid) + except Exception: + sidecar = None + if not sidecar: + continue + index_map[sidecar.session_id] = sidecar.compact( + include_runtime=True, + active_stream_ids=active_stream_ids, + ) + recovered_sidecars.append(sidecar) + if recovered_sidecars: + try: + _diag_stage(diag, "all_sessions.recover_missing_index_write") + _write_session_index(updates=recovered_sidecars) + except Exception: + logger.debug("Failed to persist recovered sidebar index rows") _diag_stage(diag, "all_sessions.refresh_sidecar_metadata") index_message_counts = _index_message_count_map(index) refreshed_index_rows = _refresh_index_rows_from_sidecar_metadata( diff --git a/api/routes.py b/api/routes.py index 66402609963..bf16d9569c4 100644 --- a/api/routes.py +++ b/api/routes.py @@ -1703,6 +1703,9 @@ def _session_list_cache_get( return None, False ts, stamp, payload = entry if stamp != current_stamp: + if allow_stale: + _SESSIONS_CACHE.move_to_end(key) + return copy.deepcopy(payload), False _SESSIONS_CACHE.pop(key, None) return None, False # #4808: widen the freshness window while a turn is streaming so the fixed @@ -1826,9 +1829,7 @@ def _session_list_cache_streaming_freeze_marker(): def _session_list_cache_source_stamp(key: tuple) -> tuple[tuple[int, int], tuple[int, int], tuple[int, int], tuple[int, int], tuple[int, int], object, int]: - _cache_profile, _cache_all_profiles, cache_show_cli_sessions, *_rest = key - if not cache_show_cli_sessions: - return ((0, 0), (0, 0), (0, 0), (0, 0), (0, 0), None, 0) + _cache_profile, _cache_all_profiles, _cache_show_cli_sessions, *_rest = key try: settings_file = SETTINGS_FILE except Exception: @@ -1838,6 +1839,10 @@ def _session_list_cache_source_stamp(key: tuple) -> tuple[tuple[int, int], tuple swv = _SETTINGS_WRITE_VERSION except Exception: swv = 0 + # WebUI-origin sessions can also receive settled rows in state.db when the + # official Hermes Desktop App continues the same agent session. The sidebar + # therefore watches state.db even when the CLI/external-session tab is hidden. + # # Streaming hold-down (#4672): while a turn is in flight, collapse the # volatile state.db-derived components (db/WAL stat, gateway metadata, index # stat, content fingerprint) to a marker that only changes when a stream diff --git a/tests/test_session_sidebar_cache.py b/tests/test_session_sidebar_cache.py index 49f48c073c0..11896363bdc 100644 --- a/tests/test_session_sidebar_cache.py +++ b/tests/test_session_sidebar_cache.py @@ -142,6 +142,15 @@ def builder(): follower_diag = _StageRecorder() owner_diag = _StageRecorder() + wait_seen = threading.Event() + original_follower_stage = follower_diag.stage + + def follower_stage(name): + original_follower_stage(name) + if name == "session_list_cache_wait": + wait_seen.set() + + follower_diag.stage = follower_stage def owner(): routes._get_cached_session_list_payload( @@ -159,18 +168,21 @@ def follower(): owner_thread = threading.Thread(target=owner) follower_thread = threading.Thread(target=follower) - owner_thread.start() - assert started.wait(1.0) - follower_thread.start() - release.set() - owner_thread.join(2) - follower_thread.join(2) + try: + owner_thread.start() + assert started.wait(1.0) + follower_thread.start() + assert wait_seen.wait(1.0) + finally: + release.set() + owner_thread.join(2) + follower_thread.join(2) assert "session_list_cache_wait" in follower_diag.stages assert "session_list_cache_hit" in owner_diag.stages or "session_list_cache_stored" in owner_diag.stages -def test_session_list_cache_follower_reuses_stale_payload_during_slow_rebuild(): +def test_session_list_cache_follower_reuses_stale_payload_during_slow_rebuild(monkeypatch): routes._session_list_cache_clear() key = routes._session_list_cache_key( @@ -188,6 +200,15 @@ def test_session_list_cache_follower_reuses_stale_payload_during_slow_rebuild(): stamp, payload, ) + # Simulate the #4834 path: state.db/WAL/fingerprint changes after the + # stale payload was cached. Followers must still be able to use that stale + # payload while the owner rebuild is blocked; otherwise sidebar polling can + # pile up behind a slow rebuild. + monkeypatch.setattr( + routes, + "_session_list_cache_source_stamp", + lambda _key: ("changed",), + ) started = threading.Event() release = threading.Event() @@ -217,13 +238,16 @@ def follower(): owner_thread = threading.Thread(target=owner) follower_thread = threading.Thread(target=follower) - owner_thread.start() - assert started.wait(1.0) - follower_thread.start() - follower_thread.join(1.0) - assert not follower_thread.is_alive() - release.set() - owner_thread.join(2.0) + try: + owner_thread.start() + assert started.wait(1.0) + follower_thread.start() + follower_thread.join(1.0) + assert not follower_thread.is_alive() + finally: + release.set() + owner_thread.join(2.0) + follower_thread.join(2.0) assert follower_result["payload"] == _session_cache_payload("stale") assert owner_result["payload"] == _session_cache_payload("fresh") diff --git a/tests/test_streaming_session_sidebar.py b/tests/test_streaming_session_sidebar.py index 6692628bbd8..481bd07de61 100644 --- a/tests/test_streaming_session_sidebar.py +++ b/tests/test_streaming_session_sidebar.py @@ -7,6 +7,8 @@ The sidebar filter must exempt actively-streaming sessions from the empty- Untitled rule so they remain visible while the user navigates away. """ +import json + import pytest import api.models as models @@ -165,3 +167,39 @@ def test_compact_output_contains_active_stream_id(_isolate): assert compact.get("active_stream_id") == "test-stream-123", ( "compact() must include active_stream_id for the sidebar filter." ) + +def test_messageful_session_missing_from_index_is_recovered_in_sidebar(_isolate): + """A saved messageful sidecar must remain discoverable even if _index misses it. + + During an active run, all_sessions() can see the in-memory SESSIONS overlay. + After the worker exits, only the sidecar JSON remains. If the incremental + index missed that row, the sidebar must recover it instead of dropping a + session that GET /api/session can still load by id. + """ + indexed = new_session() + indexed.messages.append({"role": "user", "content": "Indexed", "timestamp": 900.0}) + indexed.messages.append({"role": "assistant", "content": "Present", "timestamp": 901.0}) + indexed.title = "Already indexed" + indexed.save() + + s = new_session() + s.messages.append({"role": "user", "content": "Hello", "timestamp": 1000.0}) + s.messages.append({"role": "assistant", "content": "Hi", "timestamp": 1001.0}) + s.title = "Indexed recovery" + s.save() + + # Simulate the observed settled-worker shape: detail load can still read + # the sidecar file, but a non-empty sidebar index lacks this session and the + # in-memory runtime overlay is gone. + assert Session.load(s.session_id) is not None + models.SESSION_INDEX_FILE.write_text( + json.dumps([indexed.compact()]), + encoding="utf-8", + ) + SESSIONS.clear() + + rows = all_sessions() + by_id = {row["session_id"]: row for row in rows} + assert indexed.session_id in by_id + assert s.session_id in by_id + assert by_id[s.session_id]["message_count"] == 2 diff --git a/tests/test_webui_state_db_context_reconciliation.py b/tests/test_webui_state_db_context_reconciliation.py index 00958ac0b78..22b66f35d66 100644 --- a/tests/test_webui_state_db_context_reconciliation.py +++ b/tests/test_webui_state_db_context_reconciliation.py @@ -1,4 +1,3 @@ -import json import queue import sqlite3 from collections import OrderedDict @@ -131,6 +130,19 @@ def run_conversation(self, **kwargs): "external gateway assistant", ] + reloaded = models.Session.load(sid) + saved_contents = [m.get("content") for m in (reloaded.messages if reloaded else [])] + assert saved_contents == [ + "old user", + "old assistant", + "external gateway user", + "external gateway assistant", + "new webui turn", + "ok", + ] + assert saved_contents.count("old user") == 1 + assert saved_contents.count("external gateway user") == 1 + def test_state_db_delta_after_context_allows_recovered_turn_prefix(): from api.models import state_db_delta_after_context diff --git a/tests/test_webui_state_db_reconciliation.py b/tests/test_webui_state_db_reconciliation.py index ca3870c76ee..11daacee6ee 100644 --- a/tests/test_webui_state_db_reconciliation.py +++ b/tests/test_webui_state_db_reconciliation.py @@ -69,6 +69,31 @@ def _make_state_db(path: Path, sid: str, rows): conn.close() +def _append_state_db_rows(path: Path, sid: str, rows): + conn = sqlite3.connect(path) + try: + for row in rows: + conn.execute( + "INSERT INTO messages (session_id, role, content, timestamp, tool_call_id, tool_calls, tool_name) VALUES (?, ?, ?, ?, ?, ?, ?)", + ( + sid, + row["role"], + row["content"], + row.get("timestamp", 1000.0), + row.get("tool_call_id"), + row.get("tool_calls"), + row.get("tool_name"), + ), + ) + conn.execute( + "UPDATE sessions SET message_count = (SELECT COUNT(*) FROM messages WHERE session_id = ?) WHERE id = ?", + (sid, sid), + ) + conn.commit() + finally: + conn.close() + + def _install_test_session(monkeypatch, tmp_path, sid, sidecar_messages): import api.config as config import api.models as models @@ -173,6 +198,80 @@ def test_state_db_duplicate_backfills_turn_duration(): assert merged[0]["_turnDuration"] == 42.5 +def test_api_sessions_overlays_webui_state_db_summary_after_desktop_append(monkeypatch, tmp_path): + import api.routes as routes + + sid = "webui_desktop_sidebar_reconcile" + sidecar_messages = [ + {"role": "user", "content": "old user", "timestamp": 1000.0}, + {"role": "assistant", "content": "old assistant", "timestamp": 1001.0}, + ] + _install_test_session(monkeypatch, tmp_path, sid, sidecar_messages) + _make_state_db(tmp_path / "state.db", sid, list(sidecar_messages)) + monkeypatch.setattr(routes, "load_settings", lambda: {"show_cli_sessions": False}) + routes._clear_session_list_cache() + + first = _GetHandler("/api/sessions?sidebar_source=webui") + routes.handle_get(first, urlparse(first.path)) + assert first.status == 200 + first_row = next(row for row in first.response_json["sessions"] if row["session_id"] == sid) + assert first_row["message_count"] == 2 + + # Simulate the official Hermes Desktop App continuing the same WebUI-origin + # Hermes Agent session and settling its final rows into state.db. The second + # request happens immediately, so it only updates if the WebUI sidebar cache + # observes state.db changes even when the CLI/external-session tab is hidden. + _append_state_db_rows( + tmp_path / "state.db", + sid, + [ + {"role": "user", "content": "desktop user", "timestamp": 1002.0}, + {"role": "assistant", "content": "desktop assistant", "timestamp": 1003.0}, + ], + ) + + second = _GetHandler("/api/sessions?sidebar_source=webui") + routes.handle_get(second, urlparse(second.path)) + assert second.status == 200 + row = next(row for row in second.response_json["sessions"] if row["session_id"] == sid) + assert row["message_count"] == 4 + assert row["last_message_at"] == 1003.0 + assert row["updated_at"] == 1003.0 + + +def test_api_session_full_load_does_not_duplicate_state_db_prefix(monkeypatch, tmp_path): + import api.routes as routes + + sid = "webui_desktop_full_reconcile" + sidecar_messages = [ + {"role": "user", "content": "turn 1", "timestamp": 1000.0}, + {"role": "assistant", "content": "answer 1", "timestamp": 1001.0}, + {"role": "user", "content": "turn 2", "timestamp": 1002.0}, + ] + desktop_tail = [ + {"role": "assistant", "content": "answer 2 from desktop", "timestamp": 1003.0}, + {"role": "user", "content": "desktop follow-up", "timestamp": 1004.0}, + {"role": "assistant", "content": "desktop final", "timestamp": 1005.0}, + ] + _install_test_session(monkeypatch, tmp_path, sid, sidecar_messages) + _make_state_db(tmp_path / "state.db", sid, sidecar_messages + desktop_tail) + + handler = _GetHandler(f"/api/session?session_id={sid}&messages=1&resolve_model=0") + routes.handle_get(handler, urlparse(handler.path)) + + assert handler.status == 200 + messages = handler.response_json["session"]["messages"] + assert [m["content"] for m in messages] == [ + "turn 1", + "answer 1", + "turn 2", + "answer 2 from desktop", + "desktop follow-up", + "desktop final", + ] + assert handler.response_json["session"]["message_count"] == 6 + + def test_api_session_includes_state_db_messages_newer_than_webui_sidecar(monkeypatch, tmp_path): import api.routes as routes From fda5a241b22d61beee68ed24699cf0ff168f29f3 Mon Sep 17 00:00:00 2001 From: Frank Song Date: Thu, 25 Jun 2026 22:40:29 +0800 Subject: [PATCH 2/3] Fix sidebar reconciliation review issues --- api/models.py | 13 ++++++--- tests/test_session_sidebar_cache.py | 19 +++++++++++-- tests/test_webui_state_db_reconciliation.py | 30 +++++++++++++++++++++ 3 files changed, 56 insertions(+), 6 deletions(-) diff --git a/api/models.py b/api/models.py index 8015809b9bc..f4a70db0a15 100644 --- a/api/models.py +++ b/api/models.py @@ -3425,12 +3425,17 @@ def _apply_sidebar_state_db_override_metadata(sessions: list[dict], metadata: di state_last = float(state_db_last_message_at or 0) except (TypeError, ValueError): state_last = 0.0 + # ``current_last`` intentionally includes ``updated_at``: if a + # sidecar metadata-only write happened after the state.db append, + # keep the conservative anti-resurrection guard and wait for a + # newer settled state.db message before overlaying counts again. if state_count > current_count and (state_last <= 0 or state_last > current_last): + try: + existing_actual = max(0, int(session.get('actual_message_count') or 0)) + except (TypeError, ValueError): + existing_actual = 0 session['message_count'] = state_count - session['actual_message_count'] = max( - state_count, - int(session.get('actual_message_count') or 0) if str(session.get('actual_message_count') or '').isdigit() else 0, - ) + session['actual_message_count'] = max(state_count, existing_actual) if state_last > 0: session['last_message_at'] = max(float(session.get('last_message_at') or 0), state_last) session['updated_at'] = max(float(session.get('updated_at') or 0), state_last) diff --git a/tests/test_session_sidebar_cache.py b/tests/test_session_sidebar_cache.py index 11896363bdc..c6a328e8d04 100644 --- a/tests/test_session_sidebar_cache.py +++ b/tests/test_session_sidebar_cache.py @@ -1,10 +1,23 @@ import threading from types import SimpleNamespace +import pytest + import api.routes as routes from api import session_events +@pytest.fixture(autouse=True) +def _isolated_session_list_cache_state(): + routes._session_list_cache_clear() + with routes._SESSIONS_CACHE_LOCK: + routes._SESSIONS_CACHE_INFLIGHT.clear() + yield + routes._session_list_cache_clear() + with routes._SESSIONS_CACHE_LOCK: + routes._SESSIONS_CACHE_INFLIGHT.clear() + + class _StageRecorder: def __init__(self): self.stages = [] @@ -74,8 +87,9 @@ def builder_profile_b(): assert calls == ["default", "other", "default_all"] -def test_session_list_cache_singleflight_rebuild_once(): +def test_session_list_cache_singleflight_rebuild_once(monkeypatch): routes._session_list_cache_clear() + monkeypatch.setattr(routes, "_session_list_cache_source_stamp", lambda _key: ("stable",)) started = threading.Event() release = threading.Event() @@ -121,8 +135,9 @@ def reader(): assert calls == 1 -def test_session_list_cache_follower_wait_stage_when_rebuild_inflight(): +def test_session_list_cache_follower_wait_stage_when_rebuild_inflight(monkeypatch): routes._session_list_cache_clear() + monkeypatch.setattr(routes, "_session_list_cache_source_stamp", lambda _key: ("stable",)) started = threading.Event() release = threading.Event() diff --git a/tests/test_webui_state_db_reconciliation.py b/tests/test_webui_state_db_reconciliation.py index 11daacee6ee..8815e104f82 100644 --- a/tests/test_webui_state_db_reconciliation.py +++ b/tests/test_webui_state_db_reconciliation.py @@ -125,6 +125,36 @@ def _install_test_session(monkeypatch, tmp_path, sid, sidecar_messages): return session +def test_sidebar_state_db_overlay_preserves_numeric_actual_count(): + import api.models as models + + sid = "webui_float_actual_count" + sessions = [ + { + "session_id": sid, + "source_tag": "webui", + "message_count": 2, + "actual_message_count": 5.0, + "last_message_at": 1001.0, + "updated_at": 1001.0, + } + ] + + models._apply_sidebar_state_db_override_metadata( + sessions, + { + sid: { + "_state_db_source": "webui", + "_state_db_message_count": 4, + "_state_db_last_message_at": 1003.0, + } + }, + ) + + assert sessions[0]["message_count"] == 4 + assert sessions[0]["actual_message_count"] == 5 + + def test_tail_cancelled_partial_blocks_state_db_replay(): from api.models import merge_session_messages_append_only From f6fc7fc4d1c7e0b2aa08105beaa9f0fe38a431ec Mon Sep 17 00:00:00 2001 From: nesquena-hermes Date: Fri, 26 Jun 2026 00:47:00 +0000 Subject: [PATCH 3/3] Release XX (v0.51.668): WebUI<->Desktop session reconciliation (#4834, fixes #4833) --- CHANGELOG.md | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 951f15e38fd..425a34761a5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,12 @@ ## [Unreleased] +## [v0.51.668] — 2026-06-26 — Release XX (WebUI stays in sync after the Desktop app continues a session) + +### Fixed + +- **A WebUI-created session stays in sync after the official Hermes Desktop app continues the same Agent session.** When the Desktop app appended settled rows to the shared `state.db` and you returned to WebUI, the sidebar/detail could show a stale or incomplete transcript and the next WebUI turn's context missed the externally-appended messages. The sidebar now refreshes WebUI-origin rows from settled `state.db` count/timestamp even when external/CLI sessions are hidden (overlaying only when the DB strictly grew, metadata-only, with the active-stream hold-down preserved), and full loads avoid duplicating the sidecar prefix when merging `state.db`. Read-only reconciliation — no Desktop change, no live-stream mirroring. Thanks @franksong2702. (#4834, fixes #4833) + ## [v0.51.667] — 2026-06-26 — Release XW (compaction markers stay internal; code-shaped tool output stays intact) ### Fixed