Release XX (v0.51.668): WebUI<->Desktop session reconciliation (#4834, fixes #4833)

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

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,45 @@ 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
+            # ``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, 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)
         title = session.get('title')
         if (
             state_db_title
@@ -3489,6 +3565,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(

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

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()
@@ -142,6 +157,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 +183,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 +215,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 +253,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")

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

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