Fix WebUI state.db session reconciliation
This commit is contained in:
parent
fce305993a
commit
0e2214b6bd
6 changed files with 296 additions and 19 deletions
101
api/models.py
101
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(
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -142,6 +142,15 @@ def test_session_list_cache_follower_wait_stage_when_rebuild_inflight():
|
|||
|
||||
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 test_session_list_cache_follower_wait_stage_when_rebuild_inflight():
|
|||
|
||||
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 test_session_list_cache_follower_reuses_stale_payload_during_slow_rebuild():
|
|||
|
||||
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")
|
||||
|
|
|
|||
|
|
@ -7,6 +7,8 @@ initial streaming turn, the session still looks like Untitled + 0-messages
|
|||
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
|
||||
|
|
|
|||
|
|
@ -1,4 +1,3 @@
|
|||
import json
|
||||
import queue
|
||||
import sqlite3
|
||||
from collections import OrderedDict
|
||||
|
|
@ -131,6 +130,19 @@ def test_next_webui_turn_context_includes_state_db_external_messages(monkeypatch
|
|||
"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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue