diff --git a/CHANGELOG.md b/CHANGELOG.md index fa4f3ef67a..ea5dab2278 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,11 @@ ## [Unreleased] +### Fixed + +- Streaming finalization now treats compression-exhausted or tool-tail agent results as errors instead of completed turns, so long tool-heavy sessions do not appear done when Hermes Agent failed before writing a final assistant answer. When Hermes Agent rotated the session id during automatic compression before that terminal failure, WebUI now preserves the pre-compression snapshot, migrates continuation state first, and persists the final error on the continuation session instead of the stale parent row. +- Completed transcripts no longer render internal `[CONTEXT COMPACTION — REFERENCE ONLY]` reference cards; compression-exhausted runs now surface as explicit errors instead. + ## [v0.51.267] — 2026-06-04 — Release II (stage-r17 — TTS + CSRF forwarded-header security hardening) ### Security diff --git a/api/streaming.py b/api/streaming.py index d7550b9368..4bc63276a7 100644 --- a/api/streaming.py +++ b/api/streaming.py @@ -807,6 +807,12 @@ def _classify_provider_error(err_str: str, exc=None, *, silent_failure: bool = F _is_rate_limit = (not _is_quota) and ( 'rate limit' in _err_lower or '429' in err_str or (exc is not None and 'RateLimitError' in _exc_name) ) + _is_compression_exhausted = ( + 'compression_exhausted' in _err_lower + or 'compression exhausted' in _err_lower + or ('context length exceeded' in _err_lower and 'cannot compress further' in _err_lower) + or ('context compression' in _err_lower and 'max compression attempts' in _err_lower) + ) if _is_quota: return { 'label': 'Out of credits', @@ -831,6 +837,12 @@ def _classify_provider_error(err_str: str, exc=None, *, silent_failure: bool = F 'type': 'model_not_found', 'hint': 'The selected model was not found by the provider. Check the model ID in Settings or run `hermes model` to verify it exists for your provider.', } + if _is_compression_exhausted: + return { + 'label': 'Context compression exhausted', + 'type': 'compression_exhausted', + 'hint': 'The conversation context is too large to compress safely. Start a new conversation or retry with a narrower task.', + } if silent_failure: return { 'label': 'No response from provider', @@ -3610,6 +3622,52 @@ def _assistant_reply_added_after_current_turn(result_messages, previous_context, ) +def _session_lacks_final_assistant_answer(messages) -> bool: + """Return True when the persisted transcript ends before a final answer.""" + for msg in reversed(list(messages or [])): + if not isinstance(msg, dict): + continue + if msg.get('_error'): + return False + if _is_context_compression_marker(msg): + continue + role = msg.get('role') + if role == 'tool': + return True + if role == 'assistant': + content = msg.get('content') + if isinstance(content, list): + text = '\n'.join( + str(part.get('text') or part.get('content') or '') + for part in content + if isinstance(part, dict) + ) + else: + text = str(content or '') + if msg.get('tool_calls'): + return True + if text.strip(): + return False + continue + if role == 'user': + return True + return True + + +def _agent_result_terminal_failure(result) -> bool: + """Return True for agent results that must not be finalized as done.""" + if not isinstance(result, dict): + return False + status = str(result.get('status') or result.get('state') or '').strip().lower() + if status in {'failed', 'error', 'partial', 'compression_exhausted'}: + return True + if result.get('compression_exhausted'): + return True + if result.get('failed') or result.get('partial'): + return True + return False + + _TOOL_RESULT_SNIPPET_MAX = 4000 @@ -5877,6 +5935,128 @@ def _periodic_checkpoint(): if isinstance(_part, dict) and isinstance(_part.get('text'), str): _part['text'] = _strip_xml_tool_calls(_part['text']) + # ── Handle context compression side effects ── + # If compression fired inside run_conversation, the agent may have + # rotated its session_id. Detect and fix the mismatch before any + # terminal-failure return so snapshot preservation, continuation + # registration, and subsequent error persistence all target the + # continuation session instead of the stale parent. + # + # Lock migration: when session_id rotates, we alias the new ID to + # the *same* Lock object under SESSION_AGENT_LOCKS so that + # subsequent callers using _get_session_agent_lock(new_sid) get the + # same Lock the streaming thread is already holding. We then pop + # the old-id entry to prevent a leak. This is safe because we + # already hold _agent_lock (the Lock object itself), so the + # reference stays alive even after the dict entry is removed. + # Concurrent readers that already looked up the old ID will still + # see the same Lock object until they release it. + _compression_origin_session_id = session_id + _compression_continuation_session_id = None + _agent_sid = getattr(agent, 'session_id', None) + _compressed = False + if _agent_sid and _agent_sid != session_id: + old_sid = session_id + new_sid = _agent_sid + _compression_origin_session_id = old_sid + _compression_continuation_session_id = new_sid + s.session_id = new_sid + # Carry profile identity across the compression boundary. + # Without this, s.profile stays None on the continuation + # session. On the next request, _run_agent_streaming calls + # get_hermes_home_for_profile(getattr(s, 'profile', None)) + # which falls back to the default profile's HERMES_HOME. + # Memory writes then land in the wrong profile's MEMORY.md. + # Stamping here also ensures s.save() persists a non-null + # profile field to the continuation session's JSON file, + # covering the case where the session is later evicted from + # SESSIONS and reconstructed from disk via Session.load(). + if not s.profile and _resolved_profile_name: + s.profile = _resolved_profile_name + logger.info( + "Stamped profile=%r on continuation session %s after compression", + _resolved_profile_name, new_sid, + ) + # Preserve the original session file so the full pre-compression + # history survives even when summarisation fails. The previous + # implementation renamed old_sid.json → new_sid.json, which + # destroyed the only persistent copy of the uncompressed history + # before the new (possibly summary-only) session had been saved. + # If the LLM summariser also failed, the user was left with zero + # recoverable messages. (#2223) + # --- + # Archive the old session: write its current state to disk so + # the full conversation history survives even when context + # compression removes messages from the model's context. Skip + # the write when the file already contains up-to-date data + # (i.e. it was just saved by a checkpoint). + _preserve_pre_compression_snapshot(s, old_sid) + # The continuation is the live/tip session, not another archived + # snapshot. If the in-memory object was itself loaded from a + # pre-compression snapshot (possible on repeated compression chains + # or stale-cache repair paths), _preserve_pre_compression_snapshot() + # intentionally restores that old flag; clear it before saving the + # new continuation so sidebar/discoverability code does not hide the + # session that owns the completed turn. + s.pre_compression_snapshot = False + # Always link the continuation session to its immediate predecessor + # (the preserved snapshot). This OVERRIDES any prior + # parent_session_id because the new continuation IS the next link + # in the chain: traversal walks new → old → old.parent → ... root. + # Stage-353 Opus SHOULD-FIX: previous `if not s.parent_session_id` + # guard skipped this stamp on fork-of-fork compressions, so a + # subsequent traversal from the new continuation would jump + # over the just-preserved snapshot back to the original fork + # parent, losing access to the recoverable history in old_sid.json. + s.parent_session_id = old_sid + with LOCK: + cached_old_session = SESSIONS.pop(old_sid, None) + if cached_old_session is not None and cached_old_session is not s: + cached_old_sid = str(getattr(cached_old_session, 'session_id', '') or '') + if cached_old_sid == str(old_sid): + SESSIONS[old_sid] = cached_old_session + else: + logger.warning( + "compression cache migration skipped stale object: old_sid=%s new_sid=%s cached_session_id=%s", + old_sid, + new_sid, + cached_old_sid or None, + ) + SESSIONS[new_sid] = s + SESSIONS.move_to_end(new_sid) + while len(SESSIONS) > SESSIONS_MAX: + SESSIONS.popitem(last=False) + # Migrate the per-session lock: alias new_sid to the held + # _agent_lock reference directly (not via old_sid lookup), + # then remove the old_sid entry to prevent a leak. + with SESSION_AGENT_LOCKS_LOCK: + SESSION_AGENT_LOCKS[new_sid] = _agent_lock + SESSION_AGENT_LOCKS.pop(old_sid, None) + # Migrate cached agent to the new session ID so the turn + # count survives context compression. + from api.config import SESSION_AGENT_CACHE, SESSION_AGENT_CACHE_LOCK + _skipped_agent_migration_entry = None + with SESSION_AGENT_CACHE_LOCK: + _cached_entry = SESSION_AGENT_CACHE.pop(old_sid, None) + if _cached_entry: + _cached_agent = _cached_entry[0] + if _cached_agent_matches_session(_cached_agent, new_sid): + SESSION_AGENT_CACHE[new_sid] = _cached_entry + else: + _skipped_agent_migration_entry = _cached_entry + logger.warning( + '[webui] Skipped cached agent migration with mismatched session identity: old_sid=%s new_sid=%s agent_session_id=%s', + old_sid, + new_sid, + _cached_agent_session_identity(_cached_agent), + ) + if _skipped_agent_migration_entry is not None: + try: + _close_cached_agent_entry_at_session_boundary(old_sid, _skipped_agent_migration_entry) + except Exception: + logger.debug("Failed to close skipped compression-migration cached agent for session %s", old_sid, exc_info=True) + _compressed = True + # ── Detect silent agent failure (no assistant reply produced) ── # When the agent catches an auth/network error internally it may return # an empty final_response without raising — the stream would end with @@ -5895,8 +6075,17 @@ def _periodic_checkpoint(): _previous_context_messages, msg_text, ) + _terminal_failure = ( + _agent_result_terminal_failure(result) + or ( + not _token_sent + and _session_lacks_final_assistant_answer(_all_result_messages) + ) + ) + if _terminal_failure: + _assistant_added = False # _token_sent tracks whether on_token() was called (any streamed text) - if not _assistant_added and not _token_sent: + if _terminal_failure or (not _assistant_added and not _token_sent): if cancel_event.is_set(): _finalize_cancelled_turn(s, ephemeral=ephemeral) if not ephemeral: @@ -6046,7 +6235,6 @@ def _periodic_checkpoint(): _err_type, _err_hint, ) - put('apperror', _error_payload) # Clear stream/pending state so the session does not appear # "agent_running" on reload after a silent failure. # Persist the error so it survives page reload. @@ -6074,130 +6262,21 @@ def _periodic_checkpoint(): s.save() except Exception: pass + _error_payload['session'] = redact_session_data( + s.compact() | {'messages': s.messages, 'tool_calls': s.tool_calls} + ) + _error_payload['session_id'] = s.session_id + _error_payload['old_session_id'] = _compression_origin_session_id + if _compression_continuation_session_id is not None: + _error_payload['new_session_id'] = _compression_continuation_session_id + _error_payload['continuation_session_id'] = _compression_continuation_session_id + put('apperror', _error_payload) # Legacy #373 source tests and clients look for the # no_response type; #1765 keeps that type but improves # the catch-all label, hint, and provider details. return # apperror already closes the stream on the client side # ── Handle context compression side effects ── - # If compression fired inside run_conversation, the agent may have - # rotated its session_id. Detect and fix the mismatch so the WebUI - # continues writing to the correct session file. - # - # Lock migration: when session_id rotates, we alias the new ID to - # the *same* Lock object under SESSION_AGENT_LOCKS so that - # subsequent callers using _get_session_agent_lock(new_sid) get the - # same Lock the streaming thread is already holding. We then pop - # the old-id entry to prevent a leak. This is safe because we - # already hold _agent_lock (the Lock object itself), so the - # reference stays alive even after the dict entry is removed. - # Concurrent readers that already looked up the old ID will still - # see the same Lock object until they release it. - _compression_origin_session_id = session_id - _compression_continuation_session_id = None - _agent_sid = getattr(agent, 'session_id', None) - _compressed = False - if _agent_sid and _agent_sid != session_id: - old_sid = session_id - new_sid = _agent_sid - _compression_origin_session_id = old_sid - _compression_continuation_session_id = new_sid - s.session_id = new_sid - # Carry profile identity across the compression boundary. - # Without this, s.profile stays None on the continuation - # session. On the next request, _run_agent_streaming calls - # get_hermes_home_for_profile(getattr(s, 'profile', None)) - # which falls back to the default profile's HERMES_HOME. - # Memory writes then land in the wrong profile's MEMORY.md. - # Stamping here also ensures s.save() persists a non-null - # profile field to the continuation session's JSON file, - # covering the case where the session is later evicted from - # SESSIONS and reconstructed from disk via Session.load(). - if not s.profile and _resolved_profile_name: - s.profile = _resolved_profile_name - logger.info( - "Stamped profile=%r on continuation session %s after compression", - _resolved_profile_name, new_sid, - ) - # Preserve the original session file so the full pre-compression - # history survives even when summarisation fails. The previous - # implementation renamed old_sid.json → new_sid.json, which - # destroyed the only persistent copy of the uncompressed history - # before the new (possibly summary-only) session had been saved. - # If the LLM summariser also failed, the user was left with zero - # recoverable messages. (#2223) - # --- - # Archive the old session: write its current state to disk so - # the full conversation history survives even when context - # compression removes messages from the model's context. Skip - # the write when the file already contains up-to-date data - # (i.e. it was just saved by a checkpoint). - _preserve_pre_compression_snapshot(s, old_sid) - # The continuation is the live/tip session, not another archived - # snapshot. If the in-memory object was itself loaded from a - # pre-compression snapshot (possible on repeated compression chains - # or stale-cache repair paths), _preserve_pre_compression_snapshot() - # intentionally restores that old flag; clear it before saving the - # new continuation so sidebar/discoverability code does not hide the - # session that owns the completed turn. - s.pre_compression_snapshot = False - # Always link the continuation session to its immediate predecessor - # (the preserved snapshot). This OVERRIDES any prior - # parent_session_id because the new continuation IS the next link - # in the chain: traversal walks new → old → old.parent → ... root. - # Stage-353 Opus SHOULD-FIX: previous `if not s.parent_session_id` - # guard skipped this stamp on fork-of-fork compressions, so a - # subsequent traversal from the new continuation would jump - # over the just-preserved snapshot back to the original fork - # parent, losing access to the recoverable history in old_sid.json. - s.parent_session_id = old_sid - with LOCK: - cached_old_session = SESSIONS.pop(old_sid, None) - if cached_old_session is not None and cached_old_session is not s: - cached_old_sid = str(getattr(cached_old_session, 'session_id', '') or '') - if cached_old_sid == str(old_sid): - SESSIONS[old_sid] = cached_old_session - else: - logger.warning( - "compression cache migration skipped stale object: old_sid=%s new_sid=%s cached_session_id=%s", - old_sid, - new_sid, - cached_old_sid or None, - ) - SESSIONS[new_sid] = s - SESSIONS.move_to_end(new_sid) - while len(SESSIONS) > SESSIONS_MAX: - SESSIONS.popitem(last=False) - # Migrate the per-session lock: alias new_sid to the held - # _agent_lock reference directly (not via old_sid lookup), - # then remove the old_sid entry to prevent a leak. - with SESSION_AGENT_LOCKS_LOCK: - SESSION_AGENT_LOCKS[new_sid] = _agent_lock - SESSION_AGENT_LOCKS.pop(old_sid, None) - # Migrate cached agent to the new session ID so the turn - # count survives context compression. - from api.config import SESSION_AGENT_CACHE, SESSION_AGENT_CACHE_LOCK - _skipped_agent_migration_entry = None - with SESSION_AGENT_CACHE_LOCK: - _cached_entry = SESSION_AGENT_CACHE.pop(old_sid, None) - if _cached_entry: - _cached_agent = _cached_entry[0] - if _cached_agent_matches_session(_cached_agent, new_sid): - SESSION_AGENT_CACHE[new_sid] = _cached_entry - else: - _skipped_agent_migration_entry = _cached_entry - logger.warning( - '[webui] Skipped cached agent migration with mismatched session identity: old_sid=%s new_sid=%s agent_session_id=%s', - old_sid, - new_sid, - _cached_agent_session_identity(_cached_agent), - ) - if _skipped_agent_migration_entry is not None: - try: - _close_cached_agent_entry_at_session_boundary(old_sid, _skipped_agent_migration_entry) - except Exception: - logger.debug("Failed to close skipped compression-migration cached agent for session %s", old_sid, exc_info=True) - _compressed = True # Also detect compression via the result dict or compressor state if not _compressed: _compressor = getattr(agent, 'context_compressor', None) @@ -7181,6 +7260,8 @@ def _periodic_checkpoint(): ) except Exception: logger.debug("Failed to append interrupted turn journal event", exc_info=True) + _error_payload['session_id'] = getattr(s, 'session_id', session_id) + _error_payload['old_session_id'] = session_id put('apperror', _error_payload) finally: # Stop the periodic checkpoint thread before the final recovery path. diff --git a/static/messages.js b/static/messages.js index 0f0001fdf3..133a8e530e 100644 --- a/static/messages.js +++ b/static/messages.js @@ -2442,12 +2442,17 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){ _clearOwnerInflightState(); _clearApprovalForOwner(); _clearClarifyForOwner('terminal'); - if(S.session&&S.session.session_id===activeSid){ + let d={}; + try{ d=JSON.parse(e.data||'{}')||{}; }catch(_){ d={}; } + const currentSid=S.session&&S.session.session_id; + const eventSid=d.old_session_id||d.session_id||''; + const continuationSid=(d.session&&d.session.session_id)||d.new_session_id||d.continuation_session_id||''; + const eventMatchesCurrent=!!(currentSid&&(eventSid===currentSid||continuationSid===currentSid)); + if(S.session&&eventMatchesCurrent){ S.activeStreamId=null; clearLiveToolCards();if(!assistantText)removeThinking(); let isRecoveryControlMessage=false; try{ - const d=JSON.parse(e.data); const isRateLimit=d.type==='rate_limit'; const isQuotaExhausted=d.type==='quota_exhausted'; const isAuthMismatch=d.type==='auth_mismatch'; @@ -2455,14 +2460,24 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){ const isModelNotFound=d.type==='model_not_found'; const isCancelled=d.type==='cancelled'; const isInterrupted=d.type==='interrupted'; + const isCompressionExhausted=d.type==='compression_exhausted'; isRecoveryControlMessage=isInterrupted && (d.recovery_control===true || _streamRecoveryControlMessageText(d.message)); const isNoResponse=d.type==='no_response'||d.type==='silent_failure'; - const label=isCancelled?'Task cancelled':isInterrupted?'Response interrupted':isQuotaExhausted?'Out of credits':isRateLimit?'Rate limit reached':isGatewayAuthError?(typeof t==='function'?t('gateway_auth_label'):'Gateway authentication failed'):isAuthMismatch?(typeof t==='function'?t('provider_mismatch_label'):'Provider mismatch'):isModelNotFound?(typeof t==='function'?t('model_not_found_label'):'Model not found'):isNoResponse?'No response from provider':'Error'; + const label=isCancelled?'Task cancelled':isInterrupted?'Response interrupted':isCompressionExhausted?'Context compression exhausted':isQuotaExhausted?'Out of credits':isRateLimit?'Rate limit reached':isGatewayAuthError?(typeof t==='function'?t('gateway_auth_label'):'Gateway authentication failed'):isAuthMismatch?(typeof t==='function'?t('provider_mismatch_label'):'Provider mismatch'):isModelNotFound?(typeof t==='function'?t('model_not_found_label'):'Model not found'):isNoResponse?'No response from provider':'Error'; const hint=d.hint?`\n\n*${d.hint}*`:''; const details=d.details?String(d.details).replace(/```/g,'`\u200b``'):''; const detailsLabel=isCancelled?'Cancellation details':isInterrupted?'Interruption details':undefined; + window._compressionUi=null; + if(typeof clearCompressionUi==='function') clearCompressionUi(); if(isRecoveryControlMessage){ if(typeof showToast==='function') showToast('Stream recovery signal received. Restoring transcript...',3500,'error'); + } else if(d.session&&typeof d.session==='object'){ + S.session=d.session; + S.messages=_carryForwardEphemeralTurnFields(S.messages||[], d.session.messages||[]); + if(S.session&&S.session.session_id){ + try{localStorage.setItem('hermes-webui-session',S.session.session_id);}catch(_){} + if(typeof _setActiveSessionUrl==='function') _setActiveSessionUrl(S.session.session_id); + } } else { S.messages.push({role:'assistant',content:`**${label}:** ${d.message}${hint}`,provider_details:details,provider_details_label:detailsLabel}); } @@ -2479,13 +2494,12 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){ } })(); } else { - _markSessionViewed(activeSid, S.messages.length); + _markSessionViewed((S.session&&S.session.session_id)||activeSid, S.messages.length); renderMessages({preserveScroll:true}); } }else if(typeof trackBackgroundError==='function'){ const _errTitle=(typeof _allSessions!=='undefined'&&_allSessions.find(s=>s.session_id===activeSid)||{}).title||null; - try{const d=JSON.parse(e.data);trackBackgroundError(activeSid,_errTitle,d.message||'Error');} - catch(_){trackBackgroundError(activeSid,_errTitle,'Error');} + trackBackgroundError(activeSid,_errTitle,d.message||'Error'); } _setActivePaneIdleIfOwner(); renderSessionList(); // clear streaming indicator immediately on apperror diff --git a/static/ui.js b/static/ui.js index 33280250e8..c1f0c136d8 100644 --- a/static/ui.js +++ b/static/ui.js @@ -6359,7 +6359,10 @@ function _collectHandoffSummaryStates(messages){ function _isContextCompactionMessage(m){ if(!m||!m.role||m.role==='tool') return false; const text=msgContent(m)||String(m.content||''); - return /^\s*\[context compaction/i.test(text) || /^\s*context compaction/i.test(text); + return _isContextCompactionText(text); +} +function _isContextCompactionText(text){ + return /^\s*\[context compaction/i.test(String(text||'')) || /^\s*context compaction/i.test(String(text||'')); } function _isPreservedCompressionTaskListMarkerText(text){ return /^\s*\[your active task list was preserved across context compression\]/i.test(String(text||'')); @@ -6437,6 +6440,9 @@ function _latestCompressionReferenceMessage(messages, summaryText=''){ } return {message:null, rawIdx:-1}; } +function _shouldShowSettledCompressionReference(referenceText){ + return !!String(referenceText||'').trim() && !_isContextCompactionText(referenceText); +} function _compressionReferenceCardHtml(text, open=false){ const copy=_engineAwareCompressionCopy(); const preview=text.split(/\n+/).filter(Boolean).slice(0,2).join(' '); @@ -6957,7 +6963,7 @@ function renderMessages(options){ const referenceText=referenceMessage ? msgContent(referenceMessage)||String(referenceMessage.content||'') : sessionCompressionSummary; - const referenceNode=(!compressionState && !!referenceText && (sessionCompressionAnchor!==null || sessionCompressionAnchorKey || sessionCompressionSummary)) + const referenceNode=(!compressionState && _shouldShowSettledCompressionReference(referenceText) && (sessionCompressionAnchor!==null || sessionCompressionAnchorKey || sessionCompressionSummary)) ? (()=>{const row=document.createElement('div');row.innerHTML=`
${_compressionReferenceCardHtml(referenceText,false)}${_preservedCompressionTaskListCardsHtml(preservedCompressionTaskMessages)}
`;return row.firstElementChild;})() : null; let preservedCompressionTaskCardsAttached=!!referenceNode; @@ -6969,6 +6975,7 @@ function renderMessages(options){ let ri=0; for(const m of S.messages){ if(!m||!m.role||m.role==='tool'){ri++;continue;} + if(_isContextCompactionMessage(m)){ri++;continue;} if(_isPreservedCompressionTaskListMessage(m)){ri++;continue;} if(_isRecoveryControlMessage(m)){ri++;continue;} const hasTc=Array.isArray(m.tool_calls)&&m.tool_calls.length>0; @@ -7174,17 +7181,7 @@ function renderMessages(options){ const footHtml = `
${timeHtml}${editBtn}${ttsBtn}${forkBtn}${copyBtn}${retryBtn}${questionJumpBtn}
`; if(_isContextCompactionMessage(m)){ - if(compressionState || referenceNode){ - continue; - }else{ - currentAssistantTurn=null; - const row=document.createElement('div'); - const preservedForThisCard=preservedCompressionTaskCardsAttached?[]:preservedCompressionTaskMessages; - row.innerHTML=_contextCompactionMessageHtml(m, tsTitle, preservedForThisCard); - if(preservedForThisCard.length) preservedCompressionTaskCardsAttached=true; - inner.appendChild(row.firstElementChild); - continue; - } + continue; } if(isUser){ @@ -7297,7 +7294,7 @@ function renderMessages(options){ } inner.appendChild(node); } - const preservedOnlyNode=(!preservedCompressionTaskCardsAttached&&(!referenceMessage||compressionState)&&preservedCompressionTaskMessages.length) + const preservedOnlyNode=(!preservedCompressionTaskCardsAttached&&(!referenceNode||compressionState)&&preservedCompressionTaskMessages.length) ? (()=>{const row=document.createElement('div');row.innerHTML=`
${_preservedCompressionTaskListCardsHtml(preservedCompressionTaskMessages)}
`;return row.firstElementChild;})() : null; const preservedOnlyAnchor=preservedCompressionRawIdxs.length diff --git a/tests/test_auto_compression_card.py b/tests/test_auto_compression_card.py index 5f59faddf2..07c77f3416 100644 --- a/tests/test_auto_compression_card.py +++ b/tests/test_auto_compression_card.py @@ -379,7 +379,27 @@ def test_context_compaction_branch_precedes_user_bubble_branch(): assert context_idx != -1, "context compaction render branch not found" assert user_idx != -1, "normal user bubble render branch not found" assert context_idx < user_idx - assert "_contextCompactionMessageHtml(m, tsTitle, preservedForThisCard)" in render_prefix + assert "_contextCompactionMessageHtml(m, tsTitle, preservedForThisCard)" not in render_prefix + assert "continue;" in render_prefix[context_idx:user_idx] + + +def test_settled_transcript_suppresses_context_compaction_reference_cards(): + src = _read("static/ui.js") + + assert "function _shouldShowSettledCompressionReference" in src + assert "!_isContextCompactionText(referenceText)" in src + + visible_filter_start = src.find("const vis=S.messages.filter") + assert visible_filter_start != -1, "visible message filter not found" + visible_filter_end = src.find("$('emptyState')", visible_filter_start) + visible_filter = src[visible_filter_start:visible_filter_end] + assert "if(_isContextCompactionMessage(m)) return false;" in visible_filter + + vis_idx_start = src.find("for(const m of S.messages)", visible_filter_end) + assert vis_idx_start != -1, "raw message index loop not found" + vis_idx_end = src.find("let lastUserRawIdx", vis_idx_start) + vis_idx_loop = src[vis_idx_start:vis_idx_end] + assert "if(_isContextCompactionMessage(m)){ri++;continue;}" in vis_idx_loop def test_preserved_task_list_skips_normal_visible_message_path(): @@ -416,7 +436,8 @@ def test_preserved_task_list_renders_through_compression_card_path(): assert "tool-card-compress-reference" in helper assert "data-compression-card=\"1\"" in helper assert "li('list-todo',13)" in helper - assert "_contextCompactionMessageHtml(m, tsTitle, preservedForThisCard)" in src + assert "const preservedOnlyNode=" in src + assert "_preservedCompressionTaskListCardsHtml(preservedCompressionTaskMessages)" in src def test_context_anchor_reference_uses_session_summary_fallback(): @@ -426,7 +447,8 @@ def test_context_anchor_reference_uses_session_summary_fallback(): assert "const sessionCompressionSummary" in src assert "referenceText=referenceMessage" in src assert ": sessionCompressionSummary" in src - assert "!!referenceText && (sessionCompressionAnchor!==null || sessionCompressionAnchorKey || sessionCompressionSummary)" in src + assert "_shouldShowSettledCompressionReference(referenceText)" in src + assert "!_isContextCompactionText(referenceText)" in src def test_compression_anchor_matching_tolerates_legacy_missing_timestamp(): @@ -580,9 +602,8 @@ def test_preserved_task_list_attaches_once_per_render(): assert "const preservedCompressionTaskMessages=_latestPreservedCompressionTaskListMessages(S.messages);" in src assert "S.messages.filter(m=>_isPreservedCompressionTaskListMessage(m))" not in src assert "let preservedCompressionTaskCardsAttached=!!referenceNode;" in src - assert "const preservedForThisCard=preservedCompressionTaskCardsAttached?[]:preservedCompressionTaskMessages;" in src - assert "if(preservedForThisCard.length) preservedCompressionTaskCardsAttached=true;" in src - assert "(!preservedCompressionTaskCardsAttached&&(!referenceMessage||compressionState)&&preservedCompressionTaskMessages.length)" in src + assert "const preservedOnlyNode=" in src + assert "(!preservedCompressionTaskCardsAttached&&(!referenceNode||compressionState)&&preservedCompressionTaskMessages.length)" in src def test_preserved_task_list_is_suppressed_when_latest_todo_state_has_no_active_items(): diff --git a/tests/test_auto_compression_terminal_failure.py b/tests/test_auto_compression_terminal_failure.py new file mode 100644 index 0000000000..d2d31b6e1c --- /dev/null +++ b/tests/test_auto_compression_terminal_failure.py @@ -0,0 +1,441 @@ +"""Regression coverage for compression-exhausted stream finalization.""" + +import copy +import json +import queue +import sys +import types +from pathlib import Path + +from api import models, streaming +from api.models import Session +from api.streaming import ( + _agent_result_terminal_failure, + _session_lacks_final_assistant_answer, +) + +ROOT = Path(__file__).resolve().parents[1] + + +def _read(relpath: str) -> str: + return (ROOT / relpath).read_text(encoding="utf-8") + + +def test_compression_exhausted_after_session_rotation_preserves_snapshot_and_errors_on_continuation( + tmp_path, monkeypatch +): + session_dir = tmp_path / "sessions" + session_dir.mkdir() + monkeypatch.setattr(models, "SESSION_DIR", session_dir) + monkeypatch.setattr(models, "SESSION_INDEX_FILE", session_dir / "_index.json") + monkeypatch.setattr(streaming, "SESSION_DIR", session_dir) + models.SESSIONS.clear() + streaming.SESSIONS.clear() + streaming.STREAMS.clear() + streaming.AGENT_INSTANCES.clear() + streaming.SESSION_AGENT_LOCKS.clear() + old_sid = "old_sid" + new_sid = "new_sid" + stream_id = "stream-compression-exhausted" + session = Session( + session_id=old_sid, + title="Compression test", + workspace=str(tmp_path), + model="gpt-4o", + messages=[], + context_messages=[], + ) + session.active_stream_id = stream_id + session.pending_user_message = "Do the long task." + session.pending_started_at = 1.0 + session.save() + models.SESSIONS[old_sid] = session + streaming.SESSIONS[old_sid] = session + event_queue = queue.Queue() + streaming.STREAMS[stream_id] = event_queue + + class FakeAgent: + def __init__( + self, + model=None, + provider=None, + base_url=None, + api_key=None, + platform=None, + quiet_mode=False, + enabled_toolsets=None, + fallback_model=None, + session_id=None, + session_db=None, + stream_delta_callback=None, + reasoning_callback=None, + tool_progress_callback=None, + interim_assistant_callback=None, + clarify_callback=None, + **kwargs, + ): + self.session_id = session_id + self.stream_delta_callback = stream_delta_callback + self.context_compressor = None + self.session_prompt_tokens = 0 + self.session_completion_tokens = 0 + self.session_estimated_cost_usd = None + self.session_cache_read_tokens = 0 + self.session_cache_write_tokens = 0 + self.reasoning_config = None + self.ephemeral_system_prompt = None + self._last_error = None + + def run_conversation(self, **kwargs): + if self.stream_delta_callback: + self.stream_delta_callback("I am still working through the files.") + self.session_id = new_sid + self._last_error = "Context length exceeded: cannot compress further." + return { + "failed": True, + "partial": True, + "compression_exhausted": True, + "error": "Context length exceeded: cannot compress further.", + "messages": [ + {"role": "user", "content": kwargs.get("persist_user_message", "")}, + {"role": "assistant", "content": "I am still working through the files."}, + {"role": "assistant", "content": "", "tool_calls": [{"id": "call_1"}]}, + {"role": "tool", "tool_call_id": "call_1", "content": "large output"}, + ], + } + + def interrupt(self, _message): + return None + + fake_hermes_state = types.ModuleType("hermes_state") + fake_hermes_state.SessionDB = lambda *_args, **_kwargs: object() + + with monkeypatch.context() as m: + m.setattr(streaming, "get_session", lambda _sid: session) + m.setattr(streaming, "_get_ai_agent", lambda: FakeAgent) + m.setattr(streaming, "resolve_model_provider", lambda *_args, **_kwargs: ("gpt-4o", "openai", None)) + m.setattr("api.config.get_config", lambda *_args, **_kwargs: {}) + m.setattr("api.config._resolve_cli_toolsets", lambda *_args, **_kwargs: []) + m.setitem(sys.modules, "hermes_state", fake_hermes_state) + streaming._run_agent_streaming( + session_id=old_sid, + msg_text="Do the long task.", + model="gpt-4o", + workspace=str(tmp_path), + stream_id=stream_id, + ) + + events = [] + while not event_queue.empty(): + events.append(event_queue.get_nowait()) + apperror_payloads = [payload for event, payload in events if event == "apperror"] + assert apperror_payloads, "expected apperror SSE payload" + payload = apperror_payloads[-1] + assert payload["type"] == "compression_exhausted" + assert payload["session"]["session_id"] == new_sid + assert payload["old_session_id"] == old_sid + assert payload["new_session_id"] == new_sid + + old_payload = json.loads((session_dir / f"{old_sid}.json").read_text(encoding="utf-8")) + new_payload = json.loads((session_dir / f"{new_sid}.json").read_text(encoding="utf-8")) + assert old_payload["pre_compression_snapshot"] is True + assert old_payload["active_stream_id"] is None + assert old_payload["pending_user_message"] is None + assert new_payload["session_id"] == new_sid + assert new_payload["parent_session_id"] == old_sid + assert new_payload["pre_compression_snapshot"] is False + assert new_payload["messages"][-1]["_error"] is True + assert "Context compression exhausted" in new_payload["messages"][-1]["content"] + assert old_sid not in streaming.SESSIONS + assert streaming.SESSIONS[new_sid].session_id == new_sid + + +def test_compression_exhausted_result_is_terminal_failure_even_after_streamed_text(): + result = { + "failed": True, + "partial": True, + "compression_exhausted": True, + "error": "Context length exceeded: 119,194 tokens. Cannot compress further.", + "messages": [ + {"role": "user", "content": "Do the long task."}, + {"role": "assistant", "content": "I am still working through the files."}, + {"role": "assistant", "content": "", "tool_calls": [{"id": "call_1"}]}, + {"role": "tool", "tool_call_id": "call_1", "content": "large output"}, + ], + } + + assert _agent_result_terminal_failure(result) is True + assert _session_lacks_final_assistant_answer(result["messages"]) is True + + +def test_terminal_failure_gates_shape_check_to_no_streamed_text(): + src = _read("api/streaming.py") + start = src.find("_terminal_failure = (") + assert start != -1, "terminal failure assignment not found" + end = src.find("if _terminal_failure:", start) + assert end != -1, "terminal failure guard not found" + block = src[start:end] + + assert "_agent_result_terminal_failure(result)" in block + assert "not _token_sent" in block + assert "_session_lacks_final_assistant_answer(_all_result_messages)" in block + assert "not _assistant_added" not in block + + +def test_completed_tool_tail_without_final_assistant_is_not_successful_done(): + messages = [ + {"role": "user", "content": "Run the tool then answer."}, + {"role": "assistant", "content": "", "tool_calls": [{"id": "call_1"}]}, + {"role": "tool", "tool_call_id": "call_1", "content": "result"}, + ] + + assert _session_lacks_final_assistant_answer(messages) is True + + +def test_assistant_content_with_tool_calls_is_not_final_answer(): + messages = [ + {"role": "user", "content": "Search, then answer."}, + { + "role": "assistant", + "content": "I found a likely source and will inspect it.", + "tool_calls": [{"id": "call_1"}], + }, + ] + + assert _session_lacks_final_assistant_answer(messages) is True + + +def test_context_compaction_marker_is_not_final_answer(): + messages = [ + {"role": "user", "content": "x"}, + { + "role": "assistant", + "content": "[CONTEXT COMPACTION — REFERENCE ONLY] summary", + }, + ] + + assert _session_lacks_final_assistant_answer(messages) is True + + +def test_context_compaction_marker_before_final_text_is_successful_answer(): + messages = [ + {"role": "user", "content": "x"}, + { + "role": "assistant", + "content": "[CONTEXT COMPACTION — REFERENCE ONLY] summary", + }, + {"role": "assistant", "content": "Here is the final answer."}, + ] + + assert _session_lacks_final_assistant_answer(messages) is False + + +def test_context_compaction_marker_before_tool_tail_is_not_final_answer(): + messages = [ + {"role": "user", "content": "x"}, + { + "role": "assistant", + "content": "[CONTEXT COMPACTION — REFERENCE ONLY] summary", + }, + { + "role": "assistant", + "content": "I will inspect the result.", + "tool_calls": [{"id": "call_1"}], + }, + ] + + assert _session_lacks_final_assistant_answer(messages) is True + + +def test_final_assistant_text_is_successful_terminal_answer(): + messages = [ + {"role": "user", "content": "Run the tool then answer."}, + {"role": "assistant", "content": "", "tool_calls": [{"id": "call_1"}]}, + {"role": "tool", "tool_call_id": "call_1", "content": "result"}, + {"role": "assistant", "content": "Here is the final answer."}, + ] + + assert _session_lacks_final_assistant_answer(messages) is False + + +def test_assistant_tool_call_turn_followed_by_final_text_is_successful_answer(): + messages = [ + {"role": "user", "content": "Search, then answer."}, + { + "role": "assistant", + "content": "I found a likely source and will inspect it.", + "tool_calls": [{"id": "call_1"}], + }, + {"role": "tool", "tool_call_id": "call_1", "content": "result"}, + {"role": "assistant", "content": "Here is the final answer."}, + ] + + assert _session_lacks_final_assistant_answer(messages) is False + + +def test_compression_exhausted_apperror_clears_reference_ui_and_labels_error(): + src = _read("static/messages.js") + start = src.find("source.addEventListener('apperror'") + assert start != -1, "apperror listener not found" + end = src.find("source.addEventListener('warning'", start) + assert end != -1, "warning listener after apperror not found" + block = src[start:end] + + assert "const isCompressionExhausted=d.type==='compression_exhausted';" in block + assert "isCompressionExhausted?'Context compression exhausted'" in block + assert "if(typeof clearCompressionUi==='function') clearCompressionUi();" in block + assert "window._compressionUi=null;" in block + assert "const eventSid=d.old_session_id||d.session_id||'';" in block + assert "const continuationSid=(d.session&&d.session.session_id)||d.new_session_id||d.continuation_session_id||'';" in block + assert "if(d.session&&typeof d.session==='object')" in block + assert "S.session=d.session;" in block + + +def test_apperror_matches_only_current_or_continuation_session_for_background_errors(): + src = _read("static/messages.js") + start = src.find("source.addEventListener('apperror'") + assert start != -1, "apperror listener not found" + end = src.find("source.addEventListener('warning'", start) + assert end != -1, "warning listener after apperror not found" + block = src[start:end] + + assert "const eventSid=d.old_session_id||d.session_id||'';" in block + assert "const continuationSid=(d.session&&d.session.session_id)||d.new_session_id||d.continuation_session_id||'';" in block + assert "const eventMatchesCurrent=!!(currentSid&&(eventSid===currentSid||continuationSid===currentSid));" in block + + +def test_apperror_payload_enriched_before_enqueue(tmp_path, monkeypatch): + class _CaptureQueue: + def __init__(self): + self.events = [] + + def put_nowait(self, item): + event, payload = item + self.events.append((event, payload, copy.deepcopy(payload))) + + session_dir = tmp_path / "sessions" + session_dir.mkdir() + monkeypatch.setattr(models, "SESSION_DIR", session_dir) + monkeypatch.setattr(models, "SESSION_INDEX_FILE", session_dir / "_index.json") + monkeypatch.setattr(streaming, "SESSION_DIR", session_dir) + models.SESSIONS.clear() + streaming.SESSIONS.clear() + streaming.STREAMS.clear() + streaming.AGENT_INSTANCES.clear() + streaming.SESSION_AGENT_LOCKS.clear() + + old_sid = "old_sid_capture" + new_sid = "new_sid_capture" + stream_id = "stream-compression-exhausted-capture" + session = models.Session( + session_id=old_sid, + title="Compression test", + workspace=str(tmp_path), + model="gpt-4o", + messages=[], + context_messages=[], + ) + session.active_stream_id = stream_id + session.pending_user_message = "Do the long task." + session.pending_started_at = 1.0 + session.save() + models.SESSIONS[old_sid] = session + streaming.SESSIONS[old_sid] = session + captured = _CaptureQueue() + streaming.STREAMS[stream_id] = captured + + class FakeAgent: + def __init__( + self, + model=None, + provider=None, + base_url=None, + api_key=None, + platform=None, + quiet_mode=False, + enabled_toolsets=None, + fallback_model=None, + session_id=None, + session_db=None, + stream_delta_callback=None, + reasoning_callback=None, + tool_progress_callback=None, + interim_assistant_callback=None, + clarify_callback=None, + **kwargs, + ): + self.session_id = session_id + self.stream_delta_callback = stream_delta_callback + self.context_compressor = None + self.session_prompt_tokens = 0 + self.session_completion_tokens = 0 + self.session_estimated_cost_usd = None + self.session_cache_read_tokens = 0 + self.session_cache_write_tokens = 0 + self.reasoning_config = None + self.ephemeral_system_prompt = None + self._last_error = "Context length exceeded: cannot compress further." + + def run_conversation(self, **kwargs): + if self.stream_delta_callback: + self.stream_delta_callback("I am still working through the files.") + self.session_id = new_sid + return { + "failed": True, + "partial": True, + "compression_exhausted": True, + "error": "Context length exceeded: cannot compress further.", + "messages": [ + {"role": "user", "content": kwargs.get("persist_user_message", "")}, + {"role": "assistant", "content": "I am still working through the files."}, + {"role": "assistant", "content": "", "tool_calls": [{"id": "call_1"}]}, + {"role": "tool", "tool_call_id": "call_1", "content": "large output"}, + ], + } + + def interrupt(self, _message): + return None + + fake_hermes_state = types.ModuleType("hermes_state") + fake_hermes_state.SessionDB = lambda *_args, **_kwargs: object() + + with monkeypatch.context() as m: + m.setattr(streaming, "get_session", lambda _sid: session) + m.setattr(streaming, "_get_ai_agent", lambda: FakeAgent) + m.setattr(streaming, "resolve_model_provider", lambda *_args, **_kwargs: ("gpt-4o", "openai", None)) + m.setitem(sys.modules, "hermes_state", fake_hermes_state) + m.setattr("api.config.get_config", lambda *_args, **_kwargs: {}) + m.setattr("api.config._resolve_cli_toolsets", lambda *_args, **_kwargs: []) + m.setattr(streaming, "redact_session_data", lambda s: s) + + streaming._run_agent_streaming( + session_id=old_sid, + msg_text="Do the long task.", + model="gpt-4o", + workspace=str(tmp_path), + stream_id=stream_id, + ) + + apperror_payloads = [ + (payload, payload_before) + for event, payload, payload_before in captured.events + if event == "apperror" + ] + assert apperror_payloads, "expected apperror SSE payload" + payload_after, payload_before = apperror_payloads[-1] + assert payload_after == payload_before, "apperror payload changed after enqueue" + assert payload_after["session_id"] == new_sid + assert payload_after["old_session_id"] == old_sid + assert payload_after["new_session_id"] == new_sid + + +def test_exception_apperror_payload_includes_session_id_before_enqueue(): + src = _read("api/streaming.py") + start = src.find("_error_payload = _provider_error_payload(err_str, _exc_type, _exc_hint)") + assert start != -1, "exception apperror payload path not found" + end = src.find("put('apperror', _error_payload)", start) + assert end != -1, "exception apperror enqueue not found" + block = src[start:end] + + assert "_error_payload['session_id'] = getattr(s, 'session_id', session_id)" in block + assert "_error_payload['old_session_id'] = session_id" in block diff --git a/tests/test_issue765_streaming_persistence.py b/tests/test_issue765_streaming_persistence.py index 7a74bf784d..36935f0dcc 100644 --- a/tests/test_issue765_streaming_persistence.py +++ b/tests/test_issue765_streaming_persistence.py @@ -360,7 +360,9 @@ def test_silent_failure_path_does_not_reacquire_agent_lock(self): "with _agent_lock:\n" " if not ephemeral and not _stream_writeback_is_current(s, stream_id):" ) - silent_failure_idx = src.find("if not _assistant_added and not _token_sent:") + silent_failure_idx = src.find( + "if _terminal_failure or (not _assistant_added and not _token_sent):" + ) inner_lock_idx = src.find("with _agent_lock:", outer_lock_idx + 1) compression_idx = src.find("# ── Handle context compression side effects ──")