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

@@ -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.
 

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.

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,

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 = '<div style="display:flex;align-items:center;justify-content:center;height:100%;color:var(--text-muted);font-size:14px;padding:40px;text-align:center;">Loading conversation...</div>';
+    if (_msgInner && currentSid !== sid) _msgInner.innerHTML = '<div style="display:flex;align-items:center;justify-content:center;height:100%;color:var(--text-muted);font-size:14px;padding:40px;text-align:center;">Loading conversation...</div>';
   }
   // 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);

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

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",
+    ]

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