Stage 386: PR #2581

# Conflicts:
#	api/streaming.py
This commit is contained in:
nesquena-hermes 2026-05-19 18:20:47 +00:00
commit 86f52f67b8
10 changed files with 922 additions and 73 deletions

View file

@ -2228,17 +2228,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:
@ -2269,47 +2267,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"""
@ -2342,6 +2341,174 @@ def get_cli_session_messages(sid) -> list:
return msgs
def get_state_db_session_summary(sid) -> dict:
"""Return cheap message count/max timestamp for one state.db session.
This is intentionally narrower than ``get_state_db_session_messages`` for
metadata-only WebUI polling: callers only need a staleness signal, not a
fully materialized transcript with tool/reasoning metadata.
"""
import os
try:
import sqlite3
except ImportError:
return {}
db_path = _active_state_db_path()
if not sid or not db_path.exists():
return {}
try:
with closing(sqlite3.connect(str(db_path))) as conn:
conn.row_factory = sqlite3.Row
cur = conn.cursor()
cur.execute("PRAGMA table_info(messages)")
available = {str(row['name']) for row in cur.fetchall()}
if not {'session_id', 'timestamp'}.issubset(available):
return {}
cur.execute(
"""
SELECT COUNT(*) AS message_count, MAX(timestamp) AS last_message_at
FROM messages
WHERE session_id = ?
""",
(str(sid),),
)
row = cur.fetchone()
if not row:
return {}
count = int(row['message_count'] or 0)
last_message_at = row['last_message_at']
result = {'message_count': count}
if last_message_at not in (None, ''):
try:
result['last_message_at'] = float(last_message_at)
except (TypeError, ValueError):
pass
return result
except Exception:
return {}
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)
key = _session_message_merge_key(msg)
if max_sidecar_timestamp is not None and timestamp is not None and timestamp <= max_sidecar_timestamp:
if key in seen_message_keys:
continue
if not (isinstance(key, tuple) and key[:1] == ("message_id",)):
continue
if key in seen_message_keys:
continue
# State rows at or before the newest sidecar timestamp are normally
# assumed to have already been observed by the sidecar. The <= gate
# preserves sidecar-only ordering/metadata for equal timestamps and
# prevents duplicate legacy rows when timestamp precision differs
# between stores. Explicit message ids are authoritative, though: two
# equal-timestamp messages with different ids are distinct retries.
if (
key[0] != "message_id"
and max_sidecar_timestamp is not None
and timestamp is not None
and timestamp <= max_sidecar_timestamp
):
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, state_messages: list | None = None
) -> 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 []
if state_messages is None:
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.

View file

@ -2220,6 +2220,9 @@ from api.models import (
import_cli_session,
get_cli_sessions,
get_cli_session_messages,
get_state_db_session_messages,
get_state_db_session_summary,
merge_session_messages_append_only,
ensure_cron_project,
is_cron_session,
)
@ -3665,8 +3668,17 @@ 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 = []
state_db_summary = {}
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) only need a
# cheap staleness signal. Avoid full transcript materialization
# on the steady-state polling path.
state_db_summary = get_state_db_session_summary(sid)
_t2 = _time.monotonic()
effective_model = (
_resolve_effective_session_model_for_display(s)
@ -3690,9 +3702,32 @@ 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(cli_messages, sidecar_messages)
else:
_all_msgs = merge_session_messages_append_only(getattr(s, "messages", []) or [], state_db_messages)
if not load_messages and state_db_summary:
sidecar_messages = getattr(s, "messages", []) or []
sidecar_count = len(sidecar_messages)
try:
sidecar_last = max(
float((m or {}).get("timestamp") or 0)
for m in sidecar_messages
if isinstance(m, dict)
) if sidecar_messages else 0
except (TypeError, ValueError):
sidecar_last = 0
state_count = int(state_db_summary.get("message_count") or 0)
state_last = float(state_db_summary.get("last_message_at") or 0)
_all_msgs = sidecar_messages
_summary_message_count = max(sidecar_count, state_count)
_summary_last_message_at = max(sidecar_last, state_last)
else:
_summary_message_count = None
_summary_last_message_at = None
if load_messages:
if msg_before is not None:
# Scroll-to-top paging: msg_before is a 0-based index into
@ -3708,7 +3743,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 +3783,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 = _summary_message_count if _summary_message_count is not None else len(_all_msgs)
_merged_last_message_at = _summary_last_message_at if _summary_last_message_at is not None else 0
if _summary_last_message_at is None and _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 +3816,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.

View file

@ -40,6 +40,7 @@ from api.metering import meter
from api.run_journal import RunJournalWriter
from api.turn_journal import append_turn_journal_event_for_stream
from api.usage import prompt_cache_hit_percent
from api.models import get_state_db_session_messages, 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
@ -2340,21 +2341,22 @@ def _has_task_resume_compaction_marker(messages):
return False
def _new_turn_context_from_messages(messages, msg_text):
"""Return provider-facing history for a new user turn from a message list."""
history = _drop_checkpointed_current_user_from_context(messages, msg_text)
if _is_casual_fresh_chat_message(msg_text) and _has_task_resume_compaction_marker(history):
return []
return history
def _context_messages_for_new_turn(session, msg_text):
"""Return provider-facing history for a new user turn.
Compacted agent sessions can carry a hidden "resume the active task" summary
long after the visible UI looks like normal chat. A short greeting should
not silently reactivate that old task; explicit continuation prompts still
keep the full compacted context.
in context_messages. If the user starts a fresh casual greeting in that old
session, do not feed that stale active-task summary back to the model.
"""
history = _drop_checkpointed_current_user_from_context(
_session_context_messages(session),
msg_text,
)
if _is_casual_fresh_chat_message(msg_text) and _has_task_resume_compaction_marker(history):
return []
return history
return _new_turn_context_from_messages(_session_context_messages(session), msg_text)
def _stream_writeback_is_current(session, stream_id):
@ -3971,8 +3973,21 @@ 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)
_external_state_messages = get_state_db_session_messages(getattr(s, 'session_id', None))
_previous_messages = list(
reconciled_state_db_messages_for_session(
s,
state_messages=_external_state_messages,
) or []
)
_previous_context_messages = _new_turn_context_from_messages(
reconciled_state_db_messages_for_session(
s,
prefer_context=True,
state_messages=_external_state_messages,
),
msg_text,
)
_pre_compression_count = getattr(
getattr(agent, 'context_compressor', None),
'compression_count', 0,

View file

@ -521,18 +521,22 @@ async function newSession(flash, options={}){
}
async function loadSession(sid){
const opts = arguments[1] || {};
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.
// Legacy invariant kept for static regression tests: if(currentSid===sid) return
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;
stopApprovalPolling();hideApprovalCard();
stopApprovalPolling();hideApprovalCard(forceReload);
_yoloEnabled=false;_updateYoloPill();
if(typeof stopClarifyPolling==='function') stopClarifyPolling();
if(typeof hideClarifyCard==='function') hideClarifyCard();
if(typeof hideClarifyCard==='function') hideClarifyCard(forceReload, forceReload?'external-refresh':'dismissed');
// Show loading indicator immediately for responsiveness.
// Cleared by renderMessages() once full session data arrives.
// Persist the current composer draft before switching away so it can be
@ -541,14 +545,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.
@ -2004,6 +2008,7 @@ function _applySessionListPayload(sessData, projData){
stopStreamingPoll();
}
ensureSessionTimeRefreshPoll();
ensureActiveSessionExternalRefreshPoll();
renderSessionListFromCache(); // no-ops if rename is in progress
}
@ -2037,8 +2042,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;
@ -2060,6 +2068,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);

View file

@ -1,10 +1,12 @@
from api.models import Session
from api.models import Session, reconciled_state_db_messages_for_session
import contextlib
from types import SimpleNamespace
from api.streaming import (
_assistant_reply_added_after_current_turn,
_context_messages_for_new_turn,
_merge_display_messages_after_agent_result,
_new_turn_context_from_messages,
_sanitize_messages_for_api,
_session_context_messages,
)
@ -314,6 +316,40 @@ def test_explicit_continue_keeps_compacted_active_task_context(tmp_path):
assert _context_messages_for_new_turn(session, "继续") == compacted_task_context
def test_streaming_reconciled_context_keeps_casual_greeting_suppression():
compacted_task_context = [
{
"role": "user",
"content": (
"[CONTEXT COMPACTION — REFERENCE ONLY] Earlier turns were compacted. "
"Your current task is identified in the Active Task section — resume exactly from there."
),
"timestamp": 1.0,
},
{"role": "assistant", "content": "I will inspect api/config.py next.", "timestamp": 2.0},
]
session = SimpleNamespace(
session_id="issue2308-streaming",
messages=[{"role": "user", "content": "old task", "timestamp": 0.5}],
context_messages=compacted_task_context,
)
external_state_messages = list(compacted_task_context)
# Mirror the streaming pre-turn assembly for prefer_context=True: reconcile
# sidecar context with one state.db snapshot, then apply the normal new-turn
# context filter that suppresses casual greetings from resuming stale tasks.
previous_context_messages = _new_turn_context_from_messages(
reconciled_state_db_messages_for_session(
session,
prefer_context=True,
state_messages=external_state_messages,
),
"你好",
)
assert previous_context_messages == []
def test_all_cjk_greetings_drop_stale_compaction_context(tmp_path):
"""Pin every CJK greeting in the casual-fresh-chat set against a stale
compaction context. Catches typos like \\u5616 (嘖, "click of tongue")

