From c27f3d67a80cfc0b9a6fea73f6ab8ec7d91ec281 Mon Sep 17 00:00:00 2001 From: ruizanthony Date: Fri, 21 Aug 2026 07:01:52 +0200 Subject: [PATCH] fix(compression): fence durable transcript revisions (#6554) Co-authored-by: Anthony Ruiz Co-authored-by: nesquena-hermes --- api/models.py | 211 ++- api/streaming.py | 855 ++++++++-- tests/test_compression_snapshot_revision.py | 1496 +++++++++++++++++ ..._issue5121_provider_auth_terminal_error.py | 83 +- tests/test_state_db_read_backstop.py | 46 + 5 files changed, 2554 insertions(+), 137 deletions(-) create mode 100644 tests/test_compression_snapshot_revision.py diff --git a/api/models.py b/api/models.py index 3ceca910..00246533 100644 --- a/api/models.py +++ b/api/models.py @@ -13,7 +13,9 @@ import threading import time import uuid from contextlib import closing, contextmanager +from dataclasses import dataclass from pathlib import Path +from typing import Literal, overload try: # pragma: no cover - platform-specific imports. import fcntl as _fcntl @@ -8107,6 +8109,43 @@ def _json_loads_if_string(value): return value +@dataclass(frozen=True) +class StateDBSessionMessagesSnapshot: + """Internal message projection paired with its durable SQLite revision.""" + + messages: list + revision: dict | None + + +def _state_db_session_messages_result(messages, revision, *, with_revision): + if with_revision: + return StateDBSessionMessagesSnapshot(messages=list(messages or []), revision=revision) + return list(messages or []) + + +def _state_db_active_rows_digest(rows) -> str: + """Match the Agent fence for model-facing fields mutable in place.""" + digest = hashlib.sha256() + stamped = False + for row in rows: + keys = row.keys() + api_content = row['api_content'] if 'api_content' in keys else None + if api_content is None: + continue + stamped = True + digest.update( + json.dumps( + [row['id'], api_content], + ensure_ascii=False, + separators=(',', ':'), + default=str, + ).encode('utf-8', errors='surrogatepass') + ) + digest.update(b'\n') + return digest.hexdigest() if stamped else '' + + +@overload def get_state_db_session_messages( sid, *, @@ -8115,7 +8154,33 @@ def get_state_db_session_messages( since_timestamp=None, include_inactive: bool = False, limit=None, -) -> list: + with_revision: Literal[False] = False, +) -> list: ... + + +@overload +def get_state_db_session_messages( + sid, + *, + stitch_continuations: bool = False, + profile=None, + since_timestamp=None, + include_inactive: bool = False, + limit=None, + with_revision: Literal[True], +) -> StateDBSessionMessagesSnapshot: ... + + +def get_state_db_session_messages( + sid, + *, + stitch_continuations: bool = False, + profile=None, + since_timestamp=None, + include_inactive: bool = False, + limit=None, + with_revision: bool = False, +): """Read messages for a Hermes session from state.db. When *profile* is supplied, reads from that profile's state.db; otherwise @@ -8145,20 +8210,23 @@ def get_state_db_session_messages( ``active=0`` archive rows back in resurrects pre-compaction history and can make every later turn re-trigger compression. Pass ``include_inactive=True`` only for explicit recovery/audit views. + + ``with_revision=True`` returns a :class:`StateDBSessionMessagesSnapshot`. + Its revision is derived from the exact rows fetched by the same SQLite + query and is available only for an unbounded, active, current-segment read. + Existing callers keep the historical list return by default. """ try: import sqlite3 except ImportError: - return [] + return _state_db_session_messages_result([], None, with_revision=with_revision) if isinstance(profile, str) and profile: db_path = _get_profile_home(profile) / 'state.db' - if not db_path.exists(): - db_path = _active_state_db_path() else: db_path = _active_state_db_path() if not db_path.exists(): - return [] + return _state_db_session_messages_result([], None, with_revision=with_revision) try: with closing(open_state_db_readonly(db_path)) as conn: @@ -8168,7 +8236,7 @@ def get_state_db_session_messages( available = {str(row['name']) for row in cur.fetchall()} required = {'role', 'content', 'timestamp'} if not required.issubset(available): - return [] + return _state_db_session_messages_result([], None, with_revision=with_revision) optional = [ 'tool_call_id', 'tool_calls', @@ -8185,7 +8253,18 @@ def get_state_db_session_messages( 'api_content', ] id_col = ['id'] if 'id' in available else [] - selected = id_col + ['role', 'content', 'timestamp'] + [c for c in optional if c in available] + revision_cols = [] + if with_revision: + if 'active' in available: + revision_cols.append('active') + if 'api_content' in available: + revision_cols.append('api_content') + selected = ( + id_col + + ['role', 'content', 'timestamp'] + + revision_cols + + [c for c in optional if c in available] + ) session_chain = [str(sid)] if stitch_continuations: @@ -8244,6 +8323,7 @@ def get_state_db_session_messages( active_clause = "" if 'active' in available and not include_inactive: active_clause = " AND (active IS NULL OR active != 0)" + durable_order_column = 'id' if 'id' in available else 'timestamp' # Defensive row cap (backstop only — see docstring). Applied as a # SQL LIMIT bound parameter (?) so the tail (newest) rows are # retained and a pathological state.db can't materialize unbounded @@ -8255,10 +8335,9 @@ def get_state_db_session_messages( except (TypeError, ValueError): limit_int = None if limit_int is not None: - # The query orders ASC (oldest first); to keep the NEWEST - # rows under the cap, take a descending-ordered subquery and - # re-sort ascending — a plain LIMIT would keep the oldest. - limit_clause = " ORDER BY timestamp DESC, id DESC LIMIT ?" + # Keep the newest durable rows under the cap, then restore + # canonical append order for the caller. + limit_clause = f" ORDER BY {durable_order_column} DESC LIMIT ?" params.append(limit_int) if limit_clause: cur.execute(f""" @@ -8269,7 +8348,7 @@ def get_state_db_session_messages( {since_clause} {active_clause} {limit_clause} - ) ORDER BY timestamp ASC, id ASC + ) ORDER BY {durable_order_column} ASC """, params) else: cur.execute(f""" @@ -8278,10 +8357,34 @@ def get_state_db_session_messages( WHERE session_id IN ({placeholders}) {since_clause} {active_clause} - ORDER BY timestamp ASC, id ASC + ORDER BY {durable_order_column} ASC """, params) + rows = cur.fetchall() + revision = None + if ( + with_revision + and 'id' in available + and 'active' in available + and len(session_chain) == 1 + and str(session_chain[0]) == str(sid) + and since_timestamp is None + and not include_inactive + and limit is None + and all(row['active'] == 1 for row in rows) + ): + revision = { + 'session_id': str(sid), + 'active_message_count': len(rows), + 'max_active_message_id': max( + (int(row['id']) for row in rows), + default=0, + ), + } + if 'api_content' in available: + revision['active_rows_digest'] = _state_db_active_rows_digest(rows) + msgs = [] - for row in cur.fetchall(): + for row in rows: msg = { 'role': row['role'], 'content': row['content'], @@ -8315,8 +8418,8 @@ def get_state_db_session_messages( msg['name'] = msg['tool_name'] msgs.append(msg) except Exception: - return [] - return msgs + return _state_db_session_messages_result([], None, with_revision=with_revision) + return _state_db_session_messages_result(msgs, revision, with_revision=with_revision) def get_state_db_session_message_prefix_summary( @@ -10210,12 +10313,40 @@ def merge_session_messages_append_only( return merged_messages +@overload def reconciled_state_db_messages_for_session( - session, *, prefer_context: bool = False, state_messages: list | None = None -) -> list: + session, + *, + prefer_context: bool = False, + state_messages: list | StateDBSessionMessagesSnapshot | None = None, + with_revision: Literal[False] = False, +) -> list: ... + + +@overload +def reconciled_state_db_messages_for_session( + session, + *, + prefer_context: bool = False, + state_messages: list | StateDBSessionMessagesSnapshot | None = None, + with_revision: Literal[True], +) -> StateDBSessionMessagesSnapshot: ... + + +def reconciled_state_db_messages_for_session( + session, + *, + prefer_context: bool = False, + state_messages: list | StateDBSessionMessagesSnapshot | None = None, + with_revision: bool = False, +): """Return append-only messages reconciled with state.db for a WebUI session.""" if session is None: - return [] + return _state_db_session_messages_result([], None, with_revision=with_revision) + state_revision = None + if isinstance(state_messages, StateDBSessionMessagesSnapshot): + state_revision = state_messages.revision + state_messages = state_messages.messages local_messages = [] using_context_messages = False if prefer_context: @@ -10226,7 +10357,28 @@ def reconciled_state_db_messages_for_session( 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)) + session_id = getattr(session, 'session_id', None) + session_profile = getattr(session, 'profile', None) + if with_revision: + state_result = get_state_db_session_messages( + session_id, + profile=session_profile, + with_revision=True, + ) + elif session_profile: + state_result = get_state_db_session_messages( + session_id, + profile=session_profile, + ) + else: + # Preserve the historical one-argument call contract for ordinary + # reconciliation and for integrations that replace this reader. + state_result = get_state_db_session_messages(session_id) + if isinstance(state_result, StateDBSessionMessagesSnapshot): + state_revision = state_result.revision + state_messages = state_result.messages + else: + state_messages = state_result if prefer_context and local_messages: if using_context_messages: sidecar_messages = getattr(session, 'messages', None) or [] @@ -10253,22 +10405,35 @@ def reconciled_state_db_messages_for_session( "Compressed context for session %s has no compression anchor; using context_messages only", getattr(session, "session_id", None), ) - return list(local_messages) + return _state_db_session_messages_result( + local_messages, + state_revision, + with_revision=with_revision, + ) anchor_index = _state_db_anchor_index(state_messages, anchor_key) if anchor_index is None: logger.debug( "Compressed context for session %s has an unverifiable compression anchor; using context_messages only", getattr(session, "session_id", None), ) - return list(local_messages) + return _state_db_session_messages_result( + local_messages, + state_revision, + with_revision=with_revision, + ) state_messages = list(state_messages or [])[anchor_index + 1 :] state_messages = state_db_delta_after_context(local_messages, state_messages) - return merge_session_messages_append_only( + reconciled_messages = merge_session_messages_append_only( local_messages, state_messages, truncation_watermark=getattr(session, "truncation_watermark", None), truncation_boundary=getattr(session, "truncation_boundary", None), ) + return _state_db_session_messages_result( + reconciled_messages, + state_revision, + with_revision=with_revision, + ) def get_cli_session_messages(sid, *, profile=None) -> list: diff --git a/api/streaming.py b/api/streaming.py index ea7875ec..4431d4db 100644 --- a/api/streaming.py +++ b/api/streaming.py @@ -65,6 +65,7 @@ from api.todo_state import attach_todo_state, emit_todo_state from api.turn_journal import append_turn_journal_event_for_stream from api.usage import prompt_cache_hit_percent from api.models import ( + StateDBSessionMessagesSnapshot, _is_empty_partial_activity_message, _evict_sessions_over_cap, clear_process_wakeup_pause, @@ -1393,7 +1394,24 @@ def _provider_error_probe_text(value) -> tuple[str, int | None]: return ' '.join(t for t in _texts if t).strip(), _status_code -def _classify_provider_error(err_str: str, exc=None, *, silent_failure: bool = False) -> dict: +def _result_reports_compression_snapshot_stale(result) -> bool: + """Return whether an Agent result carries an exact stale-snapshot marker.""" + return ( + isinstance(result, dict) + and ( + result.get('error') == 'compression_snapshot_stale' + or result.get('compression_snapshot_stale') is True + ) + ) + + +def _classify_provider_error( + err_str: str, + exc=None, + *, + silent_failure: bool = False, + result=None, +) -> dict: """Classify provider/agent failure text for WebUI apperror UX. Keep this string-based until hermes-agent exposes stable structured @@ -1409,6 +1427,33 @@ def _classify_provider_error(err_str: str, exc=None, *, silent_failure: bool = F err_str = str(_probe_text or err_str or '') _err_lower = err_str.lower() _exc_name = type(exc).__name__ if exc is not None else '' + _result_is_compression_snapshot_stale = ( + _result_reports_compression_snapshot_stale(result) + ) + try: + from agent.conversation_compression import CompressionSnapshotStaleError # type: ignore[attr-defined] + except ImportError: + # Paired deployments export the typed exception. The name-only fallback + # keeps a rolling WebUI/Agent upgrade actionable without text matching. + _is_compression_snapshot_stale = ( + _result_is_compression_snapshot_stale + or _exc_name == 'CompressionSnapshotStaleError' + ) + else: + _is_compression_snapshot_stale = ( + _result_is_compression_snapshot_stale + or isinstance(exc, CompressionSnapshotStaleError) + ) + if _is_compression_snapshot_stale: + return { + 'label': 'Conversation changed during compression', + 'type': 'compression_snapshot_stale', + 'hint': ( + 'The durable conversation changed while compression was being prepared. ' + 'No automatic retry was attempted. Send your next message to reload and ' + 'reconcile the latest conversation state.' + ), + } _is_cancelled = ( 'cancelled by user' in _err_lower or 'canceled by user' in _err_lower @@ -1674,24 +1719,38 @@ def _resolve_active_turn_authority(identity, *, result=None, agent=None): if not isinstance(identity, dict): return identity resolved = dict(identity) + resolved.pop('agent_turn_boundary_resolved', None) + resolved.pop('agent_turn_boundary_source', None) + + # Treat the boundary as one coherent pair. In particular, do not retain the + # failed Agent instance's index/turn when credential self-heal creates a + # fresh Agent whose compacted projection has a different current-user index. + # The returned-result pair is authoritative when exported; current paired + # Agents keep the same pair on the instance instead. + _boundary_idx = None + _boundary_turn_id = '' + _boundary_source = '' if isinstance(result, dict): _result_idx = _coerce_current_turn_user_idx(result.get('current_turn_user_idx')) - if _result_idx is not None: - resolved['current_turn_user_idx'] = _result_idx _result_turn_id = str(result.get('turn_id') or '').strip() - if _result_turn_id: - resolved['turn_id'] = _result_turn_id + if _result_idx is not None and _result_turn_id: + _boundary_idx = _result_idx + _boundary_turn_id = _result_turn_id + _boundary_source = 'result' if agent is not None: - if resolved.get('current_turn_user_idx') is None: - _agent_idx = _coerce_current_turn_user_idx( - getattr(agent, '_persist_user_message_idx', None) - ) - if _agent_idx is not None: - resolved['current_turn_user_idx'] = _agent_idx - if not resolved.get('turn_id'): - _agent_turn_id = str(getattr(agent, '_current_turn_id', '') or '').strip() - if _agent_turn_id: - resolved['turn_id'] = _agent_turn_id + _agent_idx = _coerce_current_turn_user_idx( + getattr(agent, '_persist_user_message_idx', None) + ) + _agent_turn_id = str(getattr(agent, '_current_turn_id', '') or '').strip() + if _agent_idx is not None and _agent_turn_id and not _boundary_source: + _boundary_idx = _agent_idx + _boundary_turn_id = _agent_turn_id + _boundary_source = 'agent' + if _boundary_source: + resolved['current_turn_user_idx'] = _boundary_idx + resolved['turn_id'] = _boundary_turn_id + resolved['agent_turn_boundary_resolved'] = True + resolved['agent_turn_boundary_source'] = _boundary_source return resolved @@ -1700,7 +1759,10 @@ def _active_turn_boundary_is_valid(identity): return False if not str(identity.get('turn_id') or '').strip(): return False - return isinstance(identity.get('current_turn_user_idx'), int) + return ( + identity.get('agent_turn_boundary_resolved') is True + and isinstance(identity.get('current_turn_user_idx'), int) + ) def _active_turn_token_matches(message, identity): @@ -1765,6 +1827,22 @@ def _owner_projection_current_turn_row(messages, identity): def _find_active_turn_checkpoint_index(result_messages, previous_context, identity, msg_text): + """Locate the current turn's user row inside ``result_messages``. + + Exactly one declared index domain is supported. The WebUI token is the + strongest proof and wins when it survives the Agent projection. Otherwise + the Agent-resolved ``current_turn_user_idx`` addresses ``result["messages"]`` + directly, so only that exact index is validated. With a repeated prompt a + shifted probe (``current_turn_user_idx - len(previous_context)``) can land + on an identical historical user row and leak historical assistant prose + into the current turn, so no alternative projection is ever probed. If a + shifted projection is ever genuinely required, the Agent must carry an + explicit projection-origin/base-length discriminator instead. + + ``previous_context`` is retained for call-site compatibility only; it does + not participate in index resolution. + """ + del previous_context # single declared index domain: result_messages only result_messages = list(result_messages or []) if not isinstance(identity, dict): return None @@ -1775,28 +1853,17 @@ def _find_active_turn_checkpoint_index(result_messages, previous_context, identi if not _active_turn_boundary_is_valid(identity): return None expected_text = identity.get('text') if identity.get('text') is not None else msg_text - candidates = [] - current_turn_user_idx = identity['current_turn_user_idx'] - previous_context = list(previous_context or []) - if _messages_have_prefix(result_messages, previous_context): - candidates.append(current_turn_user_idx) - else: - offset = current_turn_user_idx - len(previous_context) - candidates.extend([offset, current_turn_user_idx]) - seen = set() - for idx in candidates: - if idx in seen or idx is None: - continue - seen.add(idx) - if idx < 0 or idx >= len(result_messages): - continue - message = result_messages[idx] - if ( - isinstance(message, dict) - and message.get('role') == 'user' - and _normalize_user_text(_message_text(message.get('content'))) == _normalize_user_text(expected_text) - ): - return idx + idx = identity['current_turn_user_idx'] + if idx < 0 or idx >= len(result_messages): + return None + message = result_messages[idx] + if ( + isinstance(message, dict) + and message.get('role') == 'user' + and _normalize_user_text(_message_text(message.get('content'))) + == _normalize_user_text(expected_text) + ): + return idx return None @@ -7217,6 +7284,74 @@ def _agent_result_terminal_failure(result) -> bool: return False +def _self_heal_result_succeeded( + result, + previous_context, + active_turn_identity, + msg_text, +) -> bool: + """Accept a credential retry only after authoritative turn success.""" + if not isinstance(result, dict): + return False + if _result_reports_compression_snapshot_stale(result): + return False + if _agent_result_terminal_failure(result): + return False + result_error = result.get('error') + if isinstance(result_error, str): + if result_error.strip(): + return False + elif result_error: + return False + if result.get('completed') is False: + return False + messages = result.get('messages') or [] + previous_context = list(previous_context or []) + if _messages_have_prefix(messages, previous_context): + current_turn_rows = list(messages[len(previous_context):]) + current_user_idx = next( + ( + index + for index, row in enumerate(current_turn_rows) + if isinstance(row, dict) + and row.get('role') == 'user' + and _normalize_user_text(_message_text(row.get('content'))) + == _normalize_user_text(msg_text) + ), + None, + ) + if current_user_idx is not None: + current_turn_rows = current_turn_rows[current_user_idx + 1:] + else: + current_user_idx = _find_active_turn_checkpoint_index( + messages, + previous_context, + active_turn_identity, + msg_text, + ) + if current_user_idx is None: + current_turn_rows = [] + else: + current_turn_rows = list(messages[current_user_idx + 1:]) + next_user_idx = next( + ( + index + for index, row in enumerate(current_turn_rows) + if isinstance(row, dict) and row.get('role') == 'user' + ), + None, + ) + if next_user_idx is not None: + current_turn_rows = current_turn_rows[:next_user_idx] + return any( + isinstance(row, dict) + and row.get('role') == 'assistant' + and not row.get('_error') + and _assistant_message_has_final_visible_text(row) + for row in current_turn_rows + ) + + _TOOL_RESULT_SNIPPET_MAX = 4000 # Tool-arg keys whose values are card content / diff-reconstruction inputs. @@ -7478,6 +7613,125 @@ def _partial_marker_already_present(messages, candidate: dict, *, before_idx: in return False +def _partial_snapshot_prefers_candidate_text(current, candidate) -> bool: + """Choose the later-looking monotonic snapshot without dropping a suffix.""" + current_text = str(current or '').strip() + candidate_text = str(candidate or '').strip() + if not candidate_text: + return False + if not current_text or candidate_text.startswith(current_text): + return True + if current_text.startswith(candidate_text): + return False + # Provider adapters can normalize a partial between the callback and the + # returned result. They are no longer prefix-comparable; retain whichever + # carries more delivered text, preferring the later result on an exact tie. + return len(candidate_text) >= len(current_text) + + +def _merge_partial_snapshots(current: dict, candidate: dict) -> dict: + """Merge two sources for one partial response into ``current`` in place.""" + if _partial_snapshot_prefers_candidate_text( + current.get('content'), candidate.get('content') + ): + current['content'] = candidate.get('content') or '' + if _partial_snapshot_prefers_candidate_text( + current.get('reasoning'), candidate.get('reasoning') + ): + current['reasoning'] = candidate.get('reasoning') + candidate_tools = candidate.get('_partial_tool_calls') or [] + current_tools = current.get('_partial_tool_calls') or [] + if candidate_tools and len(candidate_tools) >= len(current_tools): + current['_partial_tool_calls'] = copy.deepcopy(candidate_tools) + current['_partial'] = True + current.setdefault('role', 'assistant') + current.setdefault('timestamp', candidate.get('timestamp') or int(time.time())) + return current + + +def _upsert_current_turn_partial( + messages, + candidate: dict, + *, + active_turn_identity=None, +) -> dict | None: + """Persist exactly one partial assistant snapshot in the current turn. + + Live callback buffers and ``result["messages"]`` are two observations of + the same Agent response. Their bytes can differ when one observation is a + slightly newer prefix. Reconcile them into one row rather than treating + content inequality as proof of two assistant turns. + """ + if not isinstance(messages, list) or not isinstance(candidate, dict): + return None + + current_user_idx = next( + ( + index + for index in range(len(messages) - 1, -1, -1) + if _active_turn_token_matches(messages[index], active_turn_identity) + ), + None, + ) + if current_user_idx is None: + # Error settlement materializes the pending WebUI user row before this + # helper, but that recovered row intentionally has no provider-facing + # token. The last user row is therefore the display boundary. + current_user_idx = next( + ( + index + for index in range(len(messages) - 1, -1, -1) + if isinstance(messages[index], dict) + and messages[index].get('role') == 'user' + ), + None, + ) + start = (current_user_idx + 1) if current_user_idx is not None else 0 + end = len(messages) + for index in range(start, len(messages)): + if isinstance(messages[index], dict) and messages[index].get('role') == 'user': + end = index + break + + candidate_content = str(candidate.get('content') or '') + canonical_idx = next( + ( + index + for index in range(start, end) + if isinstance(messages[index], dict) + and messages[index].get('role') == 'assistant' + and not messages[index].get('_error') + and str(messages[index].get('content') or '') == candidate_content + ), + None, + ) + partial_indices = [ + index + for index in range(start, end) + if isinstance(messages[index], dict) + and messages[index].get('role') == 'assistant' + and messages[index].get('_partial') is True + and not messages[index].get('_error') + ] + if canonical_idx is None and partial_indices: + canonical_idx = partial_indices[-1] + if canonical_idx is None: + messages.append(dict(candidate)) + return messages[-1] + + canonical = messages[canonical_idx] + for partial_idx in partial_indices: + if partial_idx != canonical_idx: + _merge_partial_snapshots(canonical, messages[partial_idx]) + _merge_partial_snapshots(canonical, candidate) + # Collapse any partials left by an earlier source while preserving ordinary + # assistant/tool history in the same turn. + for index in reversed(partial_indices): + if index != canonical_idx: + del messages[index] + return canonical + + def _sse(handler, event, data): """Write one SSE event to the response stream.""" payload = f"event: {event}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n" @@ -7535,7 +7789,11 @@ def _sse_set_write_deadline(handler, seconds=None): logger.debug("Failed to arm SSE write deadline", exc_info=True) -def _materialize_pending_user_turn_before_error(session) -> bool: +def _materialize_pending_user_turn_before_error( + session, + *, + active_turn_identity=None, +) -> bool: """Persist the pending user prompt before clearing runtime stream state. Error paths often clear ``pending_user_message`` before appending an assistant @@ -7554,6 +7812,43 @@ def _materialize_pending_user_turn_before_error(session) -> bool: pending_source = getattr(session, 'pending_user_source', None) or 'webui' pending_attachments = list(getattr(session, 'pending_attachments', None) or []) + # A returned Agent result is settled into ``session.messages`` before its + # terminal classification runs. In that path the current user row can + # already own a result assistant while no longer being the final row. Do + # not append a second recovered copy: that artificial user boundary would + # split the live-buffer and result partial snapshots into two turns and + # defeat exact-once consolidation below. Only the unforgeable WebUI turn + # token is accepted here; matching repeated prompt text is not authority. + active_turn_token = ( + active_turn_identity.get('token') + if isinstance(active_turn_identity, dict) + else build_active_turn_token( + getattr(session, 'active_stream_id', None), + pending_started_at, + ) + ) + session_messages = getattr(session, 'messages', None) + if active_turn_token and isinstance(session_messages, list): + for index in range(len(session_messages) - 1, -1, -1): + message = session_messages[index] + if not ( + isinstance(message, dict) + and message.get('role') == 'user' + and message.get('_active_turn_token') == active_turn_token + ): + continue + normalized_user = _normalize_user_text( + _message_text(message.get('content')) + ) + if normalized_user != _normalize_user_text(pending_text): + break + if not any( + isinstance(later, dict) and later.get('role') == 'user' + for later in session_messages[index + 1 :] + ): + return False + break + def is_exact_checkpoint(messages): if not isinstance(messages, list) or not messages: return False @@ -7661,7 +7956,12 @@ def _build_partial_message(content_text, reasoning_text, tool_calls) -> dict | N return _msg -def _snapshot_and_append_partial_on_error(session, stream_id) -> dict | None: +def _snapshot_and_append_partial_on_error( + session, + stream_id, + *, + active_turn_identity=None, +) -> dict | None: """Snapshot streaming buffers under STREAMS_LOCK and append a _partial message. Uses _build_partial_message() for the shared thinking-strip + dict-build logic. @@ -7724,10 +8024,110 @@ def _snapshot_and_append_partial_on_error(session, stream_id) -> dict | None: return None if not isinstance(session.messages, list): session.messages = [] - if not _partial_marker_already_present(session.messages, _partial_msg): - session.messages.append(_partial_msg) - return _partial_msg - return None + return _upsert_current_turn_partial( + session.messages, + _partial_msg, + active_turn_identity=active_turn_identity, + ) + + +def _append_result_partial_on_error( + session, + result, + pre_call_context, + msg_text, + *, + active_turn_identity=None, +) -> dict | None: + """Retain only a current-turn partial assistant row from an Agent result.""" + if not isinstance(result, dict) or result.get('partial') is not True: + return None + messages = result.get('messages') + if not isinstance(messages, list) or not isinstance(pre_call_context, list): + return None + normalized_msg_text = _normalize_user_text(msg_text) + if _messages_have_prefix(messages, pre_call_context): + # Append-only result: only rows after the pre-call baseline can + # belong to this call. + current_turn_rows = messages[len(pre_call_context):] + baseline_has_current_user = bool( + pre_call_context + and isinstance(pre_call_context[-1], dict) + and pre_call_context[-1].get('role') == 'user' + and _normalize_user_text(_message_text(pre_call_context[-1].get('content'))) + == normalized_msg_text + ) + if not baseline_has_current_user: + current_user_index = next( + ( + index + for index, row in enumerate(current_turn_rows) + if isinstance(row, dict) + and row.get('role') == 'user' + and _normalize_user_text(_message_text(row.get('content'))) + == normalized_msg_text + ), + None, + ) + if current_user_index is None: + return None + current_turn_rows = current_turn_rows[current_user_index + 1:] + else: + # Compacted/replayed results can REPLACE the pre-call baseline instead + # of appending to it; a numeric suffix slice can then expose a + # historical assistant row as this turn's partial. Require the unique + # Agent-resolved turn boundary. WebUI's private token is preferred when + # it survives, but compacted Agent projections legitimately omit it. + # In that case require the authoritative Agent turn id/index and validate + # that the indexed row is this normalized current user prompt. + current_user_idx = _find_active_turn_checkpoint_index( + messages, + pre_call_context, + active_turn_identity, + msg_text, + ) + if current_user_idx is None: + return None + current_turn_rows = messages[current_user_idx + 1:] + # A later user row starts another turn. Never cross that boundary while + # selecting a partial assistant response for the current turn. + next_user_idx = next( + ( + index + for index, row in enumerate(current_turn_rows) + if isinstance(row, dict) and row.get('role') == 'user' + ), + None, + ) + if next_user_idx is not None: + current_turn_rows = current_turn_rows[:next_user_idx] + assistant_row = next( + ( + row + for row in reversed(current_turn_rows) + if isinstance(row, dict) + and row.get('role') == 'assistant' + and row.get('content') + and not row.get('_error') + ), + None, + ) + if assistant_row is None: + return None + partial_msg = _build_partial_message( + str(assistant_row.get('content') or ''), + None, + None, + ) + if partial_msg is None: + return None + if not isinstance(session.messages, list): + session.messages = [] + return _upsert_current_turn_partial( + session.messages, + partial_msg, + active_turn_identity=active_turn_identity, + ) def _last_resort_sync_from_core(session, stream_id, agent_lock): @@ -7892,6 +8292,64 @@ def _replace_session_db_in_kwargs(agent_kwargs, state_db_path): return _next_session_db +def _add_supported_run_conversation_kwarg(callable_obj, kwargs, name, value): + """Add a run-conversation kwarg only when the live Agent accepts it.""" + if not isinstance(kwargs, dict): + return False + try: + import inspect + + parameters = inspect.signature(callable_obj).parameters + except (TypeError, ValueError): + return False + if name in parameters or any( + parameter.kind is inspect.Parameter.VAR_KEYWORD + for parameter in parameters.values() + ): + kwargs[name] = value + return True + return False + + +def _build_run_conversation_kwargs( + callable_obj, + *, + user_message, + system_message, + conversation_history, + conversation_history_revision, + task_id, + persist_user_message, + persist_user_timestamp, +): + """Build one rolling-compatible Agent invocation contract. + + ``persist_user_timestamp`` is signature-gated (#6935): an older + hermes-agent whose ``run_conversation()`` predates the kwarg must not + receive it, or the call trips a TypeError before the turn starts. + """ + kwargs = { + "user_message": user_message, + "system_message": system_message, + "conversation_history": conversation_history, + "task_id": task_id, + "persist_user_message": persist_user_message, + } + _add_supported_run_conversation_kwarg( + callable_obj, + kwargs, + "persist_user_timestamp", + persist_user_timestamp, + ) + _add_supported_run_conversation_kwarg( + callable_obj, + kwargs, + "conversation_history_revision", + conversation_history_revision, + ) + return kwargs + + def _attempt_credential_self_heal( provider_id, session_id, _agent_lock_ref, *, target_model=None, ): @@ -8269,6 +8727,8 @@ def _run_agent_streaming( old_session_platform = None old_hermes_home = None old_profile_env = {} + result = None + _result_partial_pre_call_context = [] # MCP discovery moved to AFTER the per-profile HERMES_HOME mutation below # (was here at v0.51.30) — the previous placement always read the default @@ -10086,31 +10546,68 @@ 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() - _external_state_messages = get_state_db_session_messages(getattr(s, 'session_id', None)) + _external_state_snapshot = get_state_db_session_messages( + session_id, + profile=getattr(s, 'profile', None), + with_revision=True, + ) + + def _context_and_revision_from_state_snapshot(state_snapshot): + reconciled_snapshot = reconciled_state_db_messages_for_session( + s, + prefer_context=True, + state_messages=state_snapshot, + with_revision=True, + ) + if not isinstance(reconciled_snapshot, StateDBSessionMessagesSnapshot): + raise TypeError( + "state.db context reconciliation did not return a revision snapshot" + ) + context_messages = _new_turn_context_from_messages( + reconciled_snapshot.messages, + msg_text, + ) + return ( + _deduplicate_context_messages(context_messages), + reconciled_snapshot.revision, + ) + + def _refresh_context_and_revision_from_state_db(): + fresh_state_snapshot = get_state_db_session_messages( + session_id, + profile=getattr(s, 'profile', None), + with_revision=True, + ) + return _context_and_revision_from_state_snapshot(fresh_state_snapshot) + _previous_messages = list( reconciled_state_db_messages_for_session( s, - state_messages=_external_state_messages, + state_messages=_external_state_snapshot, ) or [] ) + # Keep the owner/context projection distinct from the display projection. + # Settlement and stale-result attribution require the exact pre-run owner + # context even when state.db reconciliation changes the visible transcript. _previous_owner_context_messages = list( reconciled_state_db_messages_for_session( s, prefer_context=True, - state_messages=_external_state_messages, + state_messages=_external_state_snapshot, ) or [] ) _previous_owner_context_messages = _deduplicate_context_messages( _previous_owner_context_messages ) - _previous_context_messages = _new_turn_context_from_messages( - _previous_owner_context_messages, - msg_text, + ( + _previous_context_messages, + _conversation_history_revision, + ) = _context_and_revision_from_state_snapshot( + _external_state_snapshot, ) # Dedup before feeding to agent — merge_session_messages_append_only # can produce duplicates when context_messages and state.db share # messages with different timestamps. - _previous_context_messages = _deduplicate_context_messages(_previous_context_messages) _pre_compression_count = getattr( getattr(agent, 'context_compressor', None), 'compression_count', 0, @@ -10186,7 +10683,8 @@ def _run_agent_streaming( _agent_msg_text = "\n\n".join([*_process_notifications, msg_text]).strip() user_message = _build_native_multimodal_message(workspace_ctx, _agent_msg_text, attachments, workspace, cfg=_cfg, active_provider=(resolved_provider or ""), active_model=(resolved_model or ""), requested_provider=(_session_requested_provider or "")) _persistent_state_before = _persistent_state_snapshot(_profile_home) - _run_conversation_kwargs = dict( + _run_conversation_kwargs = _build_run_conversation_kwargs( + agent.run_conversation, user_message=user_message, system_message=workspace_system_msg, conversation_history=_sanitize_messages_for_agent( @@ -10197,11 +10695,11 @@ def _run_agent_streaming( effective_base_url=resolved_base_url, requested_provider=(_session_requested_provider or ""), ), + conversation_history_revision=_conversation_history_revision, task_id=session_id, persist_user_message=msg_text, + persist_user_timestamp=getattr(s, 'pending_started_at', None), ) - if _supports_kwarg(agent.run_conversation, 'persist_user_timestamp'): - _run_conversation_kwargs['persist_user_timestamp'] = getattr(s, 'pending_started_at', None) # Only pass moa_config when a /moa override is actually active, so a # normal send never trips a TypeError on an older hermes-agent whose # run_conversation() predates the moa_config kwarg. @@ -10238,6 +10736,7 @@ def _run_agent_streaming( requested_provider=(_session_requested_provider or ""), ) _run_conversation_kwargs["user_message"] = user_message + _result_partial_pre_call_context = list(_previous_context_messages) result = agent.run_conversation(**_run_conversation_kwargs) _active_turn_identity = _resolve_active_turn_authority( _active_turn_identity, @@ -10556,6 +11055,7 @@ def _run_agent_streaming( str(_last_err) if _last_err else '', _last_err, silent_failure=not bool(_last_err), + result=result, ) _is_quota = _classification['type'] == 'quota_exhausted' _is_auth = _classification['type'] == 'auth_mismatch' @@ -10647,6 +11147,7 @@ def _run_agent_streaming( # Before emitting the error, try re-reading credentials # and retrying once with a fresh agent. _heal_result = None + _heal_stale_classification = None # Bind the session's profile so the self-heal re-resolve # AND the custom-provider override below read one # profile-owned snapshot (finding #3): otherwise a named @@ -10699,32 +11200,52 @@ def _run_agent_streaming( _self_healed = True _token_sent = False try: - _heal_kwargs = dict( + ( + _heal_context_messages, + _heal_conversation_history_revision, + ) = _refresh_context_and_revision_from_state_db() + _heal_kwargs = _build_run_conversation_kwargs( + agent.run_conversation, user_message=user_message, system_message=workspace_system_msg, conversation_history=_sanitize_messages_for_agent( - _previous_context_messages, + _heal_context_messages, cfg=_cfg, effective_model=resolved_model, effective_provider=resolved_provider, effective_base_url=resolved_base_url, requested_provider=(_session_requested_provider or ""), ), + conversation_history_revision=( + _heal_conversation_history_revision + ), task_id=session_id, persist_user_message=msg_text, + persist_user_timestamp=getattr(s, 'pending_started_at', None), ) - if _supports_kwarg(agent.run_conversation, 'persist_user_timestamp'): - _heal_kwargs['persist_user_timestamp'] = getattr(s, 'pending_started_at', None) if moa_config is not None: _heal_kwargs["moa_config"] = moa_config + _result_partial_pre_call_context = list( + _heal_context_messages + ) _heal_result = agent.run_conversation(**_heal_kwargs) _active_turn_identity = _resolve_active_turn_authority( _active_turn_identity, result=_heal_result, agent=agent, ) - _heal_all_msgs = _heal_result.get('messages') or [] - _heal_ok = _has_new_assistant_reply(_heal_all_msgs, _prev_len) or _token_sent + if _result_reports_compression_snapshot_stale(_heal_result): + _heal_stale_classification = _classify_provider_error( + '', + result=_heal_result, + ) + result = _heal_result + _heal_ok = _self_heal_result_succeeded( + _heal_result, + _heal_context_messages, + _active_turn_identity, + msg_text, + ) except Exception as _retry_exc: logger.warning( '[webui] self-heal: retry also failed: %s', _retry_exc, @@ -10763,14 +11284,19 @@ def _run_agent_streaming( # leaving _assistant_added truthy (set below). _assistant_added = True # prevent re-entering guard if not _assistant_added: - # Self-heal didn't apply or retry failed — emit error - _err_label = 'Authentication failed' - _err_type = 'auth_mismatch' - _err_hint = ( - 'The selected model may not be supported by your configured provider or ' - 'your API key is invalid. Run `hermes model` in your terminal to ' - 'update credentials, then restart the WebUI.' - ) + # Self-heal didn't apply or retry failed — emit error. + if _heal_stale_classification is not None: + _err_label = _heal_stale_classification['label'] + _err_type = _heal_stale_classification['type'] + _err_hint = _heal_stale_classification['hint'] + else: + _err_label = 'Authentication failed' + _err_type = 'auth_mismatch' + _err_hint = ( + 'The selected model may not be supported by your configured provider or ' + 'your API key is invalid. Run `hermes model` in your terminal to ' + 'update credentials, then restart the WebUI.' + ) elif _is_auth: _err_label = 'Authentication failed' _err_type = 'auth_mismatch' @@ -10801,8 +11327,13 @@ def _run_agent_streaming( # fall through to normal post-result persistence below. pass else: + _result_public_error = _err_str or f'{_err_label}.' + if _err_type == 'compression_snapshot_stale': + _result_public_error = ( + 'The conversation changed while context compression was being prepared.' + ) _error_payload = _provider_error_payload( - _err_str or f'{_err_label}.', + _result_public_error, _err_type, _err_hint, ) @@ -10827,14 +11358,28 @@ def _run_agent_streaming( ) _error_payload['hint'] = _err_hint _turn_duration = _terminal_turn_duration(s) - _materialize_pending_user_turn_before_error(s) + _materialize_pending_user_turn_before_error( + s, + active_turn_identity=_active_turn_identity, + ) s.active_stream_id = None s.pending_user_message = None s.pending_attachments = [] s.pending_started_at = None s.pending_user_source = None try: - _snapshot_and_append_partial_on_error(s, stream_id) + _snapshot_and_append_partial_on_error( + s, + stream_id, + active_turn_identity=_active_turn_identity, + ) + _append_result_partial_on_error( + s, + result, + _result_partial_pre_call_context, + msg_text, + active_turn_identity=_active_turn_identity, + ) except Exception: logger.debug("Failed to snapshot partials on error for %s", stream_id, exc_info=True) _error_content = ( @@ -11880,6 +12425,9 @@ def _run_agent_streaming( _exc_is_cancelled = _classification['type'] == 'cancelled' _exc_is_interrupted = _classification['type'] == 'interrupted' _exc_is_compression_exhausted = _classification['type'] == 'compression_exhausted' + _exc_is_compression_snapshot_stale = ( + _classification['type'] == 'compression_snapshot_stale' + ) # The user hint still points to Settings / `hermes model` from _classify_provider_error(). if _exc_is_quota: @@ -11895,6 +12443,7 @@ def _run_agent_streaming( _classification['label'], _classification['type'], _classification['hint'], ) elif _exc_is_auth: + _heal_stale_classification = None if not _self_healed: # ── Credential self-heal on 401 (#1401) ── # Bind the session's profile so the self-heal re-resolve AND the @@ -11949,70 +12498,126 @@ def _run_agent_streaming( # Retry the conversation _token_sent = False try: - _heal_kwargs2 = dict( + ( + _heal_context_messages, + _heal_conversation_history_revision, + ) = _refresh_context_and_revision_from_state_db() + _heal_kwargs2 = _build_run_conversation_kwargs( + _heal_agent.run_conversation, user_message=user_message, system_message=workspace_system_msg, conversation_history=_sanitize_messages_for_agent( - _previous_context_messages, + _heal_context_messages, cfg=_cfg, effective_model=resolved_model, effective_provider=resolved_provider, effective_base_url=resolved_base_url, requested_provider=(_session_requested_provider or ""), ), + conversation_history_revision=( + _heal_conversation_history_revision + ), task_id=session_id, persist_user_message=msg_text, + persist_user_timestamp=getattr(s, 'pending_started_at', None), ) - if _supports_kwarg(_heal_agent.run_conversation, 'persist_user_timestamp'): - _heal_kwargs2['persist_user_timestamp'] = getattr(s, 'pending_started_at', None) if moa_config is not None: _heal_kwargs2["moa_config"] = moa_config + _result_partial_pre_call_context = list( + _heal_context_messages + ) _heal_result = _heal_agent.run_conversation(**_heal_kwargs2) _active_turn_identity = _resolve_active_turn_authority( _active_turn_identity, result=_heal_result, agent=_heal_agent, ) - # Retry succeeded — persist the result normally - if s is not None: - if _checkpoint_stop is not None: - _checkpoint_stop.set() - if _ckpt_thread is not None: - _ckpt_thread.join(timeout=15) - _lock_ctx = _agent_lock if _agent_lock is not None else contextlib.nullcontext() - with _lock_ctx: - if not ephemeral and not _stream_writeback_is_current(s, stream_id): - logger.info( - "Skipping stale stream self-heal writeback for session %s stream %s; active_stream_id=%s", - getattr(s, 'session_id', session_id), - stream_id, - getattr(s, 'active_stream_id', None), + _heal_stale_classification = None + if _result_reports_compression_snapshot_stale(_heal_result): + _heal_stale_classification = _classify_provider_error( + '', + result=_heal_result, + ) + result = _heal_result + elif _self_heal_result_succeeded( + _heal_result, + _heal_context_messages, + _active_turn_identity, + msg_text, + ): + # Retry succeeded — persist the result normally. + _done_session_payload = None + if s is not None: + if _checkpoint_stop is not None: + _checkpoint_stop.set() + if _ckpt_thread is not None: + _ckpt_thread.join(timeout=15) + _lock_ctx = _agent_lock if _agent_lock is not None else contextlib.nullcontext() + with _lock_ctx: + if not ephemeral and not _stream_writeback_is_current(s, stream_id): + logger.info( + "Skipping stale stream self-heal writeback for session %s stream %s; active_stream_id=%s", + getattr(s, 'session_id', session_id), + stream_id, + getattr(s, 'active_stream_id', None), + ) + return + _result_messages = _heal_result.get('messages') + if _result_messages is None: + _result_messages = _previous_context_messages + _result_messages = _settle_result_messages( + s, + _previous_messages, + _previous_owner_context_messages, + _result_messages, + msg_text, + _turn_pending_source, + _active_turn_identity, ) - return - _result_messages = _heal_result.get('messages') - if _result_messages is None: - _result_messages = _previous_context_messages - _result_messages = _settle_result_messages( - s, - _previous_messages, - _previous_owner_context_messages, - _result_messages, - msg_text, - _turn_pending_source, - _active_turn_identity, - ) - s.save() - logger.info('[webui] self-heal (except path): retry succeeded') - return # skip error emission + # Terminal self-heal success must finalize the + # turn exactly once: clear the pending markers + # so the last-resort recovery sync in the outer + # ``finally`` cannot re-materialize this user + # turn and append a spurious "Response + # interrupted" marker after the settled answer. + s.tool_calls = _extract_tool_calls_from_messages( + s.messages, + live_tool_calls=_live_tool_calls, + ) + s.active_stream_id = None + s.pending_user_message = None + s.pending_attachments = [] + s.pending_started_at = None + s.pending_user_source = None + s.save() + _done_session_payload = redact_session_data( + _session_payload_with_full_messages( + s, tool_calls=s.tool_calls + ) + ) + if _done_session_payload is not None: + put('done', { + 'session': _done_session_payload, + 'usage': {'input_tokens': 0, 'output_tokens': 0}, + }) + put('stream_end', {'session_id': session_id}) + logger.info('[webui] self-heal (except path): retry succeeded') + return # skip error emission except Exception as _retry_exc2: logger.warning('[webui] self-heal (except path): retry failed: %s', _retry_exc2) # Fall through to emit the original error - # Self-heal didn't apply or retry failed — emit the auth error - _exc_label, _exc_type, _exc_hint = ( - 'Authentication error', 'auth_mismatch', - 'The selected model may not be supported by your configured provider. ' - 'Run `hermes model` in your terminal to switch providers, then restart the WebUI.', - ) + if _heal_stale_classification is not None: + _exc_label = _heal_stale_classification['label'] + _exc_type = _heal_stale_classification['type'] + _exc_hint = _heal_stale_classification['hint'] + _exc_is_compression_snapshot_stale = True + else: + # Self-heal didn't apply or retry failed — emit the auth error. + _exc_label, _exc_type, _exc_hint = ( + 'Authentication error', 'auth_mismatch', + 'The selected model may not be supported by your configured provider. ' + 'Run `hermes model` in your terminal to switch providers, then restart the WebUI.', + ) elif _exc_is_not_found: _exc_label, _exc_type, _exc_hint = ( _classification['label'], _classification['type'], _classification['hint'], @@ -12021,6 +12626,10 @@ def _run_agent_streaming( _exc_label, _exc_type, _exc_hint = ( _classification['label'], _classification['type'], _classification['hint'], ) + elif _exc_is_compression_snapshot_stale: + _exc_label, _exc_type, _exc_hint = ( + _classification['label'], _classification['type'], _classification['hint'], + ) elif _exc_is_compression_exhausted: _exc_label, _exc_type, _exc_hint = ( _classification['label'], _classification['type'], _classification['hint'], @@ -12028,6 +12637,12 @@ def _run_agent_streaming( else: _exc_label, _exc_type, _exc_hint = 'Error', 'error', '' + if _exc_is_compression_snapshot_stale: + # The typed exception may carry expected/observed revision details for + # diagnostics. Keep those out of the persisted transcript and SSE UI. + err_str = ( + 'The conversation changed while context compression was being prepared.' + ) _error_payload = _provider_error_payload(err_str, _exc_type, _exc_hint) if s is not None: if _checkpoint_stop is not None: @@ -12078,6 +12693,11 @@ def _run_agent_streaming( ) _error_payload['hint'] = _exc_hint _turn_duration = _terminal_turn_duration(s) + # Keep the canonical one-argument error-settlement shape pinned + # by #1361/#2136. The helper derives the same stream-owned turn + # token from active_stream_id + pending_started_at when the + # explicit identity is omitted, so repeated prompts remain + # fenced without weakening pending-turn durability. _materialize_pending_user_turn_before_error(s) s.active_stream_id = None s.pending_user_message = None @@ -12085,7 +12705,18 @@ def _run_agent_streaming( s.pending_started_at = None s.pending_user_source = None try: - _snapshot_and_append_partial_on_error(s, stream_id) + _snapshot_and_append_partial_on_error( + s, + stream_id, + active_turn_identity=_active_turn_identity, + ) + _append_result_partial_on_error( + s, + result, + _result_partial_pre_call_context, + msg_text, + active_turn_identity=_active_turn_identity, + ) except Exception: logger.debug("Failed to snapshot partials on error for %s", stream_id, exc_info=True) _error_message = { diff --git a/tests/test_compression_snapshot_revision.py b/tests/test_compression_snapshot_revision.py new file mode 100644 index 00000000..7d0b9411 --- /dev/null +++ b/tests/test_compression_snapshot_revision.py @@ -0,0 +1,1496 @@ +"""Durable transcript revision coverage for WebUI compression handoff.""" + +import queue +import sqlite3 +import sys +import types +from collections import OrderedDict + +import pytest + +from api import models, streaming +from api.models import Session + +pytestmark = pytest.mark.requires_agent_modules + + +@pytest.mark.parametrize("revision", [None, {"session_id": "session-1"}]) +def test_run_conversation_revision_kwarg_is_omitted_for_legacy_agent(revision): + class LegacyAgent: + def run_conversation( + self, + user_message, + system_message, + conversation_history, + task_id, + persist_user_message, + ): + return None + + kwargs = {} + supported = streaming._add_supported_run_conversation_kwarg( + LegacyAgent().run_conversation, + kwargs, + "conversation_history_revision", + revision, + ) + + assert supported is False + assert "conversation_history_revision" not in kwargs + + +@pytest.mark.parametrize("revision", [None, {"session_id": "session-1"}]) +@pytest.mark.parametrize("mode", ["explicit", "variadic"]) +def test_run_conversation_revision_kwarg_is_added_for_capable_agent(revision, mode): + class ExplicitAgent: + def run_conversation(self, *, conversation_history_revision=None): + return None + + class VariadicAgent: + def run_conversation(self, **kwargs): + return None + + agent = ExplicitAgent() if mode == "explicit" else VariadicAgent() + kwargs = {} + supported = streaming._add_supported_run_conversation_kwarg( + agent.run_conversation, + kwargs, + "conversation_history_revision", + revision, + ) + + assert supported is True + assert kwargs["conversation_history_revision"] is revision + + +def test_build_run_conversation_kwargs_omits_revision_for_strict_legacy_agent(): + class LegacyAgent: + def run_conversation( + self, + user_message, + system_message, + conversation_history, + task_id, + persist_user_message, + persist_user_timestamp, + ): + return None + + agent = LegacyAgent() + kwargs = streaming._build_run_conversation_kwargs( + agent.run_conversation, + user_message="hello", + system_message="system", + conversation_history=[{"role": "user", "content": "prior"}], + conversation_history_revision={"session_id": "session-1"}, + task_id="session-1", + persist_user_message="hello", + persist_user_timestamp=123.0, + ) + + assert kwargs == { + "user_message": "hello", + "system_message": "system", + "conversation_history": [{"role": "user", "content": "prior"}], + "task_id": "session-1", + "persist_user_message": "hello", + "persist_user_timestamp": 123.0, + } + + +def _make_state_db(path, sid, rows): + conn = sqlite3.connect(path) + try: + conn.execute( + """ + CREATE TABLE messages ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + session_id TEXT NOT NULL, + role TEXT, + content TEXT, + timestamp REAL, + active INTEGER + ) + """ + ) + conn.executemany( + "INSERT INTO messages (session_id, role, content, timestamp, active) " + "VALUES (?, ?, ?, ?, ?)", + [ + ( + sid, + row["role"], + row["content"], + row.get("timestamp"), + row.get("active", 1), + ) + for row in rows + ], + ) + conn.commit() + finally: + conn.close() + + +def _append_state_row(path, sid, *, role, content, timestamp, active=1): + conn = sqlite3.connect(path) + try: + conn.execute( + "INSERT INTO messages (session_id, role, content, timestamp, active) " + "VALUES (?, ?, ?, ?, ?)", + (sid, role, content, timestamp, active), + ) + conn.commit() + finally: + conn.close() + + +def _install_streaming_session(monkeypatch, tmp_path, *, sid, stream_id, messages, context_messages): + import api.config as config + + session_dir = tmp_path / "sessions" + session_dir.mkdir(exist_ok=True) + 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(streaming, "SESSIONS", OrderedDict(), raising=False) + monkeypatch.setattr(models, "_active_state_db_path", lambda: tmp_path / "state.db") + + config.STREAMS.clear() + config.CANCEL_FLAGS.clear() + config.AGENT_INSTANCES.clear() + config.SESSION_AGENT_LOCKS.clear() + + session = Session( + session_id=sid, + title="Durable revision", + workspace=str(tmp_path), + model="test-model", + messages=list(messages), + context_messages=list(context_messages), + ) + session.active_stream_id = stream_id + session.pending_user_message = "new webui turn" + session.pending_started_at = 10.0 + session.save(touch_updated_at=False) + models.SESSIONS[sid] = session + streaming.SESSIONS[sid] = session + + event_queue = queue.Queue() + config.STREAMS[stream_id] = event_queue + streaming.STREAMS[stream_id] = event_queue + + monkeypatch.setattr(streaming, "get_session", lambda _sid: session) + monkeypatch.setattr( + streaming, + "resolve_model_provider", + lambda *_args, **_kwargs: ("test-model", "test-provider", None), + ) + monkeypatch.setattr(streaming, "get_config", lambda: {}) + monkeypatch.setattr(config, "get_config", lambda: {}) + monkeypatch.setattr(config, "_resolve_cli_toolsets", lambda *_args, **_kwargs: []) + + fake_hermes_state = types.ModuleType("hermes_state") + fake_hermes_state.__dict__["SessionDB"] = lambda *_args, **_kwargs: object() + monkeypatch.setitem(sys.modules, "hermes_state", fake_hermes_state) + + return session, event_queue + + +def _drain_events(event_queue): + events = [] + while not event_queue.empty(): + events.append(event_queue.get_nowait()) + return events + + +def test_state_db_reader_returns_active_messages_with_matching_revision(tmp_path, monkeypatch): + sid = "revision-reader" + db_path = tmp_path / "state.db" + _make_state_db( + db_path, + sid, + [ + {"role": "user", "content": "inactive", "timestamp": 1.0, "active": 0}, + {"role": "user", "content": "active user", "timestamp": 2.0}, + {"role": "assistant", "content": "active answer", "timestamp": 3.0}, + ], + ) + monkeypatch.setattr(models, "_active_state_db_path", lambda: db_path) + + snapshot = models.get_state_db_session_messages(sid, with_revision=True) + + assert [message["content"] for message in snapshot.messages] == [ + "active user", + "active answer", + ] + assert snapshot.revision == { + "session_id": sid, + "active_message_count": 2, + "max_active_message_id": 3, + } + assert isinstance(models.get_state_db_session_messages(sid), list) + + +def test_state_db_reader_revision_includes_api_content_digest( + tmp_path, monkeypatch +): + """The WebUI revision must carry the Agent's model-facing sidecar fence.""" + sid = "revision-api-content-digest" + db_path = tmp_path / "state.db" + conn = sqlite3.connect(db_path) + try: + conn.execute( + "CREATE TABLE messages (" + "id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT NOT NULL, " + "role TEXT, content TEXT, timestamp REAL, active INTEGER, " + "api_content TEXT)" + ) + conn.execute( + "INSERT INTO messages " + "(session_id, role, content, timestamp, active, api_content) " + "VALUES (?, ?, ?, ?, ?, ?)", + (sid, "user", "clean", 1.0, 1, "wire-v1"), + ) + conn.commit() + finally: + conn.close() + monkeypatch.setattr(models, "_active_state_db_path", lambda: db_path) + + first = models.get_state_db_session_messages(sid, with_revision=True) + assert first.revision == { + "session_id": sid, + "active_message_count": 1, + "max_active_message_id": 1, + "active_rows_digest": ( + "50101e75ded18d78bc21646a6fcd4cddb8f07901838635dda6eac85d219c1359" + ), + } + + conn = sqlite3.connect(db_path) + try: + conn.execute( + "UPDATE messages SET api_content = ? WHERE session_id = ?", + ("wire-v2", sid), + ) + conn.commit() + finally: + conn.close() + + second = models.get_state_db_session_messages(sid, with_revision=True) + assert second.revision == { + "session_id": sid, + "active_message_count": 1, + "max_active_message_id": 1, + "active_rows_digest": ( + "de45994b9f134be94b99ce03a47da3b553c86a27e5cdcfd67885644a8372102a" + ), + } + + +def test_state_db_revision_is_unavailable_when_selected_row_has_null_active( + tmp_path, monkeypatch +): + sid = "revision-null-active" + db_path = tmp_path / "state.db" + _make_state_db( + db_path, + sid, + [{"role": "user", "content": "legacy active row", "timestamp": 1.0, "active": None}], + ) + monkeypatch.setattr(models, "_active_state_db_path", lambda: db_path) + + snapshot = models.get_state_db_session_messages(sid, with_revision=True) + + assert [message["content"] for message in snapshot.messages] == ["legacy active row"] + assert snapshot.revision is None + + +def test_state_db_revision_is_unavailable_without_active_column(tmp_path, monkeypatch): + sid = "revision-no-active-column" + db_path = tmp_path / "state.db" + conn = sqlite3.connect(db_path) + try: + conn.execute( + "CREATE TABLE messages (" + "id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT NOT NULL, " + "role TEXT, content TEXT, timestamp REAL)" + ) + conn.execute( + "INSERT INTO messages (session_id, role, content, timestamp) VALUES (?, ?, ?, ?)", + (sid, "user", "legacy schema row", 1.0), + ) + conn.commit() + finally: + conn.close() + monkeypatch.setattr(models, "_active_state_db_path", lambda: db_path) + + snapshot = models.get_state_db_session_messages(sid, with_revision=True) + + assert [message["content"] for message in snapshot.messages] == ["legacy schema row"] + assert snapshot.revision is None + + +def test_snapshot_reader_missing_explicit_profile_does_not_fall_back_to_active_db( + tmp_path, + monkeypatch, +): + sid = "missing-profile-snapshot" + foreign_db = tmp_path / "active-home" / "state.db" + foreign_db.parent.mkdir() + _make_state_db( + foreign_db, + sid, + [{"role": "user", "content": "foreign profile row", "timestamp": 1.0}], + ) + missing_home = tmp_path / "missing-profile" + monkeypatch.setattr(models, "_get_profile_home", lambda _profile: missing_home) + monkeypatch.setattr(models, "_active_state_db_path", lambda: foreign_db) + + snapshot = models.get_state_db_session_messages( + sid, + profile="missing", + with_revision=True, + ) + + assert snapshot.messages == [] + assert snapshot.revision is None + + +def test_reconciled_reader_missing_explicit_profile_does_not_fall_back_to_active_db( + tmp_path, + monkeypatch, +): + sid = "missing-profile-reconcile" + foreign_db = tmp_path / "active-home" / "state.db" + foreign_db.parent.mkdir() + _make_state_db( + foreign_db, + sid, + [{"role": "user", "content": "foreign profile row", "timestamp": 1.0}], + ) + missing_home = tmp_path / "missing-profile" + monkeypatch.setattr(models, "_get_profile_home", lambda _profile: missing_home) + monkeypatch.setattr(models, "_active_state_db_path", lambda: foreign_db) + session = Session( + session_id=sid, + profile="missing", + messages=[], + context_messages=[], + ) + + reconciled = models.reconciled_state_db_messages_for_session( + session, + with_revision=True, + ) + + assert reconciled.messages == [] + assert reconciled.revision is None + + +def test_stale_partial_requires_authoritative_current_turn_boundary(): + session = Session( + session_id="partial-authority", + messages=[], + context_messages=[], + ) + result = { + "partial": True, + "messages": [ + {"role": "user", "content": "repeat prompt"}, + {"role": "assistant", "content": "historical answer"}, + {"role": "user", "content": "later unrelated turn"}, + {"role": "assistant", "content": "later answer"}, + ] + } + identity = { + "token": "stream:123", + "text": "repeat prompt", + # This index belonged to the replaced projection but now happens to + # point at an identical historical prompt. It is not current-turn proof. + "current_turn_user_idx": 0, + "turn_id": "session:turn-current", + } + + streaming._append_result_partial_on_error( + session, + result, + [{"role": "user", "content": "unrelated baseline"}], + "repeat prompt", + active_turn_identity=identity, + ) + + assert not any(message.get("_partial") for message in session.messages) + + +def test_stale_non_prefix_partial_uses_token_and_stops_at_next_user(): + session = Session( + session_id="stale-partial-token-boundary", + messages=[], + context_messages=[], + ) + result = { + "partial": True, + "messages": [ + {"role": "user", "content": "historical"}, + {"role": "assistant", "content": "historical answer"}, + { + "role": "user", + "content": "current prompt", + "_active_turn_token": "current-turn-token", + }, + {"role": "assistant", "content": "current partial"}, + {"role": "user", "content": "later turn"}, + {"role": "assistant", "content": "later answer"}, + ], + } + + appended = streaming._append_result_partial_on_error( + session, + result, + [{"role": "user", "content": "different baseline"}], + "current prompt", + active_turn_identity={ + "token": "current-turn-token", + "turn_id": "turn-current", + "current_turn_user_idx": 2, + "text": "current prompt", + }, + ) + + assert appended is not None + assert appended["content"] == "current partial" + assert [message["content"] for message in session.messages] == [ + "current partial" + ] + + +def test_non_prefix_partial_uses_agent_turn_boundary_without_webui_token(): + session = Session(session_id="agent-boundary-partial", messages=[], context_messages=[]) + result = { + "partial": True, + "turn_id": "agent-turn-42", + "current_turn_user_idx": 2, + "messages": [ + {"role": "user", "content": "repeat prompt"}, + {"role": "assistant", "content": "historical answer"}, + {"role": "user", "content": " repeat prompt "}, + {"role": "assistant", "content": "current partial"}, + {"role": "user", "content": "later turn"}, + {"role": "assistant", "content": "later answer"}, + ], + } + identity = streaming._resolve_active_turn_authority( + {"token": "webui-token", "text": "repeat prompt", "turn_id": "", "current_turn_user_idx": None}, + result=result, + ) + + appended = streaming._append_result_partial_on_error( + session, result, [{"role": "user", "content": "replaced baseline"}], + "repeat prompt", active_turn_identity=identity, + ) + + assert appended["content"] == "current partial" + + +@pytest.mark.parametrize("terminal", ["returned", "raised"]) +def test_non_prefix_self_heal_accepts_agent_turn_boundary_without_webui_token(terminal): + result = { + "completed": True, + "turn_id": "agent-turn-43", + "current_turn_user_idx": 2, + "messages": [ + {"role": "user", "content": "repeat prompt"}, + {"role": "assistant", "content": "historical answer"}, + {"role": "user", "content": " repeat\n prompt "}, + {"role": "assistant", "content": f"healed after {terminal}"}, + {"role": "user", "content": "later turn"}, + {"role": "assistant", "content": "later answer"}, + ], + } + identity = streaming._resolve_active_turn_authority( + {"token": "webui-token", "text": "repeat prompt", "turn_id": "", "current_turn_user_idx": None}, + result=result, + ) + + assert streaming._self_heal_result_succeeded( + result, [{"role": "user", "content": "replaced baseline"}], identity, "repeat prompt" + ) + + +def test_non_prefix_agent_boundary_rejects_wrong_normalized_user(): + result = { + "completed": True, + "turn_id": "agent-turn-stale", + "current_turn_user_idx": 0, + "messages": [ + {"role": "user", "content": "different prompt"}, + {"role": "assistant", "content": "historical answer"}, + ], + } + identity = streaming._resolve_active_turn_authority( + {"token": "absent", "text": "repeat prompt", "turn_id": "", "current_turn_user_idx": None}, + result=result, + ) + + assert not streaming._self_heal_result_succeeded( + result, [{"role": "user", "content": "baseline"}], identity, "repeat prompt" + ) + + +def test_result_partial_does_not_mutate_identical_prior_turn(): + identity = {"token": "current-turn-token"} + session = Session( + session_id="result-partial-identical-content", + messages=[ + {"role": "user", "content": "historical"}, + {"role": "assistant", "content": "same answer"}, + { + "role": "user", + "content": "current prompt", + "_active_turn_token": identity["token"], + }, + ], + context_messages=[], + ) + + appended = streaming._append_result_partial_on_error( + session, + { + "partial": True, + "messages": [ + { + "role": "user", + "content": "current prompt", + "_active_turn_token": identity["token"], + }, + {"role": "assistant", "content": "same answer"}, + ], + }, + [{"role": "user", "content": "different baseline"}], + "current prompt", + active_turn_identity=identity, + ) + + assert appended is session.messages[-1] + assert appended["content"] == "same answer" + assert appended["_partial"] is True + assert session.messages[1] == {"role": "assistant", "content": "same answer"} + + +def test_stream_partial_does_not_mutate_identical_prior_turn(monkeypatch): + identity = {"token": "current-turn-token"} + session = Session( + session_id="stream-partial-identical-content", + messages=[ + {"role": "user", "content": "historical"}, + {"role": "assistant", "content": "same answer"}, + { + "role": "user", + "content": "current prompt", + "_active_turn_token": identity["token"], + }, + ], + context_messages=[], + ) + stream_id = "stream-identical-content" + monkeypatch.setitem(streaming.STREAM_PARTIAL_TEXT, stream_id, "same answer") + monkeypatch.setitem(streaming.STREAM_REASONING_TEXT, stream_id, "") + monkeypatch.setitem(streaming.STREAM_LIVE_TOOL_CALLS, stream_id, []) + + appended = streaming._snapshot_and_append_partial_on_error( + session, + stream_id, + active_turn_identity=identity, + ) + + assert appended is session.messages[-1] + assert appended["content"] == "same answer" + assert appended["_partial"] is True + assert session.messages[1] == {"role": "assistant", "content": "same answer"} + + +def test_resolved_boundary_is_rebound_to_fresh_self_heal_agent(): + class HealedAgent: + _persist_user_message_idx = 2 + _current_turn_id = "fresh-heal-turn" + + stale_identity = { + "token": "webui-token", + "text": "current prompt", + "current_turn_user_idx": 7, + "turn_id": "failed-agent-turn", + "agent_turn_boundary_resolved": True, + "agent_turn_boundary_source": "agent", + } + + resolved = streaming._resolve_active_turn_authority( + stale_identity, + result={"messages": []}, + agent=HealedAgent(), + ) + + assert resolved["current_turn_user_idx"] == 2 + assert resolved["turn_id"] == "fresh-heal-turn" + assert resolved["agent_turn_boundary_source"] == "agent" + + +@pytest.mark.parametrize( + ("live_snapshot", "result_snapshot", "expected"), + [ + ("partial response with newer suffix", "partial response", "partial response with newer suffix"), + ("partial response", "partial response with newer suffix", "partial response with newer suffix"), + ], +) +def test_differing_live_and_result_snapshots_are_exact_once( + monkeypatch, + live_snapshot, + result_snapshot, + expected, +): + token = "current-turn-token" + stream_id = "differing-partial-sources" + session = Session( + session_id="differing-partial-sources", + messages=[ + {"role": "user", "content": "historical"}, + {"role": "assistant", "content": "historical answer"}, + { + "role": "user", + "content": "current prompt", + "_active_turn_token": token, + }, + ], + context_messages=[], + ) + result = { + "partial": True, + "turn_id": "agent-current-turn", + "current_turn_user_idx": 1, + "messages": [ + {"role": "assistant", "content": "[compacted] history"}, + {"role": "user", "content": " current\n prompt "}, + {"role": "assistant", "content": result_snapshot}, + {"role": "user", "content": "later turn"}, + {"role": "assistant", "content": "later wrong"}, + ], + } + identity = streaming._resolve_active_turn_authority( + {"token": token, "text": "current prompt"}, + result=result, + ) + monkeypatch.setitem(streaming.STREAM_PARTIAL_TEXT, stream_id, live_snapshot) + monkeypatch.setitem(streaming.STREAM_REASONING_TEXT, stream_id, "") + monkeypatch.setitem(streaming.STREAM_LIVE_TOOL_CALLS, stream_id, []) + + live_row = streaming._snapshot_and_append_partial_on_error( + session, + stream_id, + active_turn_identity=identity, + ) + result_row = streaming._append_result_partial_on_error( + session, + result, + [{"role": "user", "content": "replaced baseline"}], + "current prompt", + active_turn_identity=identity, + ) + + partials = [message for message in session.messages if message.get("_partial")] + assert result_row is live_row + assert [message["content"] for message in partials] == [expected] + assert "later wrong" not in [message.get("content") for message in session.messages] + + +def test_reconciled_compressed_projection_keeps_durable_snapshot_revision(tmp_path, monkeypatch): + sid = "revision-reconcile" + db_path = tmp_path / "state.db" + _make_state_db( + db_path, + sid, + [ + {"role": "user", "content": "durable user", "timestamp": 1.0}, + {"role": "assistant", "content": "durable answer", "timestamp": 2.0}, + {"role": "user", "content": "durable follow-up", "timestamp": 3.0}, + ], + ) + monkeypatch.setattr(models, "_active_state_db_path", lambda: db_path) + compacted_context = [ + { + "role": "assistant", + "content": "[CONTEXT COMPACTION — REFERENCE ONLY]\nCompressed task summary.", + "timestamp": 4.0, + } + ] + session = Session( + session_id=sid, + messages=[{"role": "user", "content": "full visible transcript"}], + context_messages=compacted_context, + ) + + state_snapshot = models.get_state_db_session_messages(sid, with_revision=True) + reconciled = models.reconciled_state_db_messages_for_session( + session, + prefer_context=True, + state_messages=state_snapshot, + with_revision=True, + ) + + assert reconciled.messages == compacted_context + assert len(reconciled.messages) < reconciled.revision["active_message_count"] + assert reconciled.revision == state_snapshot.revision + + +def test_webui_run_passes_revision_from_original_snapshot_when_sqlite_changes_before_agent( + tmp_path, monkeypatch +): + import api.profiles as profiles + + sid = "revision-stream" + stream_id = "stream-revision" + # Keep a conflicting default-profile DB to prove the worker uses the + # session's explicit profile rather than process-global active state. + _make_state_db( + tmp_path / "state.db", + sid, + [{"role": "user", "content": "wrong profile row", "timestamp": 1.0}], + ) + profile_home = tmp_path / "named-profile-home" + profile_home.mkdir() + db_path = profile_home / "state.db" + _make_state_db( + db_path, + sid, + [ + {"role": "user", "content": "durable user", "timestamp": 1.0}, + {"role": "assistant", "content": "durable answer", "timestamp": 2.0}, + {"role": "user", "content": "durable follow-up", "timestamp": 3.0}, + ], + ) + compacted_context = [ + { + "role": "assistant", + "content": "[CONTEXT COMPACTION — REFERENCE ONLY]\nCompressed task summary.", + "timestamp": 4.0, + } + ] + session, _event_queue = _install_streaming_session( + monkeypatch, + tmp_path, + sid=sid, + stream_id=stream_id, + messages=[{"role": "user", "content": "visible user", "timestamp": 1.0}], + context_messages=compacted_context, + ) + session.profile = "named-profile" + session.save(touch_updated_at=False) + monkeypatch.setattr(profiles, "get_hermes_home_for_profile", lambda _profile: profile_home) + monkeypatch.setattr(profiles, "get_profile_runtime_env", lambda _home: {}) + + original_reconcile = streaming.reconciled_state_db_messages_for_session + reconciliation_calls = 0 + + def reconcile_then_mutate(*args, **kwargs): + nonlocal reconciliation_calls + result = original_reconcile(*args, **kwargs) + reconciliation_calls += 1 + if reconciliation_calls == 1: + _append_state_row( + db_path, + sid, + role="assistant", + content="concurrent durable row", + timestamp=5.0, + ) + return result + + monkeypatch.setattr( + streaming, + "reconciled_state_db_messages_for_session", + reconcile_then_mutate, + ) + captured = {} + + class FakeAgent: + def __init__(self, session_id=None, **_kwargs): + self.session_id = session_id + self.context_compressor = None + self.ephemeral_system_prompt = None + self._last_error = None + + def run_conversation(self, **kwargs): + captured.update(kwargs) + return { + "completed": True, + "final_response": "ok", + "messages": [ + {"role": "user", "content": kwargs["persist_user_message"]}, + {"role": "assistant", "content": "ok"}, + ], + } + + monkeypatch.setattr(streaming, "_get_ai_agent", lambda: FakeAgent) + + streaming._run_agent_streaming( + session_id=sid, + msg_text="new webui turn", + model="test-model", + workspace=str(tmp_path), + stream_id=stream_id, + attachments=[], + ) + + assert captured["task_id"] == sid + assert captured["conversation_history_revision"] == { + "session_id": sid, + "active_message_count": 3, + "max_active_message_id": 3, + } + assert captured["conversation_history_revision"]["active_message_count"] == 3 + assert len(models.get_state_db_session_messages(sid, profile="named-profile")) == 4 + + +def test_webui_run_missing_explicit_profile_passes_no_foreign_revision( + tmp_path, + monkeypatch, +): + import api.profiles as profiles + + sid = "missing-profile-stream" + stream_id = "stream-missing-profile" + foreign_db = tmp_path / "state.db" + _make_state_db( + foreign_db, + sid, + [{"role": "user", "content": "foreign profile row", "timestamp": 1.0}], + ) + missing_home = tmp_path / "missing-profile-home" + session, event_queue = _install_streaming_session( + monkeypatch, + tmp_path, + sid=sid, + stream_id=stream_id, + messages=[{"role": "user", "content": "local projection", "timestamp": 1.0}], + context_messages=[], + ) + session.profile = "missing" + session.save(touch_updated_at=False) + monkeypatch.setattr(profiles, "get_hermes_home_for_profile", lambda _profile: missing_home) + monkeypatch.setattr(profiles, "get_profile_runtime_env", lambda _home: {}) + captured = {} + + class FakeAgent: + def __init__(self, session_id=None, **_kwargs): + self.session_id = session_id + self.context_compressor = None + self.ephemeral_system_prompt = None + self._last_error = None + + def run_conversation(self, **kwargs): + captured.update(kwargs) + return { + "completed": True, + "final_response": "ok", + "messages": [ + {"role": "user", "content": kwargs["persist_user_message"]}, + {"role": "assistant", "content": "ok"}, + ], + } + + monkeypatch.setattr(streaming, "_get_ai_agent", lambda: FakeAgent) + + streaming._run_agent_streaming( + session_id=sid, + msg_text="new webui turn", + model="test-model", + workspace=str(tmp_path), + stream_id=stream_id, + attachments=[], + ) + + assert captured["conversation_history_revision"] is None + events = _drain_events(event_queue) + assert not any(event == "apperror" for event, _payload in events) + + +@pytest.mark.parametrize("delivery", ["exception", "result"]) +def test_stale_compression_snapshot_emits_actionable_error_without_replaying_turn( + tmp_path, monkeypatch, delivery +): + sid = f"revision-stale-{delivery}" + stream_id = f"stream-revision-stale-{delivery}" + _make_state_db( + tmp_path / "state.db", + sid, + [{"role": "user", "content": "durable user", "timestamp": 1.0}], + ) + session, event_queue = _install_streaming_session( + monkeypatch, + tmp_path, + sid=sid, + stream_id=stream_id, + messages=[], + context_messages=[], + ) + calls = 0 + + class CompressionSnapshotStaleError(RuntimeError): + pass + + compression_module = types.ModuleType("agent.conversation_compression") + compression_module.__dict__["CompressionSnapshotStaleError"] = CompressionSnapshotStaleError + monkeypatch.setitem(sys.modules, "agent.conversation_compression", compression_module) + + class FakeAgent: + def __init__(self, session_id=None, stream_delta_callback=None, **_kwargs): + self.session_id = session_id + self.stream_delta_callback = stream_delta_callback + self.context_compressor = None + self.ephemeral_system_prompt = None + self._last_error = None + self._persist_user_message_idx = 0 + self._current_turn_id = f"paired-turn-{delivery}" + + def run_conversation(self, **_kwargs): + nonlocal calls + calls += 1 + if self.stream_delta_callback: + self.stream_delta_callback( + "Partial work completed before the conflict with a newer suffix." + ) + if delivery == "exception": + raise CompressionSnapshotStaleError( + "expected revision count=1 max_id=1; observed count=2 max_id=9" + ) + return { + "completed": False, + "final_response": "Partial work completed before the conflict.", + "messages": [ + {"role": "user", "content": "new webui turn"}, + { + "role": "assistant", + "content": "Partial work completed before the conflict.", + }, + ], + "error": "compression_snapshot_stale", + "partial": True, + "failed": False, + "compression_snapshot_stale": True, + } + + monkeypatch.setattr(streaming, "_get_ai_agent", lambda: FakeAgent) + + streaming._run_agent_streaming( + session_id=sid, + msg_text="new webui turn", + model="test-model", + workspace=str(tmp_path), + stream_id=stream_id, + attachments=[], + ) + + events = _drain_events(event_queue) + apperrors = [payload for event, payload in events if event == "apperror"] + assert calls == 1 + assert apperrors and apperrors[-1]["type"] == "compression_snapshot_stale" + assert "next message" in apperrors[-1]["hint"].lower() + assert "expected revision" not in apperrors[-1]["message"].lower() + assert not any(event == "done" for event, _payload in events) + + reloaded = Session.load(sid) + assert reloaded is not None + assert reloaded.active_stream_id is None + assert reloaded.pending_user_message is None + assert any( + message.get("_partial") + and message.get("content") + == "Partial work completed before the conflict with a newer suffix." + for message in reloaded.messages + ) + partial_messages = [ + message for message in reloaded.messages if message.get("_partial") + ] + assert len(partial_messages) == 1, [ + ( + message.get("role"), + message.get("content"), + message.get("_active_turn_token"), + message.get("_partial"), + ) + for message in reloaded.messages + ] + assert sum( + "Partial work completed" in str(message.get("content") or "") + for message in reloaded.messages + if message.get("role") == "assistant" + ) == 1 + assert reloaded.messages[-1]["_error"] is True + assert "next message" in reloaded.messages[-1]["content"].lower() + assert session.active_stream_id is None + + +@pytest.mark.parametrize("failure_mode", ["result", "exception"]) +@pytest.mark.parametrize( + "second_result", + [ + "recovered", + "stale_error", + "stale_flag", + "stale_without_current_assistant", + "stale_non_prefix_compacted", + "stale_non_prefix_owned_partial", + "stale_non_prefix_without_current_user", + "stale_non_prefix_repeated_prompt_without_authority", + "no_response_after_streamed_token", + ], +) +def test_auth_self_heal_refreshes_revision_after_first_agent_persists_user( + tmp_path, monkeypatch, failure_mode, second_result +): + sid = f"revision-self-heal-{failure_mode}-{second_result}" + stream_id = f"stream-revision-self-heal-{failure_mode}-{second_result}" + db_path = tmp_path / "state.db" + prior_messages = [ + {"role": "user", "content": "prior user", "timestamp": 1.0}, + {"role": "assistant", "content": "prior answer", "timestamp": 2.0}, + ] + _make_state_db(db_path, sid, prior_messages) + _session, event_queue = _install_streaming_session( + monkeypatch, + tmp_path, + sid=sid, + stream_id=stream_id, + messages=prior_messages, + context_messages=prior_messages, + ) + revisions = [] + + class AuthThenRecoverAgent: + runs = 0 + + def __init__(self, session_id=None, **_kwargs): + self.session_id = session_id + self.context_compressor = None + self.ephemeral_system_prompt = None + self._last_error = None + self.stream_delta_callback = _kwargs.get("stream_delta_callback") + + def run_conversation(self, **kwargs): + type(self).runs += 1 + revisions.append(kwargs["conversation_history_revision"]) + history = list(kwargs.get("conversation_history") or []) + if type(self).runs == 1: + _append_state_row( + db_path, + sid, + role="user", + content=kwargs["persist_user_message"], + timestamp=3.0, + ) + if failure_mode == "exception": + raise RuntimeError("401 unauthorized") + return { + "messages": history, + "error": { + "type": "authentication_error", + "status_code": 401, + "message": "token invalid", + }, + } + if second_result == "no_response_after_streamed_token": + if self.stream_delta_callback is not None: + self.stream_delta_callback("streamed but not finalized") + return {"completed": True, "messages": history} + if second_result != "recovered": + if second_result == "stale_non_prefix_owned_partial": + stale_messages = [ + {"role": "assistant", "content": "compacted history"}, + { + "role": "user", + "content": "new webui turn", + "_active_turn_token": streaming.build_active_turn_token( + stream_id, 10.0 + ), + }, + {"role": "assistant", "content": "owned partial"}, + {"role": "user", "content": "later user"}, + {"role": "assistant", "content": "later wrong"}, + ] + elif second_result == "stale_non_prefix_without_current_user": + stale_messages = [ + {"role": "user", "content": "unrelated historical user"}, + {"role": "assistant", "content": "historical answer"}, + ] + elif second_result == "stale_non_prefix_compacted": + # Compacted/replayed result that REPLACES the pre-call + # baseline instead of appending to it: historical + # assistant rows (including "prior answer") sit before the + # last current-user row and one of them lands in the + # numeric suffix messages[len(baseline):]. There is no + # assistant row for the current turn. + stale_messages = [ + { + "role": "assistant", + "content": "[compacted] summary of earlier context", + }, + {"role": "user", "content": "prior user"}, + {"role": "assistant", "content": "intermediate answer"}, + {"role": "assistant", "content": "prior answer"}, + {"role": "user", "content": "new webui turn"}, + ] + elif second_result == "stale_without_current_assistant": + stale_messages = history + [ + {"role": "user", "content": "new webui turn"} + ] + elif second_result == "stale_non_prefix_repeated_prompt_without_authority": + stale_messages = [ + {"role": "user", "content": "new webui turn"}, + {"role": "assistant", "content": "prior answer"}, + ] + else: + stale_messages = history + [ + {"role": "user", "content": "new webui turn"}, + {"role": "assistant", "content": "partial stale"}, + ] + stale_result = { + "completed": False, + "final_response": ( + "owned partial" + if second_result == "stale_non_prefix_owned_partial" + else ( + "" + if second_result + in ( + "stale_without_current_assistant", + "stale_non_prefix_compacted", + "stale_non_prefix_without_current_user", + "stale_non_prefix_repeated_prompt_without_authority", + ) + else "partial stale" + ) + ), + "messages": stale_messages, + "partial": True, + "failed": False, + } + if second_result == "stale_error": + stale_result["error"] = "compression_snapshot_stale" + elif second_result == "stale_flag": + stale_result["compression_snapshot_stale"] = True + else: + stale_result["error"] = "compression_snapshot_stale" + stale_result["compression_snapshot_stale"] = True + return stale_result + return { + "completed": True, + "final_response": "recovered", + "messages": history + [ + {"role": "user", "content": "new webui turn"}, + {"role": "assistant", "content": "recovered"}, + ], + } + + monkeypatch.setattr(streaming, "_get_ai_agent", lambda: AuthThenRecoverAgent) + monkeypatch.setattr( + streaming, + "_attempt_credential_self_heal", + lambda *_args, **_kwargs: { + "api_key": "refreshed-key", + "provider": "test-provider", + "base_url": None, + "credential_pool": None, + }, + ) + + streaming._run_agent_streaming( + session_id=sid, + msg_text="new webui turn", + model="test-model", + workspace=str(tmp_path), + stream_id=stream_id, + attachments=[], + ) + + assert revisions == [ + { + "session_id": sid, + "active_message_count": 2, + "max_active_message_id": 2, + }, + { + "session_id": sid, + "active_message_count": 3, + "max_active_message_id": 3, + }, + ] + events = _drain_events(event_queue) + apperrors = [payload for event, payload in events if event == "apperror"] + reloaded = Session.load(sid) + assert reloaded is not None + assert AuthThenRecoverAgent.runs == 2 + if second_result == "recovered": + assert not apperrors + assert any(message.get("content") == "recovered" for message in reloaded.messages) + elif second_result == "no_response_after_streamed_token": + assert len(apperrors) == 1 + assert apperrors[0]["type"] == "auth_mismatch" + assert not any(event == "done" for event, _payload in events) + assert reloaded.active_stream_id is None + assert reloaded.pending_user_message is None + else: + assert len(apperrors) == 1 + assert apperrors[0]["type"] == "compression_snapshot_stale" + assert "next message" in apperrors[0]["hint"].lower() + assert not any(event == "done" for event, _payload in events) + assert reloaded.active_stream_id is None + assert reloaded.pending_user_message is None + assert reloaded.pending_attachments == [] + partials = [ + message + for message in reloaded.messages + if message.get("_partial") is True + ] + if second_result in ( + "stale_without_current_assistant", + "stale_non_prefix_compacted", + "stale_non_prefix_without_current_user", + "stale_non_prefix_repeated_prompt_without_authority", + ): + assert not partials + assert not any( + message.get("content") == "prior answer" and message.get("_partial") is True + for message in reloaded.messages + ) + else: + expected_partial = ( + "owned partial" + if second_result == "stale_non_prefix_owned_partial" + else "partial stale" + ) + assert [message.get("content") for message in partials] == [expected_partial] + if second_result == "stale_non_prefix_owned_partial": + assert not any( + message.get("content") == "later wrong" + for message in reloaded.messages + ) + assert reloaded.messages[-1]["_error"] is True + assert "next message" in reloaded.messages[-1]["content"].lower() + + +def _repeated_prompt_collision_result_messages(prompt, *, current_answer): + """Non-prefix Agent projection where BOTH candidate positions are user rows + with the same normalized text. + + Against the 2-row heal baseline (prior user + prior assistant) the legacy + shifted probe ``current_turn_user_idx - len(previous_context)`` = 3 - 2 = 1 + lands on the HISTORICAL same-text user row, which is followed only by + historical assistant prose. The declared Agent index 3 addresses + ``result["messages"]`` directly and lands on the CURRENT same-text user + row, followed only by current output (when any exists). + """ + rows = [ + {"role": "assistant", "content": "[compacted] summary of earlier context"}, + {"role": "user", "content": prompt}, # shifted probe target (historical) + {"role": "assistant", "content": "historical answer"}, # historical-only prose + {"role": "user", "content": prompt}, # exact Agent index target (current) + ] + if current_answer is not None: + rows.append({"role": "assistant", "content": current_answer}) + return rows + + +@pytest.mark.parametrize("terminal", ["returned", "raised"]) +@pytest.mark.parametrize( + "collision", + [ + "current_output_after_exact_row", + "historical_output_after_shifted_row_only", + "historical_only_stale_partial", + ], +) +def test_self_heal_repeated_prompt_never_accepts_shifted_historical_row( + tmp_path, monkeypatch, terminal, collision +): + """Required regression (2026-08-13 review): one declared index domain. + + A repeated prompt makes both legacy candidate positions (shifted + ``current_turn_user_idx - len(previous_context)`` and exact + ``current_turn_user_idx``) same-text user rows in the healed non-prefix + projection. Historical assistant output follows only the shifted row; + current output follows only the exact row. Composed through BOTH + production self-heal entries — returned auth-failure result and raised + 401 — the historical row must never be accepted by + ``_self_heal_result_succeeded`` nor persisted by + ``_append_result_partial_on_error``; settlement/done/apperror must fire + exactly once. + """ + prompt = "new webui turn" + sid = f"repeat-collision-{terminal}-{collision}" + stream_id = f"stream-{sid}" + db_path = tmp_path / "state.db" + prior_messages = [ + {"role": "user", "content": prompt, "timestamp": 1.0}, + {"role": "assistant", "content": "historical answer", "timestamp": 2.0}, + ] + _make_state_db(db_path, sid, prior_messages) + session, event_queue = _install_streaming_session( + monkeypatch, + tmp_path, + sid=sid, + stream_id=stream_id, + messages=prior_messages, + context_messages=prior_messages, + ) + session.pending_user_message = prompt + + class RepeatedPromptAgent: + runs = 0 + + def __init__(self, session_id=None, **_kwargs): + self.session_id = session_id + self.context_compressor = None + self.ephemeral_system_prompt = None + self._last_error = None + self.stream_delta_callback = _kwargs.get("stream_delta_callback") + + def run_conversation(self, **kwargs): + type(self).runs += 1 + history = list(kwargs.get("conversation_history") or []) + if type(self).runs == 1: + _append_state_row( + db_path, + sid, + role="user", + content=kwargs["persist_user_message"], + timestamp=3.0, + ) + if terminal == "raised": + raise RuntimeError("401 unauthorized") + return { + "messages": history, + "error": { + "type": "authentication_error", + "status_code": 401, + "message": "token invalid", + }, + } + # Heal retry: the 2-row refreshed baseline makes the legacy + # shifted probe (3 - 2 = 1) collide with the historical user row. + heal = { + "turn_id": "agent-heal-turn", + "current_turn_user_idx": 3, + "messages": _repeated_prompt_collision_result_messages( + prompt, + current_answer=( + "healed current answer" + if collision == "current_output_after_exact_row" + else None + ), + ), + } + if collision == "current_output_after_exact_row": + heal.update( + {"completed": True, "final_response": "healed current answer"} + ) + elif collision == "historical_output_after_shifted_row_only": + # Silent success shape: if the shifted historical row were + # accepted, "historical answer" would satisfy the retry. + heal.update({"completed": True, "final_response": ""}) + else: + # Stale partial shape: if the shifted historical row were + # accepted, "historical answer" would persist as the current + # turn's partial. + heal.update( + { + "completed": False, + "partial": True, + "failed": False, + "final_response": "", + "error": "compression_snapshot_stale", + "compression_snapshot_stale": True, + } + ) + return heal + + monkeypatch.setattr(streaming, "_get_ai_agent", lambda: RepeatedPromptAgent) + monkeypatch.setattr( + streaming, + "_attempt_credential_self_heal", + lambda *_args, **_kwargs: { + "api_key": "refreshed-key", + "provider": "test-provider", + "base_url": None, + "credential_pool": None, + }, + ) + + streaming._run_agent_streaming( + session_id=sid, + msg_text=prompt, + model="test-model", + workspace=str(tmp_path), + stream_id=stream_id, + attachments=[], + ) + + assert RepeatedPromptAgent.runs == 2 + events = _drain_events(event_queue) + done_events = [payload for event, payload in events if event == "done"] + apperrors = [payload for event, payload in events if event == "apperror"] + reloaded = Session.load(sid) + assert reloaded is not None + + # Exact-once turn settlement in every arm: the pending turn is consumed. + assert reloaded.active_stream_id is None + assert reloaded.pending_user_message is None + assert reloaded.pending_attachments == [] + + current_turn_assistants = [ + str(message.get("content") or "") + for message in reloaded.messages[2:] + if isinstance(message, dict) and message.get("role") == "assistant" + ] + if collision == "current_output_after_exact_row": + # Success arm: exactly one done, no apperror, and the settled current + # turn carries only the exact-row answer — persisted exactly once. + assert len(done_events) == 1 + assert not apperrors + assert current_turn_assistants == ["healed current answer"] + done_messages = done_events[0]["session"]["messages"] + assert [ + (message["role"], message.get("content")) for message in done_messages + ] == [ + ("user", prompt), + ("assistant", "historical answer"), + ("user", prompt), + ("assistant", "healed current answer"), + ] + assert not any(message.get("_error") for message in reloaded.messages) + else: + # Failure arms: the shifted historical row must never satisfy the + # retry — exactly one apperror, never a done. + assert not done_events + assert len(apperrors) == 1 + if collision == "historical_output_after_shifted_row_only": + assert apperrors[0]["type"] == "auth_mismatch" + else: + assert apperrors[0]["type"] == "compression_snapshot_stale" + # Historical prose is never accepted as this turn's output nor + # persisted as its partial. + assert not any( + message.get("_partial") for message in reloaded.messages + ) + assert "historical answer" not in current_turn_assistants + assert reloaded.messages[-1]["_error"] is True + # The historical assistant row itself survives untouched exactly once, at + # its original position in the transcript. + assert [ + str(message.get("content") or "") + for message in reloaded.messages + if isinstance(message, dict) and message.get("role") == "assistant" + ].count("historical answer") == 1 diff --git a/tests/test_issue5121_provider_auth_terminal_error.py b/tests/test_issue5121_provider_auth_terminal_error.py index 2cdaf854..19c9c817 100644 --- a/tests/test_issue5121_provider_auth_terminal_error.py +++ b/tests/test_issue5121_provider_auth_terminal_error.py @@ -285,11 +285,19 @@ def test_auth_401_classification_receives_stringified_probe_text(tmp_path, monke observed = {} real_classify = streaming._classify_provider_error - def _spy_classify_provider_error(err_str, exc=None, *, silent_failure=False): + def _spy_classify_provider_error( + err_str, exc=None, *, silent_failure=False, result=None + ): observed["err_str"] = err_str observed["exc"] = exc observed["silent_failure"] = silent_failure - return real_classify(err_str, exc, silent_failure=silent_failure) + observed["result"] = result + return real_classify( + err_str, + exc, + silent_failure=silent_failure, + result=result, + ) with mock.patch.object(streaming, "_classify_provider_error", side_effect=_spy_classify_provider_error): _run_stream(monkeypatch, session, "stream_auth_probe_text", agent_cls, workspace=str(tmp_path)) @@ -415,6 +423,77 @@ def test_auth_retry_success_does_not_append_error_turn(tmp_path, monkeypatch): assert not any(msg.get("_error") for msg in saved.messages) +def test_auth_exception_retry_structured_failure_still_emits_error( + tmp_path, + monkeypatch, +): + session = _prepare_session( + "auth_retry_structured_failure", + "stream_auth_retry_structured_failure", + pending_user_message="Please retry", + ) + + class ExceptionThenStructuredFailureAgent(MockAgent): + runs = 0 + + def run_conversation(self, **kwargs): + type(self).runs += 1 + if type(self).runs == 1: + raise RuntimeError("401 unauthorized") + if self.stream_delta_callback is not None: + self.stream_delta_callback("partial retry output") + return { + "error": { + "type": "authentication_error", + "status_code": 401, + "message": "retry returned a structured failure", + }, + "messages": list(kwargs.get("conversation_history") or []), + } + + fake_queue = queue.Queue() + streaming.STREAMS["stream_auth_retry_structured_failure"] = fake_queue + config.STREAM_PARTIAL_TEXT["stream_auth_retry_structured_failure"] = "" + heal_rt = { + "provider": "test-provider", + "api_key": "fresh-key", + "base_url": None, + } + + with mock.patch.object(streaming, "get_session", return_value=session), \ + mock.patch.object( + streaming, + "_get_ai_agent", + return_value=ExceptionThenStructuredFailureAgent, + ), \ + mock.patch.object( + streaming, + "resolve_model_provider", + return_value=("test-model", "test-provider", None), + ), \ + mock.patch("api.config.get_config", return_value={}), \ + mock.patch("api.config._resolve_cli_toolsets", return_value=[]), \ + mock.patch.object( + streaming, + "_attempt_credential_self_heal", + return_value=heal_rt, + ): + streaming._run_agent_streaming( + session_id=session.session_id, + msg_text=session.pending_user_message, + model="test-model", + workspace=str(tmp_path), + stream_id="stream_auth_retry_structured_failure", + ) + + saved = Session.load("auth_retry_structured_failure") + assert saved is not None + events = _queue_events(fake_queue) + assert any(event == "apperror" for event, _ in events) + assert not any(event == "done" for event, _ in events) + assert saved.messages[-1]["_error"] is True + + def test_success_repeated_assistant_text_stays_successful_current_turn(tmp_path, monkeypatch): session = _prepare_session("repeat_success", "stream_repeat_success", pending_user_message="Please say it again") _seed_prior_turn( diff --git a/tests/test_state_db_read_backstop.py b/tests/test_state_db_read_backstop.py index f0dbb49d..72c88ac2 100644 --- a/tests/test_state_db_read_backstop.py +++ b/tests/test_state_db_read_backstop.py @@ -117,3 +117,49 @@ def test_limit_composes_with_since_timestamp(tmp_path, monkeypatch): msgs = get_state_db_session_messages("s1", since_timestamp=50.0, limit=10) assert len(msgs) == 10 assert [m["content"] for m in msgs] == [f"msg{i}" for i in range(90, 100)] + + +def test_reader_uses_durable_id_order_when_timestamps_are_non_monotonic( + tmp_path, + monkeypatch, +): + db = tmp_path / "state.db" + conn = sqlite3.connect(str(db)) + conn.execute("CREATE TABLE sessions (id TEXT PRIMARY KEY)") + conn.execute( + """ + CREATE TABLE messages ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + session_id TEXT NOT NULL, + role TEXT, + content TEXT, + timestamp REAL, + active INTEGER DEFAULT 1 + ) + """ + ) + conn.execute("INSERT INTO sessions (id) VALUES ('s1')") + conn.executemany( + "INSERT INTO messages (session_id, role, content, timestamp) VALUES (?, ?, ?, ?)", + [ + ("s1", "user", "id-1 timestamp-300", 300.0), + ("s1", "assistant", "id-2 timestamp-100", 100.0), + ("s1", "user", "id-3 timestamp-200", 200.0), + ], + ) + conn.commit() + conn.close() + monkeypatch.setattr(models, "_active_state_db_path", lambda: db) + + full = get_state_db_session_messages("s1") + limited = get_state_db_session_messages("s1", limit=2) + + assert [message["content"] for message in full] == [ + "id-1 timestamp-300", + "id-2 timestamp-100", + "id-3 timestamp-200", + ] + assert [message["content"] for message in limited] == [ + "id-2 timestamp-100", + "id-3 timestamp-200", + ]