Release v0.51.292 — Release JH (stage-s4 — compression-exhausted turns surface as errors #3316 fixes #3315) (#3705)

* fix(#3315): surface compression-exhausted/no-final-answer turns as errors (#3316)

When Hermes Agent exhausts context compression in a long tool-heavy turn, the
streamed result can end on a tool result / assistant(tool_calls) turn with no
final assistant answer. WebUI was finalizing that as a completed response.
Now _session_lacks_final_assistant_answer() + _agent_result_terminal_failure()
classify these as terminal failures and surface an apperror instead. The
compression session-id migration + pre-compression snapshot now run BEFORE the
terminal-failure return (ordering bug from the prior hold) so state stays
consistent when exhaustion fires after the agent rotated session_id.

Co-authored-by: Frank Song <franksong2702@gmail.com>

* docs(changelog): v0.51.292 — Release JH (stage-s4, #3316 fixes #3315)

---------

Co-authored-by: Frank Song <franksong2702@gmail.com>
Co-authored-by: nesquena-hermes <[email protected]>

CHANGELOG.md

@@ -3,6 +3,11 @@
 
 ## [Unreleased]
 
+## [v0.51.292] — 2026-06-06 — Release JH (stage-s4 — compression-exhausted turns surface as errors, not fake completions)
+
+### Fixed
+- **Context-compression-exhausted turns are no longer finalized as a falsely "completed" response.** When Hermes Agent exhausts context compression in a long tool-heavy turn, the streamed result can end on a tool result or an assistant `tool_calls` turn with no final assistant answer. WebUI previously rendered that as a settled, completed reply. It now classifies a persisted transcript that ends in a tool/tool-call/empty-assistant tail (or an internal `[CONTEXT COMPACTION — REFERENCE ONLY]` marker) — and `compression_exhausted`/`failed`/`partial` agent results — as a terminal failure and surfaces a clear error instead. The compression session-id migration and pre-compression snapshot now run **before** the terminal-failure path returns, so frontend/backend session state stays consistent when exhaustion fires after the agent rotates `session_id`. (#3316, @franksong2702; fixes #3315)
+
 ## [v0.51.291] — 2026-06-06 — Release JG (stage-s2 — preserve live turn content when switching away mid-stream)
 
 ### Fixed

api/streaming.py

@@ -885,6 +885,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',
@@ -909,6 +915,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',
@@ -3697,6 +3709,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
 
 
@@ -5977,6 +6035,128 @@ def _run_agent_streaming(
                                 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
@@ -5995,8 +6175,17 @@ def _run_agent_streaming(
                     _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:
@@ -6146,7 +6335,6 @@ def _run_agent_streaming(
                             _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.
@@ -6174,130 +6362,21 @@ def _run_agent_streaming(
                             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)
@@ -7281,6 +7360,8 @@ def _run_agent_streaming(
                         )
                     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.

static/messages.js

@@ -2489,12 +2489,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';
@@ -2502,14 +2507,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});
           }
@@ -2526,13 +2541,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

static/ui.js

@@ -6458,7 +6458,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||''));
@@ -6536,6 +6539,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(' ');
@@ -7056,7 +7062,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=`<div class="compression-turn"><div class="compression-turn-blocks">${_compressionReferenceCardHtml(referenceText,false)}${_preservedCompressionTaskListCardsHtml(preservedCompressionTaskMessages)}</div></div>`;return row.firstElementChild;})()
     : null;
   let preservedCompressionTaskCardsAttached=!!referenceNode;
@@ -7068,6 +7074,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;
@@ -7274,17 +7281,7 @@ function renderMessages(options){
     const footHtml = `<div class="msg-foot">${timeHtml}<span class="msg-actions">${editBtn}${ttsBtn}${forkBtn}${copyBtn}${retryBtn}</span>${questionJumpBtn}</div>`;
 
     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){
@@ -7397,7 +7394,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=`<div class="compression-turn"><div class="compression-turn-blocks">${_preservedCompressionTaskListCardsHtml(preservedCompressionTaskMessages)}</div></div>`;return row.firstElementChild;})()
     : null;
   const preservedOnlyAnchor=preservedCompressionRawIdxs.length

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():

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

tests/test_issue765_streaming_persistence.py

@@ -360,7 +360,9 @@ class TestIssue765FollowupHardening:
             "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 ──")