From 467ef33a24508ed10fcefa132b0eb3fb1fb4b47a Mon Sep 17 00:00:00 2001 From: Lumen Yang Date: Wed, 13 May 2026 13:21:44 +0000 Subject: [PATCH] feat(webui): reconcile external session updates When API server runs append messages directly to state.db, reconcile WebUI sidecar sessions with those canonical rows across API responses, model-facing streaming context, and active browser refresh. Add append-only state.db merge helpers, metadata-only counts for refresh polling, and regression coverage for API visibility, context incorporation, and frontend refresh behavior. --- api/models.py | 197 +++++++--- api/routes.py | 41 ++- api/streaming.py | 9 +- static/sessions.js | 61 ++- tests/test_webui_external_refresh_frontend.py | 26 ++ ...t_webui_state_db_context_reconciliation.py | 131 +++++++ tests/test_webui_state_db_reconciliation.py | 347 ++++++++++++++++++ 7 files changed, 753 insertions(+), 59 deletions(-) create mode 100644 tests/test_webui_external_refresh_frontend.py create mode 100644 tests/test_webui_state_db_context_reconciliation.py create mode 100644 tests/test_webui_state_db_reconciliation.py diff --git a/api/models.py b/api/models.py index 0518b227b..e978c68f0 100644 --- a/api/models.py +++ b/api/models.py @@ -2226,17 +2226,15 @@ def _json_loads_if_string(value): return value -def get_cli_session_messages(sid) -> list: - """Read messages for a single CLI/external-agent session. +def get_state_db_session_messages(sid, *, stitch_continuations: bool = False) -> list: + """Read messages for a Hermes session from the active profile's state.db. - Preserve tool-call/result and reasoning metadata from the agent state.db so - CLI-origin transcripts render with the same tool cards as WebUI-native - sessions. When the requested session is the tip of a compression/CLI-close - continuation chain, return the stitched full transcript across all segments - in chronological order. Returns empty list on any error. + This generic reader intentionally works for any session source, including + WebUI-origin sessions that were later updated through another Hermes surface + such as the Gateway API Server. When ``stitch_continuations`` is true it + preserves the historical CLI/external-agent behavior of walking compatible + compression/close parent segments before reading messages. """ - if str(sid or '').startswith(f'{CLAUDE_CODE_SOURCE}_'): - return get_claude_code_session_messages(sid) try: import sqlite3 except ImportError: @@ -2267,47 +2265,48 @@ def get_cli_session_messages(sid) -> list: ] selected = ['role', 'content', 'timestamp'] + [c for c in optional if c in available] - cur.execute("PRAGMA table_info(sessions)") - session_cols = {str(row['name']) for row in cur.fetchall()} session_chain = [str(sid)] - if {'parent_session_id', 'end_reason', 'started_at', 'source'}.issubset(session_cols): - cur.execute( - """ - SELECT id, source, started_at, parent_session_id, ended_at, end_reason - FROM sessions - WHERE id = ? - """, - (sid,), - ) - rows_by_id = {} - row = cur.fetchone() - if row: - rows_by_id[str(row['id'])] = dict(row) - current_id = str(row['id']) - seen = {current_id} - for _ in range(20): - current = rows_by_id.get(current_id) - parent_id = current.get('parent_session_id') if current else None - if not parent_id or parent_id in seen: - break - cur.execute( - """ - SELECT id, source, started_at, parent_session_id, ended_at, end_reason - FROM sessions - WHERE id = ? - """, - (parent_id,), - ) - parent_row = cur.fetchone() - if not parent_row: - break - parent_dict = dict(parent_row) - rows_by_id[str(parent_row['id'])] = parent_dict - if not _is_continuation_session(parent_dict, current): - break - session_chain.insert(0, str(parent_row['id'])) - current_id = str(parent_row['id']) - seen.add(current_id) + if stitch_continuations: + cur.execute("PRAGMA table_info(sessions)") + session_cols = {str(row['name']) for row in cur.fetchall()} + if {'parent_session_id', 'end_reason', 'started_at', 'source'}.issubset(session_cols): + cur.execute( + """ + SELECT id, source, started_at, parent_session_id, ended_at, end_reason + FROM sessions + WHERE id = ? + """, + (sid,), + ) + rows_by_id = {} + row = cur.fetchone() + if row: + rows_by_id[str(row['id'])] = dict(row) + current_id = str(row['id']) + seen = {current_id} + for _ in range(20): + current = rows_by_id.get(current_id) + parent_id = current.get('parent_session_id') if current else None + if not parent_id or parent_id in seen: + break + cur.execute( + """ + SELECT id, source, started_at, parent_session_id, ended_at, end_reason + FROM sessions + WHERE id = ? + """, + (parent_id,), + ) + parent_row = cur.fetchone() + if not parent_row: + break + parent_dict = dict(parent_row) + rows_by_id[str(parent_row['id'])] = parent_dict + if not _is_continuation_session(parent_dict, current): + break + session_chain.insert(0, str(parent_row['id'])) + current_id = str(parent_row['id']) + seen.add(current_id) placeholders = ', '.join('?' for _ in session_chain) cur.execute(f""" @@ -2340,6 +2339,106 @@ def get_cli_session_messages(sid) -> list: return msgs +def _normalized_message_timestamp_for_key(value): + if value is None or value == "": + return "" + try: + timestamp = float(value) + except (TypeError, ValueError): + return str(value) + if timestamp.is_integer(): + return str(int(timestamp)) + return ("%.6f" % timestamp).rstrip("0").rstrip(".") + + +def _message_timestamp_as_float(msg): + if not isinstance(msg, dict): + return None + value = msg.get("timestamp") + if value is None or value == "": + return None + try: + return float(value) + except (TypeError, ValueError): + return None + + +def _session_message_merge_key(msg: dict): + if not isinstance(msg, dict): + return ("non_dict", repr(msg)) + message_identity = msg.get("id") or msg.get("message_id") + if message_identity: + return ("message_id", str(message_identity)) + return ( + "legacy", + str(msg.get("role") or ""), + str(msg.get("content") or ""), + _normalized_message_timestamp_for_key(msg.get("timestamp")), + str(msg.get("tool_call_id") or ""), + str(msg.get("tool_name") or msg.get("name") or ""), + ) + + +def merge_session_messages_append_only(sidecar_messages: list, state_messages: list) -> list: + """Merge sidecar/context and state.db messages without deleting local rows.""" + sidecar_messages = list(sidecar_messages or []) + state_messages = list(state_messages or []) + if not state_messages: + return sidecar_messages + if not sidecar_messages: + return state_messages + + merged_messages = [] + seen_message_keys = set() + max_sidecar_timestamp = None + for msg in sidecar_messages: + timestamp = _message_timestamp_as_float(msg) + if timestamp is not None: + max_sidecar_timestamp = timestamp if max_sidecar_timestamp is None else max(max_sidecar_timestamp, timestamp) + key = _session_message_merge_key(msg) + seen_message_keys.add(key) + merged_messages.append(msg) + for msg in state_messages: + timestamp = _message_timestamp_as_float(msg) + if max_sidecar_timestamp is not None and timestamp is not None and timestamp <= max_sidecar_timestamp: + continue + key = _session_message_merge_key(msg) + if key in seen_message_keys: + continue + seen_message_keys.add(key) + merged_messages.append(msg) + return merged_messages + + +def reconciled_state_db_messages_for_session(session, *, prefer_context: bool = False) -> list: + """Return append-only messages reconciled with state.db for a WebUI session.""" + if session is None: + return [] + local_messages = [] + if prefer_context: + context_messages = getattr(session, 'context_messages', None) + if isinstance(context_messages, list) and context_messages: + local_messages = context_messages + if not local_messages: + local_messages = getattr(session, 'messages', None) or [] + state_messages = get_state_db_session_messages(getattr(session, 'session_id', None)) + return merge_session_messages_append_only(local_messages, state_messages) + + +def get_cli_session_messages(sid) -> list: + """Read messages for a single CLI/external-agent session. + + Preserve tool-call/result and reasoning metadata from the agent state.db so + CLI-origin transcripts render with the same tool cards as WebUI-native + sessions. When the requested session is the tip of a compression/CLI-close + continuation chain, return the stitched full transcript across all segments + in chronological order. Returns empty list on any error. + """ + if str(sid or '').startswith(f'{CLAUDE_CODE_SOURCE}_'): + return get_claude_code_session_messages(sid) + return get_state_db_session_messages(sid, stitch_continuations=True) + + def count_conversation_rounds(sid: str, since: float | None = None) -> int: """Count conversation rounds for a session from state.db. diff --git a/api/routes.py b/api/routes.py index fb2caeab0..c3330c3cb 100644 --- a/api/routes.py +++ b/api/routes.py @@ -2220,6 +2220,8 @@ from api.models import ( import_cli_session, get_cli_sessions, get_cli_session_messages, + get_state_db_session_messages, + merge_session_messages_append_only, ensure_cron_project, is_cron_session, ) @@ -3665,8 +3667,16 @@ def handle_get(handler, parsed) -> bool: cli_meta = _lookup_cli_session_metadata(sid) if _session_requires_cli_metadata_lookup(s) else {} is_messaging_session = _is_messaging_session_record(s) or _is_messaging_session_record(cli_meta) cli_messages = [] + state_db_messages = [] if is_messaging_session: cli_messages = get_cli_session_messages(sid) + elif load_messages: + state_db_messages = get_state_db_session_messages(sid) + elif not is_messaging_session: + # Metadata-only callers (frontend refresh polling) still need a + # reconciled count/timestamp so externally appended state.db + # messages can be detected without fetching the full transcript. + state_db_messages = get_state_db_session_messages(sid) _t2 = _time.monotonic() effective_model = ( _resolve_effective_session_model_for_display(s) @@ -3690,9 +3700,13 @@ def handle_get(handler, parsed) -> bool: # them chronologically and dedupe exact repeats. _all_msgs = _merged_session_messages_for_display(s, cli_messages) else: - _all_msgs = s.messages + _all_msgs = merge_session_messages_append_only(s.messages, state_db_messages) else: - _all_msgs = [] + if is_messaging_session and cli_messages: + sidecar_messages = getattr(s, "messages", []) or [] + _all_msgs = merge_session_messages_append_only(sidecar_messages, cli_messages) + else: + _all_msgs = merge_session_messages_append_only(getattr(s, "messages", []) or [], state_db_messages) if load_messages: if msg_before is not None: # Scroll-to-top paging: msg_before is a 0-based index into @@ -3708,7 +3722,7 @@ def handle_get(handler, parsed) -> bool: else: _truncated_msgs = _all_msgs else: - _truncated_msgs = _all_msgs + _truncated_msgs = [] # Resolve effective context_length with model-metadata fallback so # older sessions (pre-#1318) that have context_length=0 persisted # still render a meaningful indicator on load. Mirrors the @@ -3748,8 +3762,20 @@ def handle_get(handler, parsed) -> bool: # messages already carry per-message tool metadata. Avoid sending # the full historical list with a small tail window. _session_tool_calls = [] + _merged_message_count = len(_all_msgs) + _merged_last_message_at = 0 + if _all_msgs: + try: + _merged_last_message_at = max( + float((m or {}).get("timestamp") or 0) + for m in _all_msgs + if isinstance(m, dict) + ) + except (TypeError, ValueError): + _merged_last_message_at = 0 raw = s.compact() | { "messages": _truncated_msgs, + "message_count": _merged_message_count, "tool_calls": _session_tool_calls, "active_stream_id": getattr(s, "active_stream_id", None), "pending_user_message": getattr(s, "pending_user_message", None), @@ -3769,6 +3795,15 @@ def handle_get(handler, parsed) -> bool: journal, active=bool(getattr(s, "active_stream_id", None)), ) + if _merged_last_message_at: + raw["last_message_at"] = max( + float(raw.get("last_message_at") or 0), + _merged_last_message_at, + ) + raw["updated_at"] = max( + float(raw.get("updated_at") or 0), + _merged_last_message_at, + ) if cli_meta and _is_messaging_session_record(cli_meta): raw = _merge_cli_sidebar_metadata(raw, cli_meta) # Signal to the frontend that older messages were omitted. diff --git a/api/streaming.py b/api/streaming.py index 5998bdbc6..71fa24981 100644 --- a/api/streaming.py +++ b/api/streaming.py @@ -39,6 +39,7 @@ from api.compression_anchor import visible_messages_for_anchor from api.metering import meter from api.run_journal import RunJournalWriter from api.turn_journal import append_turn_journal_event_for_stream +from api.models import reconciled_state_db_messages_for_session # Global lock for os.environ writes. Per-session locks (_agent_lock) prevent # concurrent runs of the SAME session, but two DIFFERENT sessions can still @@ -3957,8 +3958,12 @@ def _run_agent_streaming( # or has been zeroed out (e.g. via a buggy migration / manual file edit). # Truthy-check covers None, missing-attr, and 0 uniformly. _turn_started_at = _pending_started_at if _pending_started_at else time.time() - _previous_messages = list(s.messages or []) - _previous_context_messages = _context_messages_for_new_turn(s, msg_text) + _reconciled_messages = list(reconciled_state_db_messages_for_session(s) or []) + _previous_messages = _reconciled_messages + _previous_context_messages = _drop_checkpointed_current_user_from_context( + reconciled_state_db_messages_for_session(s, prefer_context=True), + msg_text, + ) _pre_compression_count = getattr( getattr(agent, 'context_compressor', None), 'compression_count', 0, diff --git a/static/sessions.js b/static/sessions.js index 613a0c770..7d89aaca5 100644 --- a/static/sessions.js +++ b/static/sessions.js @@ -517,12 +517,15 @@ async function newSession(flash, options={}){ } } -async function loadSession(sid){ +async function loadSession(sid, opts){ + opts = opts || {}; + const forceReload = !!opts.force; const currentSid = S.session ? S.session.session_id : null; // Clicking the already-open session in the sidebar is a no-op. Reloading it // tears down active pane state and can reset the long-session scroll window - // to the top even though the user did not navigate anywhere. - if(currentSid===sid) return; + // to the top even though the user did not navigate anywhere. Explicit + // refresh paths pass {force:true} when external state.db changes arrive. + if(currentSid===sid && !forceReload) return; // Mark this session as the in-flight load. Subsequent loadSession() calls // will overwrite this; stale awaits use the mismatch to bail out (#1060). _loadingSessionId = sid; @@ -538,14 +541,14 @@ async function loadSession(sid){ if (currentSid && currentSid !== sid) { _saveComposerDraftNow(currentSid, ($('msg') || {}).value || '', S.pendingFiles ? [...S.pendingFiles] : []); } - if (currentSid !== sid) { + if (currentSid !== sid || forceReload) { S.messages = []; S.toolCalls = []; _messagesTruncated = false; _oldestIdx = 0; _loadingOlder = false; const _msgInner = $('msgInner'); - if (_msgInner) _msgInner.innerHTML = '
Loading conversation...
'; + if (_msgInner && currentSid !== sid) _msgInner.innerHTML = '
Loading conversation...
'; } // Phase 1: Load metadata only (~1KB) for fast session switching. // Guard against network/server failures to prevent a permanently stuck loading state. @@ -1995,6 +1998,7 @@ function _applySessionListPayload(sessData, projData){ stopStreamingPoll(); } ensureSessionTimeRefreshPoll(); + ensureActiveSessionExternalRefreshPoll(); renderSessionListFromCache(); // no-ops if rename is in progress } @@ -2028,8 +2032,11 @@ let _gatewaySSEWarningShown = false; const _gatewayFallbackPollMs = 30000; const _streamingPollMs = 5000; const _sessionTimeRefreshMs = 60000; +const _activeSessionExternalRefreshMs = 5000; let _streamingPollTimer = null; let _sessionTimeRefreshTimer = null; +let _activeSessionExternalRefreshTimer = null; +let _activeSessionExternalRefreshInFlight = false; function startStreamingPoll(){ if(_streamingPollTimer) return; @@ -2051,6 +2058,50 @@ function ensureSessionTimeRefreshPoll(){ }, _sessionTimeRefreshMs); } +async function refreshActiveSessionIfExternallyUpdated(reason){ + if(_activeSessionExternalRefreshInFlight) return; + if(!S.session || !S.session.session_id) return; + if(S.busy || S.activeStreamId) return; + if(typeof document !== 'undefined' && document.hidden) return; + const sid = S.session.session_id; + const localCount = Number(S.session.message_count || (Array.isArray(S.messages)?S.messages.length:0) || 0); + const localLast = Number(S.session.last_message_at || S.session.updated_at || 0); + _activeSessionExternalRefreshInFlight = true; + try{ + const data = await api(`/api/session?session_id=${encodeURIComponent(sid)}&messages=0&resolve_model=0`); + if(!data || !data.session) return; + if(!S.session || S.session.session_id !== sid) return; + if(S.busy || S.activeStreamId) return; + const remoteCount = Number(data.session.message_count || 0); + const remoteLast = Number(data.session.last_message_at || data.session.updated_at || 0); + if(remoteCount > localCount || remoteLast > localLast){ + await loadSession(sid, {force:true, externalRefreshReason:reason||'poll'}); + if(typeof renderSessionList==='function') void renderSessionList(); + } + }catch(e){ + // Ignore transient refresh failures; the next poll/focus event will retry. + }finally{ + _activeSessionExternalRefreshInFlight = false; + } +} + +function ensureActiveSessionExternalRefreshPoll(){ + if(_activeSessionExternalRefreshTimer) return; + _activeSessionExternalRefreshTimer = setInterval(() => { + void refreshActiveSessionIfExternallyUpdated('poll'); + }, _activeSessionExternalRefreshMs); + if(typeof document !== 'undefined' && !document._hermesExternalRefreshVisibilityHook){ + document.addEventListener('visibilitychange', () => { + if(!document.hidden) void refreshActiveSessionIfExternallyUpdated('visible'); + }); + document._hermesExternalRefreshVisibilityHook = true; + } + if(typeof window !== 'undefined' && !window._hermesExternalRefreshFocusHook){ + window.addEventListener('focus', () => { void refreshActiveSessionIfExternallyUpdated('focus'); }); + window._hermesExternalRefreshFocusHook = true; + } +} + function startGatewayPollFallback(ms){ const intervalMs = Math.max(5000, Number(ms) || _gatewayFallbackPollMs); if(_gatewayPollTimer) clearInterval(_gatewayPollTimer); diff --git a/tests/test_webui_external_refresh_frontend.py b/tests/test_webui_external_refresh_frontend.py new file mode 100644 index 000000000..2da0f5e40 --- /dev/null +++ b/tests/test_webui_external_refresh_frontend.py @@ -0,0 +1,26 @@ +from pathlib import Path + + +SESSIONS_JS = Path("static/sessions.js").read_text(encoding="utf-8") + + +def test_load_session_supports_force_reload_for_external_refresh(): + assert "async function loadSession(sid, opts)" in SESSIONS_JS + assert "const forceReload = !!opts.force" in SESSIONS_JS + assert "if(currentSid===sid && !forceReload) return;" in SESSIONS_JS + assert "loadSession(sid, {force:true" in SESSIONS_JS + + +def test_active_session_external_refresh_uses_metadata_then_force_reload(): + assert "function ensureActiveSessionExternalRefreshPoll()" in SESSIONS_JS + assert "async function refreshActiveSessionIfExternallyUpdated(reason)" in SESSIONS_JS + assert "messages=0&resolve_model=0" in SESSIONS_JS + assert "remoteCount > localCount || remoteLast > localLast" in SESSIONS_JS + assert "if(S.busy || S.activeStreamId) return;" in SESSIONS_JS + assert "document.hidden" in SESSIONS_JS + + +def test_active_session_external_refresh_has_focus_and_visibility_hooks(): + assert "visibilitychange" in SESSIONS_JS + assert "window.addEventListener('focus'" in SESSIONS_JS + assert "ensureActiveSessionExternalRefreshPoll();" in SESSIONS_JS diff --git a/tests/test_webui_state_db_context_reconciliation.py b/tests/test_webui_state_db_context_reconciliation.py new file mode 100644 index 000000000..30c0c4679 --- /dev/null +++ b/tests/test_webui_state_db_context_reconciliation.py @@ -0,0 +1,131 @@ +import json +import queue +import sqlite3 +from collections import OrderedDict +from pathlib import Path + +import pytest + +pytestmark = pytest.mark.requires_agent_modules + + +def _make_state_db(path: Path, sid: str, rows): + conn = sqlite3.connect(path) + conn.execute( + "CREATE TABLE sessions (id TEXT PRIMARY KEY, source TEXT, title TEXT, model TEXT, started_at REAL, message_count INTEGER)" + ) + conn.execute( + "CREATE TABLE messages (id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT, role TEXT, content TEXT, timestamp REAL)" + ) + conn.execute( + "INSERT INTO sessions (id, source, title, model, started_at, message_count) VALUES (?, ?, ?, ?, ?, ?)", + (sid, "webui", "Context Reconcile", "test-model", 1000.0, len(rows)), + ) + for row in rows: + conn.execute( + "INSERT INTO messages (session_id, role, content, timestamp) VALUES (?, ?, ?, ?)", + (sid, row["role"], row["content"], row.get("timestamp", 1000.0)), + ) + conn.commit() + conn.close() + + +def test_next_webui_turn_context_includes_state_db_external_messages(monkeypatch, tmp_path): + import api.config as config + import api.models as models + import api.profiles as profiles + import api.streaming as streaming + from api.models import Session + + session_dir = tmp_path / "sessions" + session_dir.mkdir() + index_file = session_dir / "_index.json" + monkeypatch.setattr(models, "SESSION_DIR", session_dir) + monkeypatch.setattr(models, "SESSION_INDEX_FILE", index_file) + monkeypatch.setattr(models, "SESSIONS", OrderedDict(), raising=False) + monkeypatch.setattr(config, "SESSION_DIR", session_dir, raising=False) + monkeypatch.setattr(config, "SESSION_INDEX_FILE", index_file, raising=False) + monkeypatch.setattr(streaming, "SESSION_DIR", session_dir, raising=False) + monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path, raising=False) + config.STREAMS.clear() + config.CANCEL_FLAGS.clear() + config.AGENT_INSTANCES.clear() + config.SESSION_AGENT_LOCKS.clear() + + sid = "webui_context_reconcile_001" + sidecar_messages = [ + {"role": "user", "content": "old user", "timestamp": 1000.0}, + {"role": "assistant", "content": "old assistant", "timestamp": 1001.0}, + ] + session = Session( + session_id=sid, + title="Context Reconcile", + workspace=str(tmp_path), + model="test-model", + messages=list(sidecar_messages), + context_messages=list(sidecar_messages), + ) + session.active_stream_id = "stream-context-reconcile" + session.pending_user_message = "new webui turn" + session.pending_started_at = 1004.0 + session.save(touch_updated_at=False) + models.SESSIONS[sid] = session + + _make_state_db( + tmp_path / "state.db", + sid, + [ + {"role": "user", "content": "old user", "timestamp": 1000.0}, + {"role": "assistant", "content": "old assistant", "timestamp": 1001.0}, + {"role": "user", "content": "external gateway user", "timestamp": 1002.0}, + {"role": "assistant", "content": "external gateway assistant", "timestamp": 1003.0}, + ], + ) + + captured = {} + + class FakeAgent: + def __init__(self, **kwargs): + self.session_id = sid + self.context_compressor = None + self.ephemeral_system_prompt = None + + def run_conversation(self, **kwargs): + captured["conversation_history"] = kwargs.get("conversation_history") + history = kwargs.get("conversation_history") or [] + return { + "completed": True, + "final_response": "ok", + "messages": history + [ + {"role": "user", "content": kwargs.get("persist_user_message", "")}, + {"role": "assistant", "content": "ok"}, + ], + } + + monkeypatch.setattr(streaming, "_get_ai_agent", lambda: FakeAgent) + monkeypatch.setattr(streaming, "resolve_model_provider", lambda *args, **kwargs: ("test-model", None, None)) + monkeypatch.setattr(streaming, "get_config", lambda: {}) + monkeypatch.setattr(config, "get_config", lambda: {}) + monkeypatch.setattr(config, "_resolve_cli_toolsets", lambda *args, **kwargs: []) + + stream_id = "stream-context-reconcile" + config.STREAMS[stream_id] = queue.Queue() + try: + streaming._run_agent_streaming( + session_id=sid, + msg_text="new webui turn", + model="test-model", + workspace=str(tmp_path), + stream_id=stream_id, + attachments=[], + ) + finally: + config.STREAMS.pop(stream_id, None) + + history_contents = [m.get("content") for m in captured.get("conversation_history") or []] + assert history_contents == [ + "old user", + "old assistant", + "external gateway user", + "external gateway assistant", + ] diff --git a/tests/test_webui_state_db_reconciliation.py b/tests/test_webui_state_db_reconciliation.py new file mode 100644 index 000000000..393a6b23b --- /dev/null +++ b/tests/test_webui_state_db_reconciliation.py @@ -0,0 +1,347 @@ +import json +import sqlite3 +from collections import OrderedDict +from io import BytesIO +from pathlib import Path +from urllib.parse import parse_qs, urlparse + +import pytest + +pytestmark = pytest.mark.requires_agent_modules + + +class _GetHandler: + def __init__(self, path): + self.path = path + self.headers = {} + self.client_address = ("127.0.0.1", 12345) + self.status = None + self.wfile = BytesIO() + self.response_headers = [] + + def send_response(self, status): + self.status = status + + def send_header(self, key, value): + self.response_headers.append((key, value)) + + def end_headers(self): + pass + + @property + def response_json(self): + return json.loads(self.wfile.getvalue().decode("utf-8")) + + @property + def query(self): + return parse_qs(urlparse(self.path).query) + + def log_message(self, *args, **kwargs): + pass + + +def _make_state_db(path: Path, sid: str, rows): + conn = sqlite3.connect(path) + conn.execute( + "CREATE TABLE sessions (id TEXT PRIMARY KEY, source TEXT, title TEXT, model TEXT, started_at REAL, message_count INTEGER)" + ) + conn.execute( + "CREATE TABLE messages (id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT, role TEXT, content TEXT, timestamp REAL, tool_call_id TEXT, tool_calls TEXT, tool_name TEXT)" + ) + conn.execute( + "INSERT INTO sessions (id, source, title, model, started_at, message_count) VALUES (?, ?, ?, ?, ?, ?)", + (sid, "webui", "Reconcile", "test-model", 1000.0, len(rows)), + ) + 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.commit() + conn.close() + + +def _install_test_session(monkeypatch, tmp_path, sid, sidecar_messages): + import api.config as config + import api.models as models + + import api.profiles as profiles + + monkeypatch.setattr(config, "STATE_DIR", tmp_path, raising=False) + monkeypatch.setattr(config, "SESSION_DIR", tmp_path / "sessions", raising=False) + monkeypatch.setattr(models, "SESSION_DIR", tmp_path / "sessions", raising=False) + monkeypatch.setattr(models, "SESSIONS", OrderedDict(), raising=False) + monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path, raising=False) + config.SESSION_DIR.mkdir(parents=True, exist_ok=True) + + session = models.Session( + session_id=sid, + title="Reconcile", + workspace=str(tmp_path), + model="test-model", + messages=sidecar_messages, + created_at=1000.0, + updated_at=1001.0, + ) + session.save(touch_updated_at=False) + return session + + +def test_api_session_includes_state_db_messages_newer_than_webui_sidecar(monkeypatch, tmp_path): + import api.routes as routes + + sid = "webui_reconcile_001" + 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, + [ + {"role": "user", "content": "old user", "timestamp": 1000.0}, + {"role": "assistant", "content": "old assistant", "timestamp": 1001.0}, + {"role": "user", "content": "external user", "timestamp": 1002.0}, + {"role": "assistant", "content": "external assistant", "timestamp": 1003.0}, + ], + ) + + handler = _GetHandler(f"/api/session?session_id={sid}&messages=1&resolve_model=0") + routes.handle_get(handler, urlparse(handler.path)) + + assert handler.status == 200 + payload = handler.response_json + messages = payload["session"]["messages"] + assert [m["content"] for m in messages] == [ + "old user", + "old assistant", + "external user", + "external assistant", + ] + assert payload["session"]["message_count"] == 4 + + +def test_state_db_reconciliation_preserves_sidecar_only_messages(monkeypatch, tmp_path): + import api.routes as routes + + sid = "webui_reconcile_sidecar_only" + _install_test_session( + monkeypatch, + tmp_path, + sid, + [ + {"role": "user", "content": "sidecar-only draft", "timestamp": 999.0}, + {"role": "user", "content": "old user", "timestamp": 1000.0}, + ], + ) + _make_state_db( + tmp_path / "state.db", + sid, + [ + {"role": "user", "content": "old user", "timestamp": 1000.0}, + {"role": "assistant", "content": "external assistant", "timestamp": 1001.0}, + ], + ) + + 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] == [ + "sidecar-only draft", + "old user", + "external assistant", + ] + + +def test_state_db_reconciliation_does_not_collapse_repeated_content_with_different_timestamps(monkeypatch, tmp_path): + import api.routes as routes + + sid = "webui_reconcile_repeated" + _install_test_session( + monkeypatch, + tmp_path, + sid, + [{"role": "assistant", "content": "same", "timestamp": 1000.0}], + ) + _make_state_db( + tmp_path / "state.db", + sid, + [ + {"role": "assistant", "content": "same", "timestamp": 1000.0}, + {"role": "assistant", "content": "same", "timestamp": 1001.0}, + ], + ) + + 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] == ["same", "same"] + assert [m["timestamp"] for m in messages] == [1000.0, 1001.0] + + +def test_state_db_reconciliation_preserves_sidecar_order_when_timestamps_collide(monkeypatch, tmp_path): + import api.routes as routes + + sid = "webui_reconcile_same_timestamp_order" + _install_test_session( + monkeypatch, + tmp_path, + sid, + [ + {"role": "user", "content": "z user happened first", "timestamp": 1000}, + {"role": "assistant", "content": "a assistant happened second", "timestamp": 1000}, + {"role": "tool", "content": "m tool happened third", "timestamp": 1000, "tool_call_id": "call_1"}, + ], + ) + _make_state_db( + tmp_path / "state.db", + sid, + [ + {"role": "user", "content": "z user happened first", "timestamp": 1000.0}, + {"role": "assistant", "content": "a assistant happened second", "timestamp": 1000.0}, + {"role": "tool", "content": "m tool happened third", "timestamp": 1000.0, "tool_call_id": "call_1"}, + ], + ) + + 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] == [ + "z user happened first", + "a assistant happened second", + "m tool happened third", + ] + assert handler.response_json["session"]["message_count"] == 3 + + +def test_state_db_reconciliation_dedupes_numeric_equivalent_timestamps(monkeypatch, tmp_path): + import api.routes as routes + + sid = "webui_reconcile_numeric_timestamp" + _install_test_session( + monkeypatch, + tmp_path, + sid, + [{"role": "assistant", "content": "same timestamp", "timestamp": 1000}], + ) + _make_state_db( + tmp_path / "state.db", + sid, + [{"role": "assistant", "content": "same timestamp", "timestamp": 1000.0}], + ) + + 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] == ["same timestamp"] + assert handler.response_json["session"]["message_count"] == 1 + + +def test_state_db_reconciliation_preserves_repeated_sidecar_rows(monkeypatch, tmp_path): + import api.routes as routes + + sid = "webui_reconcile_repeated_sidecar" + _install_test_session( + monkeypatch, + tmp_path, + sid, + [ + {"role": "assistant", "content": "", "timestamp": 1000}, + {"role": "assistant", "content": "", "timestamp": 1000}, + {"role": "assistant", "content": "done", "timestamp": 1001}, + ], + ) + _make_state_db( + tmp_path / "state.db", + sid, + [{"role": "assistant", "content": "", "timestamp": 1000.0}], + ) + + 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] == ["", "", "done"] + assert handler.response_json["session"]["message_count"] == 3 + + +def test_metadata_fast_path_reports_reconciled_state_db_count(monkeypatch, tmp_path): + import api.routes as routes + + sid = "webui_reconcile_metadata" + _install_test_session( + monkeypatch, + tmp_path, + sid, + [ + {"role": "user", "content": "old user", "timestamp": 1000.0}, + {"role": "assistant", "content": "old assistant", "timestamp": 1001.0}, + ], + ) + _make_state_db( + tmp_path / "state.db", + sid, + [ + {"role": "user", "content": "old user", "timestamp": 1000.0}, + {"role": "assistant", "content": "old assistant", "timestamp": 1001.0}, + {"role": "user", "content": "external metadata user", "timestamp": 1002.0}, + {"role": "assistant", "content": "external metadata assistant", "timestamp": 1003.0}, + ], + ) + + handler = _GetHandler(f"/api/session?session_id={sid}&messages=0&resolve_model=0") + routes.handle_get(handler, urlparse(handler.path)) + + assert handler.status == 200 + session = handler.response_json["session"] + assert session["messages"] == [] + assert session["message_count"] == 4 + assert session["last_message_at"] == 1003.0 + + +def test_state_db_reconciliation_preserves_tool_metadata(monkeypatch, tmp_path): + import api.routes as routes + + sid = "webui_reconcile_tool_metadata" + _install_test_session( + monkeypatch, + tmp_path, + sid, + [{"role": "user", "content": "old user", "timestamp": 1000.0}], + ) + tool_calls = json.dumps([{"id": "call_1", "function": {"name": "terminal"}}]) + _make_state_db( + tmp_path / "state.db", + sid, + [ + {"role": "user", "content": "old user", "timestamp": 1000.0}, + { + "role": "assistant", + "content": "used a tool", + "timestamp": 1001.0, + "tool_calls": tool_calls, + "tool_name": "terminal", + }, + ], + ) + + 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 messages[-1]["content"] == "used a tool" + assert messages[-1]["tool_name"] == "terminal" + assert messages[-1]["tool_calls"] == [{"id": "call_1", "function": {"name": "terminal"}}]