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"}}]