View file

@ -389,7 +389,7 @@ def test_focus_visibility_return_marks_active_session_viewed_and_clears_marker()
def test_completion_unread_clears_only_when_session_is_opened():
load_idx = SESSIONS_JS.find("async function loadSession(sid)")
load_idx = SESSIONS_JS.find("async function loadSession(sid")
assert load_idx != -1, "loadSession not found"
load_block = SESSIONS_JS[load_idx:SESSIONS_JS.find("function _resolveSessionModelForDisplaySoon", load_idx)]

View file

@ -28,7 +28,7 @@ def test_clicking_current_session_is_noop_before_load_session_side_effects():
load_session = _function_body(SESSIONS_JS, "async function loadSession")
current_idx = load_session.index("const currentSid = S.session ? S.session.session_id : null")
noop_idx = load_session.index("if(currentSid===sid) return")
noop_idx = load_session.index("if(currentSid===sid && !forceReload) return")
loading_idx = load_session.index("_loadingSessionId = sid")
stop_idx = load_session.index("stopApprovalPolling")

View file

@ -0,0 +1,39 @@
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)" in SESSIONS_JS
assert "const opts = arguments[1] || {};" 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
def test_force_reload_clears_stale_blocking_prompts_immediately():
"""External refresh should not leave old approval/clarify modals blocking the composer.
hideApprovalCard() and hideClarifyCard() defer hiding for their minimum-visible
timers unless force=true. That is correct for active streams, but when a
same-session external state.db update triggers loadSession(..., {force:true}),
the session has completed elsewhere and stale prompts should be removed now.
"""
assert "hideApprovalCard(forceReload)" in SESSIONS_JS
assert "hideClarifyCard(forceReload, forceReload?'external-refresh':'dismissed')" in SESSIONS_JS

View file

@ -0,0 +1,132 @@
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)
monkeypatch.setattr(models, "_active_state_db_path", lambda: tmp_path / "state.db", 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",
]

View file

@ -0,0 +1,352 @@
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.routes as routes
import api.profiles as profiles
monkeypatch.setattr(config, "STATE_DIR", tmp_path, raising=False)
session_dir = tmp_path / "sessions"
monkeypatch.setattr(config, "SESSION_DIR", session_dir, raising=False)
monkeypatch.setattr(config, "SESSION_INDEX_FILE", session_dir / "_index.json", raising=False)
monkeypatch.setattr(models, "SESSION_DIR", session_dir, raising=False)
monkeypatch.setattr(models, "SESSION_INDEX_FILE", session_dir / "_index.json", raising=False)
monkeypatch.setattr(models, "SESSIONS", OrderedDict(), raising=False)
monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path, raising=False)
monkeypatch.setattr(models, "_active_state_db_path", lambda: tmp_path / "state.db", raising=False)
monkeypatch.setattr(routes, "_active_state_db_path", lambda: tmp_path / "state.db", raising=False)
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"}}]