From af69c4a2b8ec39272fbf73066210e782c53a0b5b Mon Sep 17 00:00:00 2001 From: starship-s <45587122+starship-s@users.noreply.github.com> Date: Thu, 6 Aug 2026 23:35:13 -0600 Subject: [PATCH] perf(streaming): bound terminal session payloads Assisted-by: Hermes Agent:gpt-5.6-sol Assisted-by: Codex:gpt-5.6-luna Assisted-by: Claude Code:claude-opus-5 --- api/gateway_chat.py | 19 +- api/routes.py | 44 +- api/session_ops.py | 41 ++ api/streaming.py | 129 ++++-- static/messages.js | 300 +++++++++--- static/sessions.js | 3 +- tests/test_auto_compression_card.py | 2 +- .../test_auto_compression_terminal_failure.py | 13 +- tests/test_cancelled_turn_status.py | 13 +- tests/test_compression_phantom_barrier.py | 6 +- tests/test_extension_turn_lifecycle.py | 14 + tests/test_issue2655_frontend.py | 2 +- tests/test_issue3929_process_wakeup_pause.py | 204 ++++++++- ...ssue4720_done_scroll_jump_first_message.py | 29 +- ...5224_terminal_error_transcript_preserve.py | 2 - ...test_issue6751_api_content_agent_replay.py | 2 + ...t_issue856_background_completion_unread.py | 5 +- tests/test_live_activity_timeline.py | 2 +- tests/test_live_stream_ux.py | 6 +- ...test_live_to_final_anchor_visible_order.py | 29 +- tests/test_session_rotate_url_sync.py | 15 +- tests/test_sidebar_first_turn_visibility.py | 42 ++ tests/test_sprint42.py | 8 +- ...st_streaming_done_payload_message_count.py | 18 +- tests/test_streaming_markdown.py | 2 +- tests/test_terminal_session_tail_and_merge.py | 430 ++++++++++++++++++ 26 files changed, 1148 insertions(+), 232 deletions(-) create mode 100644 tests/test_terminal_session_tail_and_merge.py diff --git a/api/gateway_chat.py b/api/gateway_chat.py index 72ac71f032a..1bcf0dd86e4 100644 --- a/api/gateway_chat.py +++ b/api/gateway_chat.py @@ -34,7 +34,7 @@ unregister_stream_owner, update_active_run, ) -from api.helpers import _redact_text, redact_session_data +from api.helpers import _redact_text from api.models import clear_process_wakeup_pause, get_session, merge_session_messages_append_only from api.run_journal import RunJournalWriter, bound_run_journal_snapshot_args @@ -737,7 +737,7 @@ def _settle_gateway_terminal_error(session_id, stream_id, workspace, model, mode _classify_provider_error, _materialize_pending_user_turn_before_error, _provider_error_payload, - _session_payload_with_full_messages, + _best_effort_terminal_session_payload, _snapshot_and_append_partial_on_error, _terminal_turn_duration, ) @@ -788,9 +788,9 @@ def _settle_gateway_terminal_error(session_id, stream_id, workspace, model, mode terminal_session_persisted = True except Exception: logger.debug("Failed to persist gateway terminal error settlement", exc_info=True) - error_payload["session"] = redact_session_data( - _session_payload_with_full_messages(session, tool_calls=[]) - ) + terminal_session_payload = _best_effort_terminal_session_payload(session) + if terminal_session_payload is not None: + error_payload["session"] = terminal_session_payload error_payload["session_id"] = session.session_id error_payload["terminal_session_persisted"] = terminal_session_persisted if terminal_session_persisted: @@ -1351,9 +1351,12 @@ def _restore_cancelled_success_writeback(): session_id, goal_exc, ) - from api.streaming import _session_payload_with_full_messages - gateway_session_payload = _session_payload_with_full_messages(s, tool_calls=[]) - put_gateway_event("done", {"session": redact_session_data(gateway_session_payload), "usage": usage}) + from api.streaming import _best_effort_terminal_session_payload + gateway_done_payload = {"usage": usage} + gateway_session_payload = _best_effort_terminal_session_payload(s) + if gateway_session_payload is not None: + gateway_done_payload["session"] = gateway_session_payload + put_gateway_event("done", gateway_done_payload) put_gateway_event("stream_end", {"session_id": session_id}) except urllib.error.HTTPError as exc: try: diff --git a/api/routes.py b/api/routes.py index 1b902280568..3639f66761b 100644 --- a/api/routes.py +++ b/api/routes.py @@ -61,6 +61,7 @@ ) from api.gateway_restart import restart_active_profile_gateway from api.shares import create_or_refresh_share, load_share, revoke_share +from api.session_ops import _messages_for_limited_payload logger = logging.getLogger(__name__) @@ -8772,7 +8773,6 @@ def _message_window_for_display(messages, msg_limit=None, msg_before=None, expan return window, start_idx -_LIMITED_TOOL_CONTENT_MAX_CHARS = 4096 # Server-side ceiling on the ?msg_limit= tail-window size. A client could # otherwise request msg_limit=1000000 and force the server to assemble and # serialize an unbounded message payload (the frontend's own pagination grows @@ -8841,48 +8841,6 @@ def _state_db_backstop_limit_for_display(session, msg_before) -> int | None: return None if has_boundary_prefix else _STATE_DB_DISPLAY_ROW_BACKSTOP -_LIMITED_TOOL_CONTENT_NOTICE = ( - "\n\n[Tool output truncated in paginated session response; " - "load the full transcript to inspect the complete result.]" -) - - -def _tool_message_for_limited_payload(message): - """Return a bounded copy of large hidden tool-result rows for paginated loads.""" - if not isinstance(message, dict) or str(message.get("role") or "").lower() != "tool": - return message - content = message.get("content") - if content in (None, ""): - return message - if isinstance(content, str): - text = content - else: - try: - text = json.dumps(content, ensure_ascii=False, default=str) - except Exception: - text = str(content) - if len(text) <= _LIMITED_TOOL_CONTENT_MAX_CHARS: - return message - clipped = dict(message) - preview = text[:_LIMITED_TOOL_CONTENT_MAX_CHARS] + _LIMITED_TOOL_CONTENT_NOTICE - if isinstance(content, str): - clipped["content"] = preview - elif isinstance(content, list): - clipped["content"] = [{"type": "text", "text": preview}] - elif isinstance(content, dict): - clipped["content"] = {"_truncated": True, "preview": preview} - else: - clipped["content"] = preview - clipped["_content_truncated"] = True - clipped["_content_original_chars"] = len(text) - return clipped - - -def _messages_for_limited_payload(messages) -> list: - """Bound hidden tool-result payloads before sending a msg_limit response.""" - return [_tool_message_for_limited_payload(msg) for msg in list(messages or [])] - - def _limited_webui_messages_for_display(session, state_db_messages) -> list: """Return the display sidecar plus only necessary state.db rows for msg_limit. diff --git a/api/session_ops.py b/api/session_ops.py index 075b8a25379..ffc3a93ceb6 100644 --- a/api/session_ops.py +++ b/api/session_ops.py @@ -17,6 +17,47 @@ logger = logging.getLogger(__name__) AUTO_TITLE_LABELS = {'untitled', 'new chat'} +_LIMITED_TOOL_CONTENT_MAX_CHARS = 4096 +_LIMITED_TOOL_CONTENT_NOTICE = ( + "\n\n[Tool output truncated in paginated session response; " + "load the full transcript to inspect the complete result.]" +) + + +def _tool_message_for_limited_payload(message): + """Return a bounded copy of a large hidden tool-result row.""" + if not isinstance(message, dict) or str(message.get("role") or "").lower() != "tool": + return message + content = message.get("content") + if content in (None, ""): + return message + if isinstance(content, str): + text = content + else: + try: + text = json.dumps(content, ensure_ascii=False, default=str) + except Exception: + text = str(content) + if len(text) <= _LIMITED_TOOL_CONTENT_MAX_CHARS: + return message + clipped = dict(message) + preview = text[:_LIMITED_TOOL_CONTENT_MAX_CHARS] + _LIMITED_TOOL_CONTENT_NOTICE + if isinstance(content, str): + clipped["content"] = preview + elif isinstance(content, list): + clipped["content"] = [{"type": "text", "text": preview}] + elif isinstance(content, dict): + clipped["content"] = {"_truncated": True, "preview": preview} + else: + clipped["content"] = preview + clipped["_content_truncated"] = True + clipped["_content_original_chars"] = len(text) + return clipped + + +def _messages_for_limited_payload(messages) -> list: + """Bound hidden tool-result rows before a limited session payload is sent.""" + return [_tool_message_for_limited_payload(message) for message in list(messages or [])] def _live_active_stream_id(session) -> str | None: diff --git a/api/streaming.py b/api/streaming.py index 21ff90059ad..6ca8313fa09 100644 --- a/api/streaming.py +++ b/api/streaming.py @@ -23,7 +23,7 @@ import copy import inspect from pathlib import Path -from typing import Optional +from typing import Any, Optional logger = logging.getLogger(__name__) @@ -72,7 +72,11 @@ record_process_wakeup_provider_unavailable_pause, reconciled_state_db_messages_for_session, ) -from api.session_ops import mark_session_title_generated, session_has_manual_title +from api.session_ops import ( + _messages_for_limited_payload, + mark_session_title_generated, + session_has_manual_title, +) from api.process_event_utils import ( build_active_turn_token, claim_async_delegation_delivery, @@ -104,6 +108,53 @@ def _session_payload_with_full_messages(session, *, tool_calls=None): return raw +def _redacted_terminal_session_payload(session) -> dict: + """Return a bounded, redacted session snapshot for ordinary terminal SSE.""" + messages = list(getattr(session, 'messages', None) or []) + renderable_indexes = [] + renderable_count = 0 + for index, message in enumerate(messages): + if visible_messages_for_anchor([message], auto_compression=True): + renderable_count += 1 + renderable_indexes.append(index) + if len(renderable_indexes) > 30: + renderable_indexes.pop(0) + start = renderable_indexes[0] if renderable_count > 30 else 0 + start = max(start, len(messages) - 300) + retained = _messages_for_limited_payload(messages[start:]) + raw = session.compact() | { + 'messages': retained, + 'message_count': len(messages), + '_messages_offset': start, + '_messages_truncated': bool(start), + } + terminal_tool_calls = [] + for tool_call in getattr(session, 'tool_calls', None) or []: + if not isinstance(tool_call, dict): + continue + assistant_msg_idx = tool_call.get('assistant_msg_idx') + if isinstance(assistant_msg_idx, bool) or not isinstance(assistant_msg_idx, int): + continue + if assistant_msg_idx == -1: + terminal_tool_calls.append(dict(tool_call)) + elif start <= assistant_msg_idx < len(messages): + projected = dict(tool_call) + projected['assistant_msg_idx'] = assistant_msg_idx - start + terminal_tool_calls.append(projected) + raw['_tool_calls_truncated'] = len(terminal_tool_calls) > 300 + raw['tool_calls'] = terminal_tool_calls[-300:] + attach_todo_state(raw, messages) + return redact_session_data(raw) + + +def _best_effort_terminal_session_payload(session) -> dict | None: + try: + return _redacted_terminal_session_payload(session) + except Exception: + logger.debug("Failed to build terminal session payload", exc_info=True) + return None + + def _compact_for_echo_compare(value: str) -> str: """Normalize visible stream text for duplicate echo detection.""" return re.sub(r'\s+', '', str(value or '')) @@ -123,28 +174,18 @@ def _strip_compact_echo_suffix(value: str, suffix: str, *, search_window: int = return raw, False -def _redacted_session_payload_with_full_messages(session, *, tool_calls=None) -> dict | None: - """Best-effort terminal SSE session payload for already-persisted state.""" - try: - return redact_session_data( - _session_payload_with_full_messages(session, tool_calls=tool_calls) - ) - except Exception: - logger.debug("Failed to build redacted session payload", exc_info=True) - return None - - def _ephemeral_session_payload(session_id: str, messages) -> dict: """Project the non-persistent ``/btw`` terminal session for public SSE.""" - return redact_session_data( - {'session_id': session_id, 'messages': messages if isinstance(messages, list) else []} - ) + raw: dict[str, Any] = {'session_id': session_id} + if isinstance(messages, list) and messages: + raw['messages'] = messages + return redact_session_data(raw) def _cancel_event_payload( message: str = "Cancelled by user", *, - session: dict | None = None, + session: object | None = None, ) -> dict: """Return base cancel terminal event metadata.""" payload = { @@ -153,8 +194,13 @@ def _cancel_event_payload( 'status': 'cancelled', } if session: - payload['session'] = session - payload['session_id'] = session.get('session_id') + if isinstance(session, dict): + session_payload = session + else: + session_payload = _best_effort_terminal_session_payload(session) + if session_payload: + payload['session'] = session_payload + payload['session_id'] = session_payload.get('session_id') return payload @@ -8465,7 +8511,7 @@ def _agent_status_callback(kind, message): if cancel_event.is_set(): with _agent_lock: _finalize_cancelled_turn(s, ephemeral=ephemeral, message='Task cancelled before start.', stream_id=stream_id) - put('cancel', _cancel_event_payload('Cancelled before start')) + put('cancel', _cancel_event_payload('Cancelled before start', session=None if ephemeral else s)) return # Resolve profile home for this agent run — use the session's own profile @@ -9751,7 +9797,7 @@ def _fallback_entries(_raw): logger.debug("Failed to interrupt agent before start") with _agent_lock: _finalize_cancelled_turn(s, ephemeral=ephemeral, message='Task cancelled before start.', stream_id=stream_id) - put('cancel', _cancel_event_payload('Cancelled by user')) + put('cancel', _cancel_event_payload('Cancelled by user', session=None if ephemeral else s)) return # Prepend workspace context so the agent always knows which directory @@ -9983,7 +10029,7 @@ def _periodic_checkpoint(): ) except Exception: logger.debug("Failed to append cancelled turn journal event", exc_info=True) - put('cancel', _cancel_event_payload('Cancelled by user')) + put('cancel', _cancel_event_payload('Cancelled by user', session=None if ephemeral else s)) return # ── Ephemeral mode (/btw): deliver answer, skip persistence, cleanup ── if ephemeral: @@ -9996,9 +10042,7 @@ def _periodic_checkpoint(): # payload is still public output. Project the ephemeral # session before enqueueing it so raw Agent ``api_content`` or # provenance aliases cannot cross the wire. - _ephemeral_session = _ephemeral_session_payload( - session_id, result.get('messages', []) - ) + _ephemeral_session = _ephemeral_session_payload(session_id, []) put('done', { 'session': _ephemeral_session, 'usage': {'input_tokens': 0, 'output_tokens': 0}, @@ -10032,7 +10076,7 @@ def _periodic_checkpoint(): ) except Exception: logger.debug("Failed to append cancelled turn journal event", exc_info=True) - put('cancel', _cancel_event_payload('Cancelled by user')) + put('cancel', _cancel_event_payload('Cancelled by user', session=None if ephemeral else s)) return _writeback_timings = [] _writeback_started = time.perf_counter() @@ -10093,7 +10137,7 @@ def _periodic_checkpoint(): ) except Exception: logger.debug("Failed to append cancelled turn journal event", exc_info=True) - put('cancel', _cancel_event_payload('Cancelled by user')) + put('cancel', _cancel_event_payload('Cancelled by user', session=None if ephemeral else s)) return _result_messages = _settle_result_messages( s, @@ -10346,7 +10390,7 @@ def _periodic_checkpoint(): ) except Exception: logger.debug("Failed to append cancelled turn journal event", exc_info=True) - put('cancel', _cancel_event_payload('Cancelled by user')) + put('cancel', _cancel_event_payload('Cancelled by user', session=None if ephemeral else s)) return _err_str = str(_last_err) if _last_err else '' if _is_quota: @@ -10573,9 +10617,9 @@ def _periodic_checkpoint(): s.save() except Exception: pass - _error_payload['session'] = redact_session_data( - _session_payload_with_full_messages(s, tool_calls=s.tool_calls) - ) + _terminal_session_payload = _best_effort_terminal_session_payload(s) + if _terminal_session_payload is not None: + _error_payload['session'] = _terminal_session_payload _error_payload['session_id'] = s.session_id _error_payload['old_session_id'] = _compression_origin_session_id if _compression_continuation_session_id is not None: @@ -11015,7 +11059,7 @@ def _periodic_checkpoint(): ) except Exception: logger.debug("Failed to append cancelled turn journal event", exc_info=True) - put('cancel', _cancel_event_payload('Cancelled by user')) + put('cancel', _cancel_event_payload('Cancelled by user', session=None if ephemeral else s)) return with _stream_writeback_stage(_writeback_timings, "session_save"): s.save() @@ -11033,7 +11077,7 @@ def _periodic_checkpoint(): ) except Exception: logger.debug("Failed to append cancelled turn journal event", exc_info=True) - put('cancel', _cancel_event_payload('Cancelled by user')) + put('cancel', _cancel_event_payload('Cancelled by user', session=None if ephemeral else s)) return if not ephemeral: try: @@ -11137,7 +11181,7 @@ def _periodic_checkpoint(): ) except Exception: logger.debug("Failed to append cancelled turn journal event", exc_info=True) - put('cancel', _cancel_event_payload('Cancelled by user')) + put('cancel', _cancel_event_payload('Cancelled by user', session=None if ephemeral else s)) return try: _latest_pause_owner = get_session(getattr(s, 'session_id', session_id)) @@ -11169,7 +11213,7 @@ def _periodic_checkpoint(): ) except Exception: logger.debug("Failed to append cancelled turn journal event", exc_info=True) - put('cancel', _cancel_event_payload('Cancelled by user')) + put('cancel', _cancel_event_payload('Cancelled by user', session=None if ephemeral else s)) return with _stream_writeback_stage(_writeback_timings, "process_wakeup_pause_clear_save"): s.save(touch_updated_at=False) @@ -11192,7 +11236,7 @@ def _periodic_checkpoint(): ) except Exception: logger.debug("Failed to append cancelled turn journal event", exc_info=True) - put('cancel', _cancel_event_payload('Cancelled by user')) + put('cancel', _cancel_event_payload('Cancelled by user', session=None if ephemeral else s)) return _success_writeback_committed = True usage = { @@ -11428,8 +11472,10 @@ def _periodic_checkpoint(): except Exception as _goal_exc: logger.debug("Goal continuation hook failed for session %s: %s", session_id, _goal_exc) with _stream_writeback_stage(_writeback_timings, "done_payload"): - raw_session = _session_payload_with_full_messages(s, tool_calls=tool_calls) - _done_payload = {'session': redact_session_data(raw_session), 'usage': usage} + _done_payload = {'usage': usage} + _terminal_session_payload = _best_effort_terminal_session_payload(s) + if _terminal_session_payload is not None: + _done_payload['session'] = _terminal_session_payload if _tool_limit_reached: _done_payload['terminal_state'] = 'tool_limit_reached' _done_payload['terminal_reason'] = 'max_iterations' @@ -11571,7 +11617,7 @@ def _periodic_checkpoint(): ) except Exception: logger.debug("Failed to append cancelled turn journal event", exc_info=True) - put('cancel', _cancel_event_payload('Cancelled by user')) + put('cancel', _cancel_event_payload('Cancelled by user', session=None if ephemeral else s)) return _exc_is_quota = _classification['type'] == 'quota_exhausted' # Exception quota text still includes: 'more credits' in _exc_lower, 'can only afford' in _exc_lower, 'fewer max_tokens' in _exc_lower. @@ -11822,6 +11868,9 @@ def _periodic_checkpoint(): ) except Exception: logger.debug("Failed to append interrupted turn journal event", exc_info=True) + _terminal_session_payload = _best_effort_terminal_session_payload(s) + if _terminal_session_payload is not None: + _error_payload['session'] = _terminal_session_payload _error_payload['session_id'] = getattr(s, 'session_id', session_id) _error_payload['old_session_id'] = session_id put('apperror', _error_payload) @@ -12431,7 +12480,7 @@ def cancel_stream(stream_id: str) -> bool: 'timestamp': int(time.time()), }) _cs.save() - _cancel_session_payload = _redacted_session_payload_with_full_messages(_cs) + _cancel_session_payload = _redacted_terminal_session_payload(_cs) except Exception: logger.debug("Failed to clear session state on cancel for %s", _cancel_session_id) diff --git a/static/messages.js b/static/messages.js index 14b5f2875d5..22d070ef224 100644 --- a/static/messages.js +++ b/static/messages.js @@ -2079,6 +2079,118 @@ function _dispatchExtensionTurnLifecycle(type,sessionId,streamId,details={}){ return false; } } +function _applyEmbeddedTerminalSession(incoming,currentSession,currentMessages,currentOffset){ + if(incoming==null){ + if(!currentSession||typeof currentSession!=='object') return null; + return {...currentSession,messages:Array.isArray(currentMessages)?currentMessages:(Array.isArray(currentSession.messages)?currentSession.messages:[])}; + } + if(typeof incoming!=='object'||Array.isArray(incoming)) return null; + const incomingMessages=Array.isArray(incoming.messages)?incoming.messages:null; + if(Object.prototype.hasOwnProperty.call(incoming,'tool_calls')&&!Array.isArray(incoming.tool_calls)) return null; + if(Object.prototype.hasOwnProperty.call(incoming,'_tool_calls_truncated')&& + typeof incoming._tool_calls_truncated!=='boolean') return null; + const incomingSid=String(incoming.session_id||''); + const incomingParentSid=String(incoming.parent_session_id||''); + const currentSid=String(currentSession&¤tSession.session_id||''); + if(!incomingSid||!incomingMessages) return null; + const sameCoordinateLineage=incomingSid===currentSid||incomingParentSid===currentSid; + if(!currentSession||!Array.isArray(currentMessages)||!currentSid||!sameCoordinateLineage){ + return {...incoming,messages:incomingMessages}; + } + const isOffset=value=>Number.isFinite(value)&&Number.isInteger(value)&&value>=0; + const incomingOffset=incoming._messages_offset; + const messageCount=incoming.message_count; + const hasWindowMetadata=Array.isArray(incomingMessages)&& + Object.prototype.hasOwnProperty.call(incoming,'_messages_offset')&& + Object.prototype.hasOwnProperty.call(incoming,'message_count'); + if(!hasWindowMetadata||!isOffset(incomingOffset)||!isOffset(messageCount)||!isOffset(currentOffset)) return null; + if(Object.prototype.hasOwnProperty.call(incoming,'_messages_truncated')&& + (typeof incoming._messages_truncated!=='boolean'||incoming._messages_truncated!==!!incomingOffset)) return null; + const currentEnd=currentOffset+currentMessages.length; + const currentCount=currentSession.message_count; + const incomingEnd=incomingOffset+incomingMessages.length; + if(incomingEnd!==messageCount) return null; + if((currentCount!==undefined&¤tCount!==currentEnd)||incomingOffset>currentEnd){ + if(!Object.prototype.hasOwnProperty.call(incoming,'_messages_truncated')) return null; + return {...incoming,messages:incomingMessages}; + } + if(incomingEnd=incomingOffset) mergedMessages.push(incomingMessages[coordinate-incomingOffset]); + else mergedMessages.push(currentMessages[coordinate-currentOffset]); + } + const translateCalls=(calls,offset,width)=>{ + const translated=[]; + for(const call of Array.isArray(calls)?calls:[]){ + if(!call||typeof call!=='object') continue; + const next={...call}; + if(Object.prototype.hasOwnProperty.call(next,'assistant_msg_idx')){ + const index=next.assistant_msg_idx; + if(index===-1) { + next.assistant_msg_idx=-1; + } else if(!isOffset(index)||index>=width) { + continue; + } else { + next.assistant_msg_idx=offset+index-mergedOffset; + } + } + translated.push(next); + } + return translated; + }; + const hasIncomingToolCalls=Object.prototype.hasOwnProperty.call(incoming,'tool_calls'); + const currentCalls=translateCalls(currentSession.tool_calls,currentOffset,currentMessages.length); + const incomingCalls=translateCalls(incoming.tool_calls,incomingOffset,incomingMessages.length); + const merged={...currentSession,...incoming,messages:mergedMessages}; + if(hasIncomingToolCalls){ + const callIdentity=call=>String(call&&(call.tid||call.id||call.tool_call_id||call.tool_use_id||call.call_id)||'').trim(); + const callShapeIdentity=call=>{ + const index=call&&call.assistant_msg_idx; + const name=String(call&&call.name||'').trim(); + return isOffset(index)&&name?`${index}:${name}`:''; + }; + const incomingIds=new Set(incomingCalls.map(callIdentity).filter(Boolean)); + const toolCallsTruncated=incoming._tool_calls_truncated===true; + const incomingShapeCounts=new Map(); + const currentShapeCounts=new Map(); + if(toolCallsTruncated){ + for(const call of incomingCalls){ + if(callIdentity(call)) continue; + const shapeIdentity=callShapeIdentity(call); + if(shapeIdentity) incomingShapeCounts.set(shapeIdentity,(incomingShapeCounts.get(shapeIdentity)||0)+1); + } + for(const call of currentCalls){ + if(callIdentity(call)) continue; + const shapeIdentity=callShapeIdentity(call); + if(shapeIdentity) currentShapeCounts.set(shapeIdentity,(currentShapeCounts.get(shapeIdentity)||0)+1); + } + } + merged.tool_calls=[ + ...currentCalls.filter(call=>{ + const index=call.assistant_msg_idx; + if(!isOffset(index)) return false; + const identity=callIdentity(call); + if(!toolCallsTruncated){ + if(index>=incomingOffset-mergedOffset) return false; + return !identity||!incomingIds.has(identity); + } + if(identity) return !incomingIds.has(identity); + const shapeIdentity=callShapeIdentity(call); + const currentCount=currentShapeCounts.get(shapeIdentity)||0; + currentShapeCounts.set(shapeIdentity,currentCount-1); + return currentCount>(incomingShapeCounts.get(shapeIdentity)||0); + }), + ...incomingCalls, + ]; + } + else if(Object.prototype.hasOwnProperty.call(currentSession,'tool_calls')) merged.tool_calls=currentCalls; + merged._messages_offset=mergedOffset; + merged._messages_truncated=!!mergedOffset; + merged.messages=mergedMessages; + return merged; +} function attachLiveStream(activeSid, streamId, uploaded=[], options={}){ if(!activeSid||!streamId) return; @@ -2273,22 +2385,6 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){ const backendRecovery=/^the live worker stopped before this run finished\.?$/i.test(normalized); return !!(systemRecovery || backendRecovery); } - function _streamRecoveryControlMessage(m){ - if(!m||m.role==='tool') return false; - if(m.recovery_control===true) return true; - // Backward-compat ONLY for pre-marker persisted sessions: match the two - // fully-anchored synthetic recovery strings. Do NOT fall back to - // provider_details_label — a genuine "Response interrupted" card the user - // SHOULD see also carries the 'Interruption details' label, and filtering - // on it would drop a real interruption from the transcript (the inverse - // data-loss class flagged on the sibling #3300). Marker + strict text only. - const text=String(typeof msgContent==='function'?msgContent(m):(m.content||'')); - return _streamRecoveryControlMessageText(text); - } - function _filterRecoveryControlMessages(messages){ - if(!Array.isArray(messages)) return []; - return messages.filter((m)=>!_streamRecoveryControlMessage(m)); - } function _replaceMarkerOnlyAssistantWithStreamError(messages){ if(!Array.isArray(messages)) return false; const msg=[...messages].reverse().find(m=>m&&m.role==='assistant'); @@ -6081,7 +6177,33 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){ _clearAnchorProseIncrementalNode(); const isActiveSession=_isSessionCurrentPane(activeSid); const isSessionViewed=_isSessionActivelyViewed(activeSid); - const completedSession=d.session||{session_id:activeSid}; + const _currentTerminalSession= S.session&&S.session.session_id===activeSid ? S.session : null; + const _currentTerminalMessages=_currentTerminalSession&&Array.isArray(S.messages)?S.messages:[]; + const _hasTerminalSession=!!(d.session&&typeof d.session==='object'); + const _mergedTerminalSession=_hasTerminalSession&&_applyEmbeddedTerminalSession( + d.session, + _currentTerminalSession, + _currentTerminalMessages, + _oldestIdx, + ); + const _terminalToolCallsProvided=!!( + _mergedTerminalSession&& + Object.prototype.hasOwnProperty.call(d.session,'tool_calls') + ); + const completedSession=_mergedTerminalSession||( + _currentTerminalSession + ? {..._currentTerminalSession,messages:_currentTerminalMessages} + : {session_id:activeSid} + ); + const _terminalSessionSameSid=!!( + _currentTerminalSession&&completedSession.session_id===_currentTerminalSession.session_id + ); + if(_terminalSessionSameSid){ + completedSession.messages=_carryForwardEphemeralTurnFields( + _currentTerminalMessages, + completedSession.messages||[], + ); + } const completedSid=completedSession.session_id||activeSid; const completedMessageCount=completedSession.message_count != null ? completedSession.message_count @@ -6122,10 +6244,17 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){ const _prevCost=(S.session&&S.session.estimated_cost)||0; const _prevCacheRead=(S.session&&S.session.cache_read_tokens)||0; const _prevCacheWrite=(S.session&&S.session.cache_write_tokens)||0; - S.session=d.session;S.messages=_carryForwardEphemeralTurnFields(S.messages||[], d.session.messages||[]);if(typeof _messagesTruncated!=='undefined')_messagesTruncated=!!d.session._messages_truncated; - // #4720: reset _oldestIdx (full-load symmetry; keeps the #4613 anchor aligned). - if(typeof _oldestIdx!=='undefined')_oldestIdx=d.session._messages_offset||0; - S.messages=_filterRecoveryControlMessages(S.messages || []); + S.session=completedSession; + if(_terminalSessionSameSid){ + S.messages=completedSession.messages; + } else if(d.session){ + S.messages=_carryForwardEphemeralTurnFields(_currentTerminalMessages,completedSession.messages||[]); + } + if(_mergedTerminalSession){ + if(typeof _messagesTruncated!=='undefined')_messagesTruncated=!!completedSession._messages_truncated; + if(typeof _oldestIdx!=='undefined')_oldestIdx=completedSession._messages_offset||0; + } + completedSession.messages=S.messages; if(typeof _hydrateTodosFromSession==='function') _hydrateTodosFromSession(S.session); if(typeof clearVisibleMessageRowCache==='function') clearVisibleMessageRowCache(); if(S.session&&S.session.session_id){ @@ -6136,7 +6265,7 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){ if( window._compressionUi&&window._compressionUi.automatic&& window._compressionUi.sessionId===activeSid&& - d.session&&d.session.session_id + completedSession&&completedSession.session_id ){ if(window._compressionUi.phase==='running'){ // Turn completed (done event) but the compression UI is still in @@ -6149,7 +6278,7 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){ if(typeof clearCompressionUi==='function') clearCompressionUi(); else window._compressionUi=null; } else { - window._compressionUi={...window._compressionUi, sessionId:d.session.session_id}; + window._compressionUi={...window._compressionUi, sessionId:completedSession.session_id}; } } // Find the last assistant message once for both reasoning persistence and timestamp @@ -6226,9 +6355,11 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){ const hasTu=Array.isArray(m.content)&&m.content.some(p=>p&&p.type==='tool_use'); return hasTc||hasPartialTc||hasTu; }); - if(!hasMessageToolMetadata&&d.session.tool_calls&&d.session.tool_calls.length){ - S.toolCalls=d.session.tool_calls.map(tc=>tc); - S.toolCalls=_mergeSettledToolCallsWithLiveMetadata(d.session.tool_calls); + if(_terminalToolCallsProvided){ + S.toolCalls=_mergeSettledToolCallsWithLiveMetadata(completedSession.tool_calls); + } else if(!hasMessageToolMetadata&&completedSession.tool_calls&&completedSession.tool_calls.length){ + S.toolCalls=completedSession.tool_calls.map(tc=>tc); + S.toolCalls=_mergeSettledToolCallsWithLiveMetadata(completedSession.tool_calls); } else { if(hasMessageToolMetadata) S._settledLiveToolMetadata=S.toolCalls.map(tc=>({...tc,done:true})); S.toolCalls=hasMessageToolMetadata?[]:S.toolCalls.map(tc=>({...tc,done:true})); @@ -6310,6 +6441,9 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){ if(!lastAsst&&d.session&&Array.isArray(d.session.messages)){ lastAsst=[...d.session.messages].reverse().find(m=>m&&m.role==='assistant')||null; } + if(!lastAsst&&completedSession&&Array.isArray(completedSession.messages)){ + lastAsst=[...completedSession.messages].reverse().find(m=>m&&m.role==='assistant')||null; + } if(isActiveSession&&_pendingGoalContinuation&&typeof queueSessionMessage==='function'){ const _goalNext=_pendingGoalContinuation; _pendingGoalContinuation=null; @@ -6516,8 +6650,23 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){ const _extensionErrorType=(d.type==='cancelled'||d.type==='interrupted')?'turn:cancel':'turn:error'; 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)); + const incomingSessionSid=d.session&&typeof d.session==='object' ? String(d.session.session_id||'') : ''; + const continuationSid=d.new_session_id||d.continuation_session_id||''; + const eventMatchesCurrent=!!(currentSid&&( + eventSid===currentSid||continuationSid===currentSid||incomingSessionSid===currentSid + )); + const _currentTerminalSession=S.session; + const _currentTerminalMessages=Array.isArray(S.messages)?S.messages:[]; + const _terminalSessionSameSid=!!( + eventMatchesCurrent&&d.session&&_currentTerminalSession&& + (!d.session.session_id||_currentTerminalSession.session_id===d.session.session_id) + ); + const _terminalSession=eventMatchesCurrent&&_applyEmbeddedTerminalSession( + d.session, + _currentTerminalSession, + _currentTerminalMessages, + _oldestIdx, + ); if(eventMatchesCurrent){ _flushReasoningToAnchor(); _applyToAnchor('apperror',{ @@ -6558,11 +6707,23 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){ 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; - const _nextMsgs3018=(d.session.messages||[]).filter(m=>m&&m.role); - _attachProjectedAnchorSceneToLastAssistant(_nextMsgs3018); - S.messages=_carryForwardEphemeralTurnFields(S.messages||[], _nextMsgs3018); + } else if(d.session&&typeof d.session==='object'&&_terminalSession){ + if(_terminalSessionSameSid){ + _terminalSession.messages=_carryForwardEphemeralTurnFields( + _currentTerminalMessages, + _terminalSession.messages||[], + ); + } + if(typeof _messagesTruncated!=='undefined')_messagesTruncated=!!_terminalSession._messages_truncated; + if(typeof _oldestIdx!=='undefined')_oldestIdx=_terminalSession._messages_offset||0; + S.session=_terminalSession; + if(_terminalSessionSameSid){ + S.messages=_terminalSession.messages; + } else { + S.messages=_carryForwardEphemeralTurnFields(_currentTerminalMessages,_terminalSession.messages||[]); + } + _terminalSession.messages=S.messages; + _attachProjectedAnchorSceneToLastAssistant(S.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); @@ -6596,13 +6757,16 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){ (async()=>{ if(await _restoreSettledSession(source, {preserveVisibleOnShorterTerminalSnapshot:true})) return; if(S.session&&S.session.session_id===activeSid){ - S.messages=_filterRecoveryControlMessages(S.messages||[]); - _markSessionViewed(activeSid, S.messages.length); + _markSessionViewed(activeSid, S.session.message_count != null ? S.session.message_count : S.messages.length); renderMessages({preserveScroll:true}); } })(); } else { - _markSessionViewed((S.session&&S.session.session_id)||activeSid, S.messages.length); + const _apperrorViewedSid=(S.session&&S.session.session_id)||activeSid; + const _apperrorViewedCount=S.session&&S.session.message_count != null + ? S.session.message_count + : S.messages.length; + _markSessionViewed(_apperrorViewedSid, _apperrorViewedCount); renderMessages({preserveScroll:true}); } }else if(typeof trackBackgroundError==='function'){ @@ -6784,12 +6948,13 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){ if(S.session&&S.session.session_id===activeSid){ S.activeStreamId=null; } - const _applyCancelSessionPayload=(sessionPayload)=>{ - if(!sessionPayload||typeof sessionPayload!=='object'||!S.session||S.session.session_id!==activeSid) return false; - // Belt-and-suspenders: the embedded cancel snapshot must be for THIS session. - // The GET path guarantees it via the URL; the embedded path via the stream→session - // binding — but reject a mismatched id so a stray payload can't overwrite the view. - if(sessionPayload.session_id&&sessionPayload.session_id!==activeSid) return false; + const _applyCancelSessionPayload=(sessionPayload, fullSnapshot=false)=>{ + if(!sessionPayload||typeof sessionPayload!=='object'||!S.session) return false; + if(fullSnapshot&&!Array.isArray(sessionPayload.messages)) return false; + const payloadSid=String(sessionPayload.session_id||''); + const activeSessionSid=String(activeSid||''); + const currentSessionSid=String(S.session.session_id||''); + if(!payloadSid||payloadSid!==activeSessionSid||currentSessionSid!==activeSessionSid) return false; // Capture follow-intent BEFORE replacing S.messages: a reader who was // following the live stream when it got cancelled/reconnected must land at // the bottom (where the cancellation notice renders), not be stranded at a @@ -6801,13 +6966,30 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){ && !((typeof _isMessageReaderUnpinned==='function') ? _isMessageReaderUnpinned() : (typeof _messageUserUnpinned!=='undefined' && _messageUserUnpinned)); - S.session=sessionPayload; - const _nextMsgs3018=(sessionPayload.messages||[]).filter(m=>m&&m.role); - _attachProjectedAnchorSceneToLastAssistant(_nextMsgs3018); - S.messages=_carryForwardEphemeralTurnFields(S.messages||[], _nextMsgs3018); + const _currentTerminalSession=S.session; + const _currentTerminalMessages=Array.isArray(S.messages)?S.messages:[]; + const _terminalSession=fullSnapshot + ? {...sessionPayload,messages:sessionPayload.messages} + : _applyEmbeddedTerminalSession(sessionPayload,_currentTerminalSession,_currentTerminalMessages,_oldestIdx); + if(!_terminalSession) return false; + S.activeStreamId=null; + _terminalSession.messages=_carryForwardEphemeralTurnFields( + _currentTerminalMessages, + _terminalSession.messages||[], + ); + S.session=_terminalSession; + S.messages=_terminalSession.messages; + _terminalSession.messages=S.messages; + if(typeof _messagesTruncated!=='undefined')_messagesTruncated=!!_terminalSession._messages_truncated; + if(typeof _oldestIdx!=='undefined')_oldestIdx=_terminalSession._messages_offset||0; + 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); + } + _attachProjectedAnchorSceneToLastAssistant(S.messages); if(typeof _hydrateTodosFromSession==='function') _hydrateTodosFromSession(S.session); clearLiveToolCards();if(!assistantText)removeThinking(); - _markSessionViewed(activeSid, sessionPayload.message_count ?? S.messages.length); + _markSessionViewed(_terminalSession.session_id||activeSid, _terminalSession.message_count ?? S.messages.length); renderMessages({preserveScroll:true}); if(_wasFollowingAtCancel && typeof scrollToBottom==='function') scrollToBottom(); return true; @@ -6821,12 +7003,10 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){ _setActivePaneIdleIfOwner(); (async()=>{ try{ - if(_applyCancelSessionPayload(_cancelSessionPayload)) return; - // Fetch latest session from server to get accurate message list (includes cancel status) - // This ensures messages stay in sync with server, fixing race condition where local - // "*Task cancelled.*" message gets lost when done event overwrites S.messages + if(_cancelSessionPayload&&_applyCancelSessionPayload(_cancelSessionPayload)) return; + // Fetch latest session from server to get accurate message list (includes cancel status). const data=await api(`/api/session?session_id=${encodeURIComponent(activeSid)}`); - if(data&&data.session) _applyCancelSessionPayload(data.session); + if(data&&data.session) _applyCancelSessionPayload(data.session,true); }catch(_){ // Fallback to local cancel message if API fails if(S.session&&S.session.session_id===activeSid){ @@ -6946,9 +7126,9 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){ S.activeStreamId=null; clearLiveToolCards();if(!assistantText)removeThinking(); S.session=session; - const _nextMsgs3018=(session.messages||[]).filter(m=>m&&m.role); + const _nextMsgs3018=Array.isArray(session.messages)?session.messages:[]; const _currentMessages=Array.isArray(S.messages)?S.messages:[]; - const _currentVisibleMessages=_filterRecoveryControlMessages(_currentMessages || []); + const _currentVisibleMessages=_currentMessages; const _stagedMessages=_carryForwardEphemeralTurnFields(_currentMessages, _nextMsgs3018); const _currentVisibleEndsWithTerminalMarker=( _currentVisibleMessages.length>0 && @@ -6968,7 +7148,13 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){ const _resolvedMessages=_preserveCurrentTranscript ? [..._stagedMessages,..._currentVisibleMessages.slice(_stagedMessages.length)] : _stagedMessages; - S.messages=_filterRecoveryControlMessages(_resolvedMessages || []); + S.messages=_resolvedMessages || []; + session.messages=S.messages; + const _restoredOffset=Number.isInteger(session._messages_offset)&&session._messages_offset>=0 + ? session._messages_offset + : 0; + if(typeof _oldestIdx!=='undefined') _oldestIdx=_restoredOffset; + if(typeof _messagesTruncated!=='undefined') _messagesTruncated=!!session._messages_truncated; _attachProjectedAnchorSceneToLastAssistant(S.messages); if(typeof _hydrateTodosFromSession==='function') _hydrateTodosFromSession(S.session); if(S.session&&S.session.session_id){ @@ -7005,9 +7191,7 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){ renderSessionList(); _setActivePaneIdleIfOwner(); return returnStatus?'restored':true; - }catch(_){ - return returnStatus?'error':false; - } + }catch(_){ return returnStatus?'error':false; } } function _handleStreamError(source){ diff --git a/static/sessions.js b/static/sessions.js index 0c584aec1b1..fe884fe92d1 100644 --- a/static/sessions.js +++ b/static/sessions.js @@ -7180,7 +7180,8 @@ function upsertActiveSessionForLocalTurn({title='', messageCount=0, timestampMs= if(!S.session||!S.session.session_id) return; const sid=S.session.session_id; const nowSec=Math.floor((Number(timestampMs)||Date.now())/1000); - const localCount=Array.isArray(S.messages)?S.messages.length:0; + const loadedOffset=typeof _messagesTruncated!=='undefined'&&_messagesTruncated&&typeof _oldestIdx==='number'&&Number.isSafeInteger(_oldestIdx)&&_oldestIdx>=0?_oldestIdx:0; + const localCount=loadedOffset+(Array.isArray(S.messages)?S.messages.length:0); const count=Math.max(Number(S.session.message_count||0),Number(messageCount||0),localCount,1); S.session.message_count=count; S.session.last_message_at=nowSec; diff --git a/tests/test_auto_compression_card.py b/tests/test_auto_compression_card.py index 837ff474216..7b858d0e34c 100644 --- a/tests/test_auto_compression_card.py +++ b/tests/test_auto_compression_card.py @@ -971,7 +971,7 @@ def test_auto_compression_card_survives_compression_session_rotation(): src = _read("static/messages.js") assert "window._compressionUi.sessionId===activeSid" in src - assert "sessionId:d.session.session_id" in src + assert "sessionId:completedSession.session_id" in src def test_preserved_task_list_marker_is_detected_case_insensitively(): diff --git a/tests/test_auto_compression_terminal_failure.py b/tests/test_auto_compression_terminal_failure.py index 6b91354d750..44a235d26fc 100644 --- a/tests/test_auto_compression_terminal_failure.py +++ b/tests/test_auto_compression_terminal_failure.py @@ -294,9 +294,10 @@ def test_compression_exhausted_apperror_clears_reference_ui_and_labels_error(): 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 + assert "const incomingSessionSid=d.session&&typeof d.session==='object' ? String(d.session.session_id||'') : '';" in block + assert "const continuationSid=d.new_session_id||d.continuation_session_id||'';" in block + assert "if(d.session&&typeof d.session==='object'&&_terminalSession)" in block + assert "S.session=_terminalSession;" in block def test_apperror_matches_only_current_or_continuation_session_for_background_errors(): @@ -308,8 +309,10 @@ def test_apperror_matches_only_current_or_continuation_session_for_background_er 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 + assert "const incomingSessionSid=d.session&&typeof d.session==='object' ? String(d.session.session_id||'') : '';" in block + assert "const continuationSid=d.new_session_id||d.continuation_session_id||'';" in block + assert "const eventMatchesCurrent=!!(currentSid&&(" in block + assert "eventSid===currentSid||continuationSid===currentSid||incomingSessionSid===currentSid" in block def test_apperror_payload_enriched_before_enqueue(tmp_path, monkeypatch): diff --git a/tests/test_cancelled_turn_status.py b/tests/test_cancelled_turn_status.py index bfeebe5beff..d7205288e44 100644 --- a/tests/test_cancelled_turn_status.py +++ b/tests/test_cancelled_turn_status.py @@ -216,16 +216,17 @@ def test_frontend_cancel_prefers_embedded_session_payload(self): assert start != -1 and end != -1, "cancel handler not found" block = src[start:end] - assert "const _applyCancelSessionPayload=(sessionPayload)=>" in block + assert "const _applyCancelSessionPayload=(sessionPayload, fullSnapshot=false)=>" in block assert "const _cancelSessionPayload=_cancelData&&typeof _cancelData.session==='object'?_cancelData.session:null;" in block - assert "if(_applyCancelSessionPayload(_cancelSessionPayload)) return;" in block + assert "if(_cancelSessionPayload&&_applyCancelSessionPayload(_cancelSessionPayload)) return;" in block assert "const data=await api(`/api/session?session_id=${encodeURIComponent(activeSid)}`);" in block - assert block.index("if(_applyCancelSessionPayload(_cancelSessionPayload)) return;") < block.index("const data=await api("), ( + assert "if(data&&data.session) _applyCancelSessionPayload(data.session,true);" in block + assert block.index("if(_cancelSessionPayload&&_applyCancelSessionPayload(_cancelSessionPayload)) return;") < block.index("const data=await api("), ( "Cancel handler must apply the terminal SSE session payload before falling back " "to /api/session so captured _partial reasoning/tool rows are visible immediately." ) - def test_worker_cancel_events_do_not_embed_session_payload(self): + def test_worker_cancel_events_embed_bounded_session_payload(self): src = _read("api/streaming.py") worker_start = src.find("def _run_agent_streaming(") cancel_stream_start = src.find("def cancel_stream(", worker_start) @@ -234,6 +235,6 @@ def test_worker_cancel_events_do_not_embed_session_payload(self): cancel_stream_block = src[cancel_stream_start:] assert "_cancel_event_payload('Cancelled by user', s)" not in worker_block - assert "_cancel_event_payload('Cancelled by user', session=" not in worker_block - assert "None if ephemeral else s" not in worker_block + assert "_cancel_event_payload('Cancelled by user', session=" in worker_block + assert "None if ephemeral else s" in worker_block assert "_cancel_event_payload('Cancelled by user', session=_cancel_session_payload)" in cancel_stream_block diff --git a/tests/test_compression_phantom_barrier.py b/tests/test_compression_phantom_barrier.py index c1cf9038826..580daf2b608 100644 --- a/tests/test_compression_phantom_barrier.py +++ b/tests/test_compression_phantom_barrier.py @@ -58,8 +58,8 @@ def _revert_running_cleanup(source: str) -> str: old_branch = source[branch_start : else_close + 1] assert "clearCompressionUi()" in old_branch - assert "sessionId:d.session.session_id" in old_branch.replace(" ", "") - reverted = "window._compressionUi={...window._compressionUi, sessionId:d.session.session_id};" + assert "sessionId:completedSession.session_id" in old_branch.replace(" ", "") + reverted = "window._compressionUi={...window._compressionUi, sessionId:completedSession.session_id};" return source[:branch_start] + reverted + source[else_close + 1 :] @@ -132,6 +132,8 @@ class MockEventSource { S.session = {session_id: activeSid, messages: []}; S.messages = []; S.toolCalls = []; + _oldestIdx = 0; + _messagesTruncated = false; S.activeStreamId = streamId; S.busy = true; window._compressionUi = null; diff --git a/tests/test_extension_turn_lifecycle.py b/tests/test_extension_turn_lifecycle.py index 4a1f5b71eaf..fa9779abc3e 100644 --- a/tests/test_extension_turn_lifecycle.py +++ b/tests/test_extension_turn_lifecycle.py @@ -75,6 +75,8 @@ class MockEventSource { playNotificationSound = () => {}; sendBrowserNotification = () => {}; api = async () => ({}); +_oldestIdx = 0; +_messagesTruncated = false; """ @@ -113,6 +115,9 @@ class MockEventSource { stream_id: streamId, session: { session_id: activeSid, + _messages_offset: 0, + _messages_truncated: false, + message_count: 1, messages: [{role: 'assistant', content: 'settled-done'}], tool_calls: [], }, @@ -125,6 +130,9 @@ class MockEventSource { session_id: activeSid, session: { session_id: activeSid, + _messages_offset: 0, + _messages_truncated: false, + message_count: 1, messages: [{role: 'assistant', content: 'settled-error'}], }, }); @@ -136,6 +144,9 @@ class MockEventSource { session_id: activeSid, session: { session_id: activeSid, + _messages_offset: 0, + _messages_truncated: false, + message_count: 1, messages: [{role: 'assistant', content: 'settled-interrupted'}], }, }); @@ -146,6 +157,9 @@ class MockEventSource { session_id: activeSid, session: { session_id: activeSid, + _messages_offset: 0, + _messages_truncated: false, + message_count: 1, messages: [{role: 'assistant', content: 'settled-cancel'}], }, }); diff --git a/tests/test_issue2655_frontend.py b/tests/test_issue2655_frontend.py index b5da3959d00..012299b8be8 100644 --- a/tests/test_issue2655_frontend.py +++ b/tests/test_issue2655_frontend.py @@ -24,7 +24,7 @@ def test_workspace_artifacts_tab_collects_session_files_and_previews_them(): assert "panel.dataset.activeTab = _workspacePanelActiveTab" in WORKSPACE_JS assert "renderSessionArtifacts();" in SESSIONS_JS assert "typeof scheduleRenderSessionArtifacts==='function'" in MESSAGES_JS - assert "S.toolCalls=d.session.tool_calls.map" in MESSAGES_JS + assert "S.toolCalls=completedSession.tool_calls.map" in MESSAGES_JS assert ".workspace-artifact-item" in STYLE_CSS diff --git a/tests/test_issue3929_process_wakeup_pause.py b/tests/test_issue3929_process_wakeup_pause.py index 88a9a9af5d5..722898ba4af 100644 --- a/tests/test_issue3929_process_wakeup_pause.py +++ b/tests/test_issue3929_process_wakeup_pause.py @@ -149,6 +149,22 @@ def run_conversation(self, **kwargs): } +def _payload_builder_that_fails_at_compact(builder): + def _fail_payload(session): + original_compact = session.compact + + def _fail_compact(): + raise RuntimeError("compact failed") + + session.compact = _fail_compact + try: + return builder(session) + finally: + session.compact = original_compact + + return _fail_payload + + class _FakeCredentialPoolEntry: def __init__(self, payload): self._payload = dict(payload) @@ -2233,7 +2249,7 @@ def __iter__(self): session.save() models.SESSIONS[session_id] = session - original_payload = streaming._session_payload_with_full_messages + original_payload = streaming._redacted_terminal_session_payload payload_calls = {"count": 0} def _payload_and_cancel_after_success_commit(*args, **kwargs): @@ -2241,7 +2257,7 @@ def _payload_and_cancel_after_success_commit(*args, **kwargs): config.CANCEL_FLAGS[stream_id].set() return original_payload(*args, **kwargs) - monkeypatch.setattr(streaming, "_session_payload_with_full_messages", _payload_and_cancel_after_success_commit) + monkeypatch.setattr(streaming, "_redacted_terminal_session_payload", _payload_and_cancel_after_success_commit) gateway_chat._run_gateway_chat_streaming( session_id, @@ -2367,7 +2383,7 @@ def test_streaming_post_save_cancel_after_success_commit_emits_done(tmp_path, mo session.save() models.SESSIONS[session_id] = session - original_payload = streaming._session_payload_with_full_messages + original_payload = streaming._redacted_terminal_session_payload payload_calls = {"count": 0} def _payload_and_cancel_after_success_commit(*args, **kwargs): @@ -2375,7 +2391,7 @@ def _payload_and_cancel_after_success_commit(*args, **kwargs): config.CANCEL_FLAGS[stream_id].set() return original_payload(*args, **kwargs) - monkeypatch.setattr(streaming, "_session_payload_with_full_messages", _payload_and_cancel_after_success_commit) + monkeypatch.setattr(streaming, "_redacted_terminal_session_payload", _payload_and_cancel_after_success_commit) with mock.patch.object(streaming, "_get_ai_agent", return_value=_SuccessfulAgent), \ mock.patch.object(streaming, "resolve_model_provider", return_value=("test-model", "test-provider", None)), \ @@ -2419,7 +2435,7 @@ def test_streaming_no_pause_post_save_cancel_after_success_commit_emits_done(tmp session.save() models.SESSIONS[session_id] = session - original_payload = streaming._session_payload_with_full_messages + original_payload = streaming._redacted_terminal_session_payload payload_calls = {"count": 0} def _payload_and_cancel_after_success_commit(*args, **kwargs): @@ -2427,7 +2443,7 @@ def _payload_and_cancel_after_success_commit(*args, **kwargs): config.CANCEL_FLAGS[stream_id].set() return original_payload(*args, **kwargs) - monkeypatch.setattr(streaming, "_session_payload_with_full_messages", _payload_and_cancel_after_success_commit) + monkeypatch.setattr(streaming, "_redacted_terminal_session_payload", _payload_and_cancel_after_success_commit) with mock.patch.object(streaming, "_get_ai_agent", return_value=_SuccessfulAgent), \ mock.patch.object(streaming, "resolve_model_provider", return_value=("test-model", "test-provider", None)), \ @@ -2458,6 +2474,182 @@ def _payload_and_cancel_after_success_commit(*args, **kwargs): assert "cancel" not in queued_events +def test_streaming_done_survives_terminal_payload_failure(tmp_path, monkeypatch): + stream_id = "streaming-done-payload-failure" + session_id = "streaming_done_payload_failure" + stream_queue = queue.Queue() + config.STREAMS[stream_id] = stream_queue + session = Session( + session_id=session_id, + workspace=str(tmp_path), + model="test-model", + model_provider="test-provider", + messages=[{"role": "user", "content": "before"}], + context_messages=[{"role": "user", "content": "before"}], + active_stream_id=stream_id, + pending_user_message="hello", + pending_user_source="webui", + ) + session.save() + models.SESSIONS[session_id] = session + + monkeypatch.setattr( + streaming, + "_redacted_terminal_session_payload", + _payload_builder_that_fails_at_compact(streaming._redacted_terminal_session_payload), + ) + with mock.patch.object(streaming, "_get_ai_agent", return_value=_SuccessfulAgent), \ + mock.patch.object(streaming, "resolve_model_provider", return_value=("test-model", "test-provider", None)), \ + mock.patch("api.config._resolve_cli_toolsets", return_value=[]): + streaming._run_agent_streaming( + session_id=session_id, + msg_text="hello", + model="test-model", + model_provider="test-provider", + workspace=str(tmp_path), + stream_id=stream_id, + ) + + events = list(stream_queue.queue) + done_payload = next(item[1] for item in events if item[0] == "done") + assert "session" not in done_payload + assert not any(item[0] == "apperror" for item in events) + + +def test_ephemeral_done_does_not_embed_parent_transcript(tmp_path, monkeypatch): + stream_id = "streaming-ephemeral-done-bounded" + session_id = "streaming_ephemeral_done_bounded" + stream_queue = queue.Queue() + config.STREAMS[stream_id] = stream_queue + messages = [{"role": "user", "content": f"row-{index}"} for index in range(2000)] + session = Session( + session_id=session_id, + workspace=str(tmp_path), + model="test-model", + model_provider="test-provider", + messages=messages, + context_messages=list(messages), + active_stream_id=stream_id, + pending_user_message="What is this?", + pending_user_source="webui", + ) + session.save() + models.SESSIONS[session_id] = session + + with mock.patch.object(streaming, "_get_ai_agent", return_value=_SuccessfulAgent), \ + mock.patch.object(streaming, "resolve_model_provider", return_value=("test-model", "test-provider", None)), \ + mock.patch("api.config._resolve_cli_toolsets", return_value=[]): + streaming._run_agent_streaming( + session_id=session_id, + msg_text="What is this?", + model="test-model", + model_provider="test-provider", + workspace=str(tmp_path), + stream_id=stream_id, + ephemeral=True, + ) + + done_payload = next(item[1] for item in list(stream_queue.queue) if item[0] == "done") + assert done_payload["answer"] == "Stream reply" + assert done_payload["ephemeral"] is True + assert done_payload["session"] == {"session_id": session_id} + assert "messages" not in done_payload["session"] + assert "row-1999" not in json.dumps(done_payload) + + +def test_gateway_done_survives_terminal_payload_failure(tmp_path, monkeypatch): + stream_id = "gateway-done-payload-failure" + session_id = "gateway_done_payload_failure" + stream_queue = queue.Queue() + config.STREAMS[stream_id] = stream_queue + monkeypatch.setattr(gateway_chat, "RunJournalWriter", lambda *_args, **_kwargs: None) + monkeypatch.setattr(gateway_chat, "gateway_approval_unavailable_reason", lambda *_args, **_kwargs: None) + monkeypatch.setattr(config, "get_config", lambda: {"webui_gateway_base_url": "http://gateway.test"}) + + class _GatewayResponse: + def __enter__(self): + return self + + def __exit__(self, *_exc): + return False + + def __iter__(self): + return iter([ + b'data: {"choices":[{"delta":{"content":"Gateway reply"}}]}\n', + b"data: [DONE]\n", + ]) + + monkeypatch.setattr(gateway_chat.urllib.request, "urlopen", lambda *_args, **_kwargs: _GatewayResponse()) + monkeypatch.setattr( + streaming, + "_redacted_terminal_session_payload", + _payload_builder_that_fails_at_compact(streaming._redacted_terminal_session_payload), + ) + session = Session( + session_id=session_id, + workspace=str(tmp_path), + model="test-model", + model_provider="test-provider", + messages=[], + context_messages=[], + active_stream_id=stream_id, + pending_user_message="hello", + pending_user_source="webui", + ) + session.save() + models.SESSIONS[session_id] = session + + gateway_chat._run_gateway_chat_streaming( + session_id, + "hello", + "test-model", + str(tmp_path), + stream_id, + model_provider="test-provider", + ) + + events = list(stream_queue.queue) + done_payload = next(item[1] for item in events if item[0] == "done") + assert "session" not in done_payload + assert not any(item[0] in {"apperror", "gateway_error"} for item in events) + + +def test_gateway_terminal_error_survives_terminal_payload_failure(tmp_path, monkeypatch): + stream_id = "gateway-error-payload-failure" + session_id = "gateway_error_payload_failure" + session = Session( + session_id=session_id, + workspace=str(tmp_path), + model="test-model", + model_provider="test-provider", + messages=[], + context_messages=[], + active_stream_id=stream_id, + pending_user_message="hello", + pending_user_source="webui", + ) + session.save() + models.SESSIONS[session_id] = session + monkeypatch.setattr( + streaming, + "_redacted_terminal_session_payload", + _payload_builder_that_fails_at_compact(streaming._redacted_terminal_session_payload), + ) + + payload = gateway_chat._settle_gateway_terminal_error( + session_id, + stream_id, + str(tmp_path), + "test-model", + "test-provider", + "gateway exploded", + ) + + assert payload["type"] == "error" + assert payload["session_id"] == session_id + assert "session" not in payload + + def test_stale_credential_empty_process_wakeup_still_records_pause(tmp_path): session = Session( session_id="wakeup_pause_stale", diff --git a/tests/test_issue4720_done_scroll_jump_first_message.py b/tests/test_issue4720_done_scroll_jump_first_message.py index 790fb769b66..dcb56fbf397 100644 --- a/tests/test_issue4720_done_scroll_jump_first_message.py +++ b/tests/test_issue4720_done_scroll_jump_first_message.py @@ -27,31 +27,28 @@ def _compact(text: str) -> str: def test_done_handler_resets_oldest_idx_from_payload_offset(): """The done handler must reset _oldestIdx alongside _messagesTruncated.""" compact = _compact(MESSAGES_JS) - # The truncated flag and the offset reset must be wired off the SAME done - # payload (d.session), mirroring sessions.js / ui.js full-load paths. - assert "_messagesTruncated=!!d.session._messages_truncated" in compact, ( + # Rotated sessions reset both cursors from the installed terminal payload. + assert "_messagesTruncated=!!completedSession._messages_truncated" in compact, ( "done handler should still set _messagesTruncated from the done payload" ) - assert "_oldestIdx=d.session._messages_offset||0" in compact, ( + assert "_oldestIdx=completedSession._messages_offset||0" in compact, ( "#4720: done handler must reset _oldestIdx from the done payload offset " "so the absolute scroll anchor stays valid after the render-window expansion" ) -def test_done_handler_oldest_idx_reset_is_guarded_and_ordered_before_filter(): - """Reset must be typeof-guarded and happen before the messages are re-filtered/rendered.""" - compact = _compact(MESSAGES_JS) - assert "if(typeof_oldestIdx!=='undefined')_oldestIdx=d.session._messages_offset||0" in compact, ( +def test_done_handler_oldest_idx_reset_is_guarded_for_terminal_windows(): + """Terminal window reconciliation updates the raw-coordinate cursor.""" + done_start = MESSAGES_JS.index("source.addEventListener('done'") + done_end = MESSAGES_JS.index("source.addEventListener('stream_end'", done_start) + done_block = MESSAGES_JS[done_start:done_end] + compact = _compact(done_block) + assert "if(typeof_oldestIdx!=='undefined')_oldestIdx=completedSession._messages_offset||0" in compact, ( "_oldestIdx reset should be typeof-guarded like _messagesTruncated" ) - # The reset must precede _filterRecoveryControlMessages (which precedes the - # done-path renderMessages), so the anchor coordinate system is correct when - # the transcript is rebuilt. - reset_idx = compact.index("_oldestIdx=d.session._messages_offset||0") - filter_idx = compact.index("S.messages=_filterRecoveryControlMessages") - assert reset_idx < filter_idx, ( - "_oldestIdx must be reset before the done-path re-filter/render" - ) + cursor_idx = compact.index("_oldestIdx=completedSession._messages_offset||0") + assert "_filterRecoveryControlMessages" not in compact + assert cursor_idx < compact.index("completedSession.messages=S.messages") def test_oldest_idx_reset_matches_full_load_offset_semantics(): diff --git a/tests/test_issue5224_terminal_error_transcript_preserve.py b/tests/test_issue5224_terminal_error_transcript_preserve.py index 29ac384de1a..0262ca1c467 100644 --- a/tests/test_issue5224_terminal_error_transcript_preserve.py +++ b/tests/test_issue5224_terminal_error_transcript_preserve.py @@ -57,8 +57,6 @@ const helpers = [ "_isMarkerOnlyAssistantMessage", "_streamRecoveryControlMessageText", - "_streamRecoveryControlMessage", - "_filterRecoveryControlMessages", "_replaceMarkerOnlyAssistantWithStreamError", "_messageIdentityKey", "_isHistoricalAnchorActivityScene", diff --git a/tests/test_issue6751_api_content_agent_replay.py b/tests/test_issue6751_api_content_agent_replay.py index e78850cd7fc..736075391d0 100644 --- a/tests/test_issue6751_api_content_agent_replay.py +++ b/tests/test_issue6751_api_content_agent_replay.py @@ -877,11 +877,13 @@ def test_issue6751_ephemeral_terminal_sse_projects_agent_messages(monkeypatch): } ], ) + empty_payload = _ephemeral_session_payload("empty-sid", []) assert payload == { "session_id": "ephemeral-sid", "messages": [{"role": "assistant", "content": "visible"}], } + assert empty_payload == {"session_id": "empty-sid"} def test_issue6751_schema_scrubber_preserves_tool_argument_business_payload(monkeypatch): diff --git a/tests/test_issue856_background_completion_unread.py b/tests/test_issue856_background_completion_unread.py index 0368a108c98..677a38e31f9 100644 --- a/tests/test_issue856_background_completion_unread.py +++ b/tests/test_issue856_background_completion_unread.py @@ -64,7 +64,8 @@ def test_background_completion_unread_uses_explicit_marker_not_message_delta(): def test_background_done_sets_marker_when_session_not_actively_viewed(): done_block = _done_block() assert "const isSessionViewed=_isSessionActivelyViewed(activeSid);" in done_block - assert "const completedSession=d.session||{session_id:activeSid};" in done_block + assert "const _mergedTerminalSession=_hasTerminalSession&&_applyEmbeddedTerminalSession(" in done_block + assert "const completedSession=_mergedTerminalSession||(" in done_block assert "const completedSid=completedSession.session_id||activeSid;" in done_block assert "const completedMessageCount=completedSession.message_count != null" in done_block assert "if(!isSessionViewed && typeof _markSessionCompletionUnread==='function')" in done_block @@ -437,7 +438,7 @@ def test_hidden_active_done_still_updates_current_pane_but_not_read_state(): active_const_idx = done_block.find("const isActiveSession=_isSessionCurrentPane(activeSid);") viewed_const_idx = done_block.find("const isSessionViewed=_isSessionActivelyViewed(activeSid);") active_guard_idx = done_block.find("if(isActiveSession){", viewed_const_idx) - session_update_idx = done_block.find("S.session=d.session", active_guard_idx) + session_update_idx = done_block.find("S.session=completedSession", active_guard_idx) render_idx = done_block.find("renderMessages(", active_guard_idx) load_dir_idx = done_block.find("preservePreview", active_guard_idx) mark_viewed_idx = done_block.find("if(isSessionViewed) _markSessionViewed(completedSid", active_guard_idx) diff --git a/tests/test_live_activity_timeline.py b/tests/test_live_activity_timeline.py index 1a0e03c7d04..fb44241ccec 100644 --- a/tests/test_live_activity_timeline.py +++ b/tests/test_live_activity_timeline.py @@ -140,7 +140,7 @@ def test_reattach_normalizes_live_activity_group_placement_by_burst_anchor(): def test_done_handler_preserves_live_tool_burst_metadata_for_settled_render(): assert "function _mergeSettledToolCallsWithLiveMetadata(rawCalls)" in MESSAGES_JS assert "activityBurstId" in MESSAGES_JS - assert "S.toolCalls=_mergeSettledToolCallsWithLiveMetadata(d.session.tool_calls);" in MESSAGES_JS + assert "S.toolCalls=_mergeSettledToolCallsWithLiveMetadata(completedSession.tool_calls);" in MESSAGES_JS assert "S.toolCalls=_mergeSettledToolCallsWithLiveMetadata(session.tool_calls||[]);" in MESSAGES_JS diff --git a/tests/test_live_stream_ux.py b/tests/test_live_stream_ux.py index 04ea28ef862..e7571364fea 100644 --- a/tests/test_live_stream_ux.py +++ b/tests/test_live_stream_ux.py @@ -9,10 +9,8 @@ def test_stale_interrupted_event_marks_recovery_control(): assert "\"recovery_control\": True" in RUN_JOURNAL_PY -def test_done_and_restore_filters_recovery_messages_from_frontend_state(): - assert "_filterRecoveryControlMessages(S.messages || [])" in MESSAGES_JS - assert "if(!m||m.role==='tool') return false;" in MESSAGES_JS - assert "if(m.recovery_control===true) return true;" in MESSAGES_JS +def test_done_and_restore_preserve_recovery_messages_for_coordinate_merges(): + assert "_filterRecoveryControlMessages(S.messages || [])" not in MESSAGES_JS assert "continue exactly where you left off" in MESSAGES_JS assert "do not retry the same tool call" in MESSAGES_JS diff --git a/tests/test_live_to_final_anchor_visible_order.py b/tests/test_live_to_final_anchor_visible_order.py index 589ed37bb1c..20904daab35 100644 --- a/tests/test_live_to_final_anchor_visible_order.py +++ b/tests/test_live_to_final_anchor_visible_order.py @@ -813,24 +813,26 @@ def test_stream_end_restore_attaches_projected_anchor_scene_before_render(): assert "function _attachProjectedAnchorSceneToLastAssistant" in MESSAGES_JS carry_idx = restore.index("const _stagedMessages=_carryForwardEphemeralTurnFields(_currentMessages, _nextMsgs3018);") - filter_idx = restore.index("S.messages=_filterRecoveryControlMessages(_resolvedMessages || []);") + install_idx = restore.index("S.messages=_resolvedMessages || [];" ) attach_idx = restore.index("_attachProjectedAnchorSceneToLastAssistant(S.messages);") render_idx = restore.index("syncTopbar();renderMessages({preserveScroll:true})") - assert carry_idx < filter_idx < attach_idx < render_idx + assert carry_idx < install_idx < attach_idx < render_idx + assert "_filterRecoveryControlMessages" not in restore def test_cancel_settlement_attaches_projected_anchor_scene_before_render(): cancel = _event_listener_body(MESSAGES_JS, "cancel") - fetch_idx = cancel.index("const _nextMsgs3018=(sessionPayload.messages||[]).filter(m=>m&&m.role);") - attach_idx = cancel.index("_attachProjectedAnchorSceneToLastAssistant(_nextMsgs3018);") - carry_idx = cancel.index("S.messages=_carryForwardEphemeralTurnFields(S.messages||[], _nextMsgs3018);") + session_idx = cancel.index("const _currentTerminalSession=S.session;") + apply_idx = cancel.index("_applyEmbeddedTerminalSession(", session_idx) + attach_idx = cancel.index("_attachProjectedAnchorSceneToLastAssistant(S.messages);") render_idx = cancel.index("renderMessages({preserveScroll:true});") - assert fetch_idx < attach_idx < carry_idx < render_idx + assert session_idx < apply_idx < attach_idx < render_idx + assert "_filterRecoveryControlMessages" not in cancel - embedded_idx = cancel.index("if(_applyCancelSessionPayload(_cancelSessionPayload)) return;") - fallback_get_idx = cancel.index("const data=await api(`/api/session?session_id=${encodeURIComponent(activeSid)}`);") - fallback_apply_idx = cancel.index("if(data&&data.session) _applyCancelSessionPayload(data.session);") + embedded_idx = cancel.index("if(_cancelSessionPayload&&_applyCancelSessionPayload(_cancelSessionPayload)) return;") + fallback_get_idx = cancel.index("const data=await api(`/api/session?session_id=${encodeURIComponent(activeSid)}`);", embedded_idx) + fallback_apply_idx = cancel.index("if(data&&data.session) _applyCancelSessionPayload(data.session,true);", fallback_get_idx) assert embedded_idx < fallback_get_idx < fallback_apply_idx fallback_push_idx = cancel.index("S.messages.push({role:'assistant',content:`**Task cancelled:**") @@ -843,11 +845,12 @@ def test_application_error_settlement_attaches_projected_anchor_scene_before_ren apperror = _event_listener_body(MESSAGES_JS, "apperror") assert "_applyToAnchor('apperror'" in apperror - session_idx = apperror.index("const _nextMsgs3018=(d.session.messages||[]).filter(m=>m&&m.role);") - attach_idx = apperror.index("_attachProjectedAnchorSceneToLastAssistant(_nextMsgs3018);") - carry_idx = apperror.index("S.messages=_carryForwardEphemeralTurnFields(S.messages||[], _nextMsgs3018);") + session_idx = apperror.index("const _currentTerminalSession=S.session;") + apply_idx = apperror.index("_applyEmbeddedTerminalSession(", session_idx) + attach_idx = apperror.index("_attachProjectedAnchorSceneToLastAssistant(S.messages);") render_idx = apperror.index("renderMessages({preserveScroll:true});") - assert session_idx < attach_idx < carry_idx < render_idx + assert session_idx < apply_idx < attach_idx < render_idx + assert "_filterRecoveryControlMessages" not in apperror synthetic_push_idx = apperror.index("S.messages.push({role:'assistant',content:`**${label}:**") synthetic_attach_idx = apperror.index("_attachProjectedAnchorSceneToLastAssistant(S.messages);", synthetic_push_idx) diff --git a/tests/test_session_rotate_url_sync.py b/tests/test_session_rotate_url_sync.py index 40779959f1a..a6685f447d0 100644 --- a/tests/test_session_rotate_url_sync.py +++ b/tests/test_session_rotate_url_sync.py @@ -1,6 +1,5 @@ """Regression tests for session id rotation URL sync.""" from pathlib import Path -import re REPO_ROOT = Path(__file__).parent.parent.resolve() MESSAGES_JS = (REPO_ROOT / "static" / "messages.js").read_text(encoding="utf-8") @@ -10,14 +9,10 @@ def test_stream_completion_syncs_rotated_session_id_to_tab_state(): """When compact/restore returns a new session id, the tab anchor follows it.""" # #3018 inserted a carry-forward of ephemeral per-turn fields into both the # completion (_finishDone) and settled-restore assignments; match the new shapes. - completion_marker = re.compile( - r"S\.session=d\.session;\s*" - r"S\.messages=_carryForwardEphemeralTurnFields\(S\.messages\|\|\[\], d\.session\.messages\|\|\[\]\);" - ) - settled_marker = "S.session=session;\n const _nextMsgs3018=(session.messages||[]).filter(m=>m&&m.role);" + completion_marker = "S.session=completedSession;" + settled_marker = "S.session=session;\n const _nextMsgs3018=Array.isArray(session.messages)?session.messages:[];" - completion_match = completion_marker.search(MESSAGES_JS) - completion_pos = completion_match.start() if completion_match else -1 + completion_pos = MESSAGES_JS.find(completion_marker) settled_pos = MESSAGES_JS.find(settled_marker) assert completion_pos != -1 assert settled_pos != -1 @@ -27,8 +22,8 @@ def test_stream_completion_syncs_rotated_session_id_to_tab_state(): # stale-prefix guard before the tab-state sync, so keep the assertion local # to the handler while widening the slice enough to cover the new helper # state and the unchanged localStorage/update-url writes. - completion_block = MESSAGES_JS[completion_pos : completion_pos + 1000] - settled_block = MESSAGES_JS[settled_pos : settled_pos + 1800] + completion_block = MESSAGES_JS[completion_pos : completion_pos + 1800] + settled_block = MESSAGES_JS[settled_pos : settled_pos + 2400] for block in (completion_block, settled_block): assert "localStorage.setItem('hermes-webui-session',S.session.session_id);" in block diff --git a/tests/test_sidebar_first_turn_visibility.py b/tests/test_sidebar_first_turn_visibility.py index 5b3400cc30c..2b9b9694910 100644 --- a/tests/test_sidebar_first_turn_visibility.py +++ b/tests/test_sidebar_first_turn_visibility.py @@ -1,6 +1,11 @@ """Regressions for first-turn sessions appearing in the sidebar immediately.""" +import json import pathlib +import shutil +import subprocess + +import pytest REPO = pathlib.Path(__file__).parent.parent @@ -65,6 +70,43 @@ def test_sessions_js_has_local_turn_upsert_helper(self): "Optimistic row should render as streaming until the backend reconciles." ) + def test_optimistic_upsert_counts_absolute_paginated_message_end(self): + node = shutil.which("node") + if not node: + pytest.skip("node not on PATH") + source = read("static/sessions.js") + start = source.index("function upsertActiveSessionForLocalTurn") + brace = source.index("){", start) + 1 + depth = 1 + end = brace + 1 + while depth: + depth += source[end] == "{" + depth -= source[end] == "}" + end += 1 + helper = source[start:end] + script = f""" +const assert=require('node:assert/strict'); +const upsert=eval('(' + {json.dumps(helper)} + ')'); +global._allSessions=[]; +global.renderSessionListFromCache=()=>{{}}; +function count(offset,truncated){{ + global.S={{session:{{session_id:'sid',message_count:0}},messages:Array.from({{length:4}},()=>({{}}))}}; + if(offset===undefined) delete global._oldestIdx; else global._oldestIdx=offset; + if(truncated===undefined) delete global._messagesTruncated; else global._messagesTruncated=truncated; + upsert(); + return S.session.message_count; +}} +assert.equal(count(90,true),94); +assert.equal(count(90,false),4); +assert.equal(count(0,true),4); +assert.equal(count(undefined,true),4); +assert.equal(count(-1,true),4); +console.log('ok'); +""" + result = subprocess.run([node, "-e", script], capture_output=True, text=True, check=False) + assert result.returncode == 0, result.stderr or result.stdout + assert result.stdout.strip() == "ok" + def test_messages_comments_document_why_each_optimistic_upsert_stays_separate(self): src = read("static/messages.js") assert "First optimistic pass" in src and "before /api/chat/start" in src diff --git a/tests/test_sprint42.py b/tests/test_sprint42.py index c629e0f6477..7357d0c29f5 100644 --- a/tests/test_sprint42.py +++ b/tests/test_sprint42.py @@ -752,11 +752,11 @@ def test_streaming_persists_reasoning_in_session(): assert "_rm['reasoning'] = _existing_reasoning" in src, \ "the no-think-block branch must still persist _reasoning_text into the assistant message" - # Persistence block must come BEFORE the settled raw_session payload is built + # Persistence block must come BEFORE the settled terminal payload is built persist_idx = src.index("Persist reasoning trace in the session") - raw_session_idx = src.index("raw_session = _session_payload_with_full_messages") - assert persist_idx < raw_session_idx, \ - "Reasoning persistence block must appear before raw_session assignment" + terminal_payload_idx = src.index("_terminal_session_payload = _best_effort_terminal_session_payload(s)", persist_idx) + assert persist_idx < terminal_payload_idx, \ + "Reasoning persistence block must appear before terminal payload assignment" def test_done_handler_patches_reasoning_field(): diff --git a/tests/test_streaming_done_payload_message_count.py b/tests/test_streaming_done_payload_message_count.py index 64440c1135b..1ea813d2091 100644 --- a/tests/test_streaming_done_payload_message_count.py +++ b/tests/test_streaming_done_payload_message_count.py @@ -70,25 +70,27 @@ def test_full_message_payload_includes_todo_state_snapshot(): assert payload["todo_state"]["ts"] == 101 -def test_done_payload_uses_full_message_count_helper(): +def test_done_payload_uses_bounded_terminal_message_count_helper(): done_idx = STREAMING_SOURCE.index("put('done', _done_payload)") - block_start = STREAMING_SOURCE.rfind("raw_session =", 0, done_idx) + block_start = STREAMING_SOURCE.rfind("_done_payload =", 0, done_idx) block = STREAMING_SOURCE[block_start:done_idx] - assert "_session_payload_with_full_messages(s, tool_calls=tool_calls)" in block + assert "_best_effort_terminal_session_payload(s)" in block + assert "_session_payload_with_full_messages(s" not in block assert "s.compact() | {'messages': s.messages" not in block -def test_apperror_payload_uses_full_message_count_helper(): +def test_apperror_payload_uses_bounded_terminal_message_count_helper(): error_idx = STREAMING_SOURCE.index("put('apperror', _error_payload)") - block_start = STREAMING_SOURCE.rfind("_error_payload['session']", 0, error_idx) + block_start = STREAMING_SOURCE.rfind("_terminal_session_payload =", 0, error_idx) block = STREAMING_SOURCE[block_start:error_idx] - assert "_session_payload_with_full_messages(s, tool_calls=s.tool_calls)" in block + assert "_best_effort_terminal_session_payload(s)" in block + assert "_session_payload_with_full_messages(s" not in block assert "s.compact() | {'messages': s.messages" not in block -def test_gateway_done_payload_uses_full_message_count_helper(): +def test_gateway_done_payload_uses_bounded_terminal_message_count_helper(): """The gateway-routed chat `done` SSE shares the settled-payload path and must also report a message_count matching the embedded transcript (sibling of the two streaming.py sites).""" @@ -97,5 +99,5 @@ def test_gateway_done_payload_uses_full_message_count_helper(): block_start = gateway_source.rfind("gateway_session_payload =", 0, done_idx) block = gateway_source[block_start:done_idx] - assert "_session_payload_with_full_messages(s, tool_calls=[])" in block + assert "_best_effort_terminal_session_payload(s)" in block assert 's.compact() | {"messages": s.messages' not in block diff --git a/tests/test_streaming_markdown.py b/tests/test_streaming_markdown.py index bf6260b4f03..b4c105a2350 100644 --- a/tests/test_streaming_markdown.py +++ b/tests/test_streaming_markdown.py @@ -588,7 +588,7 @@ def test_done_handler_prefers_message_tool_metadata_for_settled_render(self): assert fn, "'done' handler not found" done_before_render = fn[:fn.index("renderMessages({preserveScroll:true})")] assert "const hasMessageToolMetadata=S.messages.some" in done_before_render - assert "!hasMessageToolMetadata&&d.session.tool_calls&&d.session.tool_calls.length" in done_before_render + assert "!hasMessageToolMetadata&&completedSession.tool_calls&&completedSession.tool_calls.length" in done_before_render assert "S.toolCalls=hasMessageToolMetadata?[]:S.toolCalls.map" in done_before_render diff --git a/tests/test_terminal_session_tail_and_merge.py b/tests/test_terminal_session_tail_and_merge.py new file mode 100644 index 00000000000..17549660bf3 --- /dev/null +++ b/tests/test_terminal_session_tail_and_merge.py @@ -0,0 +1,430 @@ +"""Behavioral checks for bounded terminal session settlement.""" + +import json +import shutil +import subprocess +from pathlib import Path +from types import SimpleNamespace + +import pytest + +import api.streaming as streaming +from tests.js_source_extract import extract_function + + +STREAMING_SOURCE = Path("api/streaming.py").read_text(encoding="utf-8") +GATEWAY_SOURCE = Path("api/gateway_chat.py").read_text(encoding="utf-8") +MESSAGES_SOURCE = Path("static/messages.js").read_text(encoding="utf-8") + + +class _Session(SimpleNamespace): + def compact(self): + return {"session_id": self.session_id, "message_count": 1, "title": "stale"} + + +def _node(script): + node = shutil.which("node") + if not node: + pytest.skip("node is required") + result = subprocess.run([node, "-e", script], text=True, capture_output=True, check=False) + assert result.returncode == 0, result.stderr or result.stdout + return result.stdout.strip() + + +def _arrow(source, name): + start = source.index(f"const {name}=") + brace = source.index("{", source.index("=", start)) + depth = 1 + end = brace + 1 + while depth: + depth += source[end] == "{" + depth -= source[end] == "}" + end += 1 + return source[source.index("=", start) + 1:end] + + +def test_terminal_tail_is_raw_bounded_and_authoritative(): + messages = [{"role": "user", "content": f"row-{i}"} for i in range(6000)] + todo = {"todos": [{"id": "full-only", "status": "in_progress"}]} + messages.insert(0, {"role": "tool", "content": json.dumps(todo)}) + messages += [ + {"role": "assistant", "content": "hidden", "recovery_control": True}, + {"role": "tool", "content": "{}"}, + ] + messages += [{"role": "assistant", "content": f"tail-{i}"} for i in range(8)] + messages.append({"role": "assistant", "content": "settled"}) + payload = streaming._redacted_terminal_session_payload( + _Session(session_id="tail", messages=messages) + ) + assert len(payload["messages"]) <= 300 + assert payload["messages"] == messages[payload["_messages_offset"]:] + assert payload["message_count"] == len(messages) + assert payload["_messages_truncated"] is True + assert payload["_tool_calls_truncated"] is False + assert any(message.get("recovery_control") for message in payload["messages"]) + actual = [m for m in payload["messages"] if streaming.visible_messages_for_anchor([m], auto_compression=True)] + expected = [m for m in messages if streaming.visible_messages_for_anchor([m], auto_compression=True)][-30:] + assert actual == expected + assert payload["todo_state"]["todos"] == todo["todos"] + assert payload["tool_calls"] == [] + + +def test_terminal_anchor_uses_raw_indexes_for_repeated_objects(): + repeated = {"role": "user", "content": "same"} + messages = [repeated] * 31 + [{"role": "assistant", "content": "done"}] + payload = streaming._redacted_terminal_session_payload(_Session(session_id="repeat", messages=messages)) + assert payload["_messages_offset"] == 2 + assert payload["messages"] == messages[2:] + + +def test_terminal_tool_calls_are_projected_and_bounded(): + messages = [{"role": "user", "content": str(i)} for i in range(1000)] + calls = [{"id": f"call-{i}", "assistant_msg_idx": 970 + i % 30} for i in range(301)] + calls += [{"id": "bad", "assistant_msg_idx": "970"}, {"id": "legacy", "assistant_msg_idx": -1}] + payload = streaming._redacted_terminal_session_payload( + _Session(session_id="calls", messages=messages, tool_calls=calls) + ) + assert len(payload["tool_calls"]) == 300 + assert payload["_tool_calls_truncated"] is True + assert payload["tool_calls"][0]["id"] == "call-2" + assert payload["tool_calls"][-1]["id"] == "legacy" + assert all(call["assistant_msg_idx"] == -1 or 0 <= call["assistant_msg_idx"] < 30 for call in payload["tool_calls"]) + filtered_only = streaming._redacted_terminal_session_payload( + _Session(session_id="calls", messages=messages, tool_calls=calls[:300] + [{"assistant_msg_idx": "970"}]) + ) + assert filtered_only["_tool_calls_truncated"] is False + + +def test_truncated_terminal_tool_calls_reconcile_with_expanded_browser_state(): + messages = [ + {"role": "user", "content": "run tools"}, + {"role": "assistant", "content": "working"}, + {"role": "assistant", "content": "done"}, + ] + calls = [ + { + "id": f"call-{i}", + "name": "terminal", + "snippet": f"current-{i}", + "assistant_msg_idx": 1, + } + for i in range(301) + ] + payload = streaming._redacted_terminal_session_payload( + _Session(session_id="calls", messages=messages, tool_calls=calls) + ) + assert [call["id"] for call in payload["tool_calls"]] == [f"call-{i}" for i in range(1, 301)] + + helper = extract_function(MESSAGES_SOURCE, "_applyEmbeddedTerminalSession") + incoming = json.loads(json.dumps(payload)) + incoming["_tool_calls_truncated"] = True + incoming["tool_calls"][0]["snippet"] = "incoming-wins" + expanded_calls = json.loads(json.dumps(calls)) + expanded_calls[0]["snippet"] = "expanded-prefix" + script = f""" +const assert=require('node:assert/strict'); +const apply=eval('(' + {json.dumps(helper)} + ')'); +const incoming={json.dumps(incoming)}; +const messages={json.dumps(messages)}; +const current={{session_id:'calls',message_count:3,messages,tool_calls:{json.dumps(expanded_calls)}}}; +const merged=apply(incoming,current,messages,0); +assert.ok(merged); +assert.equal(merged.tool_calls.length,301); +assert.equal(merged.tool_calls.filter(c=>c.id==='call-0').length,1); +assert.equal(merged.tool_calls.find(c=>c.id==='call-0').snippet,'expanded-prefix'); +assert.equal(merged.tool_calls.filter(c=>c.id==='call-1').length,1); +assert.equal(merged.tool_calls.find(c=>c.id==='call-1').snippet,'incoming-wins'); +const ordinary=apply( + {{session_id:'calls',_messages_offset:2,message_count:4,messages:[{{id:'remote-2'}},{{id:'remote-3'}}],tool_calls:[],_tool_calls_truncated:false}}, + {{session_id:'calls',message_count:4,messages:[{{id:'local-0'}},{{id:'local-1'}},{{id:'local-2'}},{{id:'local-3'}}],tool_calls:[{{id:'before',assistant_msg_idx:0}},{{id:'overlap',assistant_msg_idx:2}}]}}, + [{{id:'local-0'}},{{id:'local-1'}},{{id:'local-2'}},{{id:'local-3'}}],0, +); +assert.deepEqual(ordinary.tool_calls.map(c=>c.id),['before']); +const unkeyed=apply( + {{session_id:'calls',_messages_offset:0,message_count:3,messages,tool_calls:[{{name:'terminal',assistant_msg_idx:1,snippet:'incoming'}}],_tool_calls_truncated:true}}, + {{session_id:'calls',message_count:3,messages,tool_calls:[{{name:'terminal',assistant_msg_idx:1,snippet:'current'}}]}}, + messages,0, +); +assert.deepEqual(unkeyed.tool_calls.map(c=>c.snippet),['incoming']); +const failClosed=apply( + {{session_id:'calls',_messages_offset:0,message_count:3,messages,tool_calls:[],_tool_calls_truncated:true}}, + {{session_id:'calls',message_count:3,messages,tool_calls:[{{assistant_msg_idx:1,snippet:'nameless'}},{{name:'terminal',assistant_msg_idx:-1}}]}}, + messages,0, +); +assert.deepEqual(failClosed.tool_calls,[]); +assert.equal(apply( + {{session_id:'calls',_messages_offset:0,message_count:3,messages,tool_calls:[],_tool_calls_truncated:'yes'}}, + {{session_id:'calls',message_count:3,messages}},messages,0, +),null); +console.log('ok'); +""" + assert _node(script) == "ok" + + +def test_truncated_unkeyed_tool_calls_use_occurrence_counts(): + helper = extract_function(MESSAGES_SOURCE, "_applyEmbeddedTerminalSession") + script = f""" +const assert=require('node:assert/strict'); +const apply=eval('(' + {json.dumps(helper)} + ')'); +const messages=[{{id:'user'}},{{id:'assistant'}},{{id:'done'}}]; +const incoming={{session_id:'calls',_messages_offset:0,message_count:3,messages,tool_calls:[ + {{name:'terminal',assistant_msg_idx:1,snippet:'incoming-2'}}, + {{name:'terminal',assistant_msg_idx:1,snippet:'incoming-3'}}, +],_tool_calls_truncated:true}}; +const current={{session_id:'calls',message_count:3,messages,tool_calls:[ + {{name:'terminal',assistant_msg_idx:1,snippet:'current-0'}}, + {{name:'terminal',assistant_msg_idx:1,snippet:'current-1'}}, + {{name:'terminal',assistant_msg_idx:1,snippet:'current-2'}}, + {{name:'terminal',assistant_msg_idx:1,snippet:'current-3'}}, +]}}; +const merged=apply(incoming,current,messages,0); +assert.ok(merged); +assert.deepEqual(merged.tool_calls.map(c=>c.snippet),['current-0','current-1','incoming-2','incoming-3']); +assert.equal(merged.tool_calls.filter(c=>c.name==='terminal'&&c.assistant_msg_idx===1).length,4); +assert.equal(merged.tool_calls.filter(c=>c.snippet==='incoming-2').length,1); +assert.equal(merged.tool_calls.filter(c=>c.snippet==='incoming-3').length,1); +const mixed=apply( + {{...incoming,tool_calls:[{{id:'identified',name:'terminal',assistant_msg_idx:1,snippet:'incoming-id'}},incoming.tool_calls[1]]}}, + current,messages,0, +); +assert.deepEqual(mixed.tool_calls.map(c=>c.snippet),['current-0','current-1','current-2','incoming-id','incoming-3']); +console.log('ok'); +""" + assert _node(script) == "ok" + + +def test_limited_tool_output_uses_the_existing_character_bound(): + from api.session_ops import _tool_message_for_limited_payload + + message = {"role": "tool", "content": "界🙂" * 3000} + bounded = _tool_message_for_limited_payload(message) + assert bounded["_content_truncated"] is True + assert bounded["content"].startswith(message["content"][:4096]) + assert len(bounded["content"]) < len(message["content"]) + assert json.loads(json.dumps(bounded, ensure_ascii=False))["content"] == bounded["content"] + assert bounded["_content_original_chars"] == len(message["content"]) + + +def test_terminal_helper_raises_and_producers_do_not_emit_null_done(): + class Broken(_Session): + def compact(self): + raise RuntimeError("broken compact") + + with pytest.raises(RuntimeError, match="broken compact"): + streaming._redacted_terminal_session_payload(Broken(session_id="broken", messages=[])) + start = STREAMING_SOURCE.index("def _redacted_terminal_session_payload") + end = STREAMING_SOURCE.index("\ndef _best_effort_terminal_session_payload", start) + assert "except Exception" not in STREAMING_SOURCE[start:end] + best_effort = STREAMING_SOURCE[end:] + assert "except Exception" in best_effort[:best_effort.index("\ndef _compact_for_echo_compare")] + assert "_best_effort_terminal_session_payload(s)" in STREAMING_SOURCE + assert "_redacted_session_payload_with_full_messages" not in STREAMING_SOURCE + assert all("session=" not in line for line in GATEWAY_SOURCE.splitlines() if 'put_gateway_event("cancel"' in line) + + +def test_coordinate_merge_preserves_widths_and_fails_closed(): + helper = extract_function(MESSAGES_SOURCE, "_applyEmbeddedTerminalSession") + apperror_start = MESSAGES_SOURCE.index("source.addEventListener('apperror'") + path_start = MESSAGES_SOURCE.index( + "const currentSid=S.session&&S.session.session_id;", apperror_start + ) + path_end = MESSAGES_SOURCE.index("if(eventMatchesCurrent){", path_start) + apperror_session_path = MESSAGES_SOURCE[path_start:path_end] + script = f""" +const assert=require('node:assert/strict'); +const apply=eval('(' + {json.dumps(helper)} + ')'); +const _applyEmbeddedTerminalSession=apply; +for(const size of [30,90,500,6868]){{ + const local=Array.from({{length:size}},(_,i)=>({{id:'local-'+i}})); + const result=apply({{session_id:'same',message_count:999,messages:[{{id:'remote'}}]}},{{session_id:'same',messages:local}},local,0); + assert.strictEqual(result,null); +}} +const local=[{{id:'user'}}]; +const appended=apply({{session_id:'same',_messages_offset:0,message_count:2,messages:[{{id:'user-persisted'}},{{id:'final'}}]}},{{session_id:'same',messages:local}},local,0); +assert.deepEqual(appended.messages.map(m=>m.id),['user-persisted','final']); +assert.strictEqual(apply({{session_id:'same',_messages_offset:0,message_count:1,_messages_truncated:true,messages:[{{id:'bad-flag'}}]}},{{session_id:'same',messages:local}},local,0),null); +const sameWindow=Array.from({{length:30}},(_,i)=>({{id:'local-'+(971+i)}})); +const sameIncoming=Array.from({{length:30}},(_,i)=>({{id:'remote-'+(971+i)}})); +const stale=apply({{session_id:'same',_messages_offset:971,message_count:1000,_messages_truncated:true,messages:sameIncoming.slice(0,29)}},{{session_id:'same',message_count:1001,messages:sameWindow}},sameWindow,971); + assert.strictEqual(stale,null); + const union=apply({{session_id:'same',_messages_offset:971,message_count:1001,messages:sameIncoming}},{{session_id:'same',message_count:1001,messages:sameWindow}},sameWindow,971); + assert.equal(union.messages.length,30); assert.strictEqual(union.messages[0],sameIncoming[0]); +const localOnlyMessages=[{{id:'persisted-0'}},{{id:'persisted-1'}},{{id:'persisted-2'}},{{id:'local-only'}}]; +const settled=apply({{session_id:'same',_messages_offset:1,message_count:3,_messages_truncated:true,messages:[{{id:'persisted-1'}},{{role:'assistant',content:'settled final answer'}}]}},{{session_id:'same',message_count:3,messages:localOnlyMessages}},localOnlyMessages,0); + assert.ok(settled); + assert.deepEqual(settled.messages,[{{id:'persisted-1'}},{{role:'assistant',content:'settled final answer'}}]); + assert.strictEqual(settled._messages_offset,1); assert.strictEqual(settled._messages_truncated,true); +const currentCalls=[{{assistant_msg_idx:1,duration:3,snippet:'retained'}},{{id:'same',assistant_msg_idx:0,duration:3}},{{id:'legacy',assistant_msg_idx:-1}}]; + const incomingCalls=[{{id:'same',assistant_msg_idx:0,snippet:'settled'}},{{id:'new',assistant_msg_idx:0}},{{id:'legacy-new',assistant_msg_idx:-1}},{{id:'bad',assistant_msg_idx:'1'}}]; + const withCalls=apply({{session_id:'same',_messages_offset:92,message_count:94,messages:[{{id:'auth'}},{{id:'cancel'}}],tool_calls:incomingCalls}},{{session_id:'same',message_count:93,messages:[{{id:'old'}},{{id:'hidden'}},{{id:'tail'}}],tool_calls:currentCalls}},[{{id:'old'}},{{id:'hidden'}},{{id:'tail'}}],90); + assert.deepEqual(withCalls.tool_calls.map(c=>[c.id??null,c.assistant_msg_idx,c.duration,c.snippet]),[[null,1,3,'retained'],['same',2,undefined,'settled'],['new',2,undefined,undefined],['legacy-new',-1,undefined,undefined]]); +const omittedCalls=apply({{session_id:'same',_messages_offset:92,message_count:94,messages:[{{id:'auth'}},{{id:'cancel'}}]}},{{session_id:'same',message_count:93,messages:[{{id:'old'}},{{id:'hidden'}},{{id:'tail'}}],tool_calls:currentCalls}},[{{id:'old'}},{{id:'hidden'}},{{id:'tail'}}],90); + assert.deepEqual(omittedCalls.tool_calls.map(c=>[c.id??null,c.assistant_msg_idx]),[[null,1],['same',0],['legacy',-1]]); +const hidden=[{{id:'old'}},{{id:'hidden',recovery_control:true}},{{id:'tail'}}]; +const hiddenUnion=apply({{session_id:'same',_messages_offset:92,message_count:94,messages:[{{id:'auth'}},{{id:'cancel'}}]}},{{session_id:'same',messages:hidden}},hidden,90); +assert.deepEqual(hiddenUnion.messages.map(m=>m.id),['old','hidden','auth','cancel']); +const disjoint=apply({{session_id:'same',_messages_offset:1002,message_count:1003,_messages_truncated:true,messages:[{{role:'assistant',content:'settled final answer'}}],tool_calls:[]}},{{session_id:'same',messages:sameWindow}},sameWindow,910); + assert.deepEqual(disjoint.messages,[{{role:'assistant',content:'settled final answer'}}]); + assert.strictEqual(disjoint._messages_offset,1002); + assert.strictEqual(disjoint._messages_truncated,true); + assert.deepEqual(disjoint.tool_calls,[]); +const gap=apply({{session_id:'same',_messages_offset:93,message_count:94,messages:[{{id:'gap'}}]}},{{session_id:'same',messages:[{{id:'old'}},{{id:'tail'}}]}},[{{id:'old'}},{{id:'tail'}}],90); + assert.strictEqual(gap,null); +const sparse=[{{id:'raw-90'}},{{id:'raw-92'}}]; +const sparseResult=apply({{session_id:'same',message_count:94,_messages_offset:92,messages:[{{id:'raw-92'}},{{id:'raw-93'}}]}},{{session_id:'same',message_count:94,messages:sparse}},sparse,90); + assert.strictEqual(sparseResult,null); +assert.strictEqual(apply({{messages:[{{id:'bad'}}]}},null,local,0),null); +const prefix=Array.from({{length:6}},(_,i)=>({{id:'p'+i}})); +const current={{session_id:'before',message_count:6,messages:prefix,tool_calls:[ + {{id:'keep',assistant_msg_idx:3}},{{id:'dup',assistant_msg_idx:4,meta:'old'}} +]}}; +const continuation={{session_id:'after',parent_session_id:'before',message_count:8, + _messages_offset:5,_messages_truncated:true, + messages:[{{id:'p5'}},{{id:'tail'}},{{id:'done'}}],tool_calls:[ + {{id:'dup',assistant_msg_idx:0,meta:'new'}},{{id:'new',assistant_msg_idx:1}} +]}}; +const merged=apply(continuation,current,prefix,0); +assert.deepEqual(merged.messages.map(m=>m.id),['p0','p1','p2','p3','p4','p5','tail','done']); +assert.equal(merged.session_id,'after'); assert.equal(merged.parent_session_id,'before'); +assert.equal(merged._messages_offset,0); assert.equal(merged.message_count,8); +assert.equal(merged._messages_truncated,false); +assert.deepEqual(merged.tool_calls.map(c=>[c.id,c.assistant_msg_idx,c.meta]),[ + ['keep',3,undefined],['dup',5,'new'],['new',6,undefined] +]); +for(const parent of ['other','']){{ + const raw=apply({{...continuation,parent_session_id:parent}},current,prefix,0); + assert.equal(raw.messages.length,3); assert.deepEqual(raw.messages,continuation.messages); +}} +const missingParent={{...continuation}}; delete missingParent.parent_session_id; +assert.equal(apply(missingParent,current,prefix,0).messages.length,3); +const S={{session:{{session_id:'before',message_count:6}},messages:prefix}}; +const activeSid='before',_oldestIdx=0; +function applyAppError(d){{ + {apperror_session_path} + return {{eventMatchesCurrent,_terminalSession}}; +}} +const apperror=applyAppError({{old_session_id:'before',new_session_id:'after',session:continuation}}); +assert.equal(apperror.eventMatchesCurrent,true); +assert.deepEqual(apperror._terminalSession.messages.map(m=>m.id),merged.messages.map(m=>m.id)); +assert.equal(apperror._terminalSession.session_id,'after'); +assert.equal(apperror._terminalSession.parent_session_id,'before'); +console.log('ok'); +""" + assert _node(script) == "ok" + + +def test_done_tool_calls_presence_distinguishes_authoritative_empty_from_absent(): + start = MESSAGES_SOURCE.index("const hasMessageToolMetadata=S.messages.some") + end = MESSAGES_SOURCE.index("if(typeof renderSessionArtifacts", start) + settlement = MESSAGES_SOURCE[start:end] + script = f""" +const assert=require('node:assert/strict'); +const _mergeSettledToolCallsWithLiveMetadata=rawCalls=>(rawCalls||[]).map(call=>({{...call,done:true}})); +function settle(completedSession, provided){{ + const S={{messages:[{{role:'assistant',content:'answer'}}],toolCalls:[{{id:'live'}}]}}; + const _terminalToolCallsProvided=provided; + {settlement} + return S.toolCalls; +}} +assert.deepEqual(settle({{tool_calls:[]}},true),[]); +assert.deepEqual(settle({{}},false),[{{id:'live',done:true}}]); +console.log('ok'); +""" + assert _node(script) == "ok" + + +def test_cancel_full_get_and_sid_validation_are_behavioral(): + cancel = _arrow(MESSAGES_SOURCE, "_applyCancelSessionPayload") + merge = extract_function(MESSAGES_SOURCE, "_applyEmbeddedTerminalSession") + script = f""" +const assert=require('node:assert/strict'); +const _applyEmbeddedTerminalSession=eval('(' + {json.dumps(merge)} + ')'); +const apply=eval('(' + {json.dumps(cancel)} + ')'); +const activeSid='same'; let _oldestIdx=90,_messagesTruncated=true; +const local=Array.from({{length:90}},(_,i)=>({{id:'local-'+i}})); +const S={{session:{{session_id:activeSid}},activeStreamId:'stream',messages:local}}; +const _carryForwardEphemeralTurnFields=(a,b)=>b,_attachProjectedAnchorSceneToLastAssistant=()=>{{}},_hydrateTodosFromSession=()=>{{}}; +const _isMessagePaneNearBottom=()=>true,_isMessageReaderUnpinned=()=>false,_messageUserUnpinned=false; +const clearLiveToolCards=()=>{{}},assistantText='',removeThinking=()=>{{}},_markSessionViewed=()=>{{}},renderMessages=()=>{{}},scrollToBottom=()=>{{}}; +const _setActiveSessionUrl=()=>{{}},localStorage={{setItem(){{}}}}; +assert.equal(apply({{session_id:activeSid,_messages_offset:1002,message_count:1003,messages:[{{id:'gap'}}]}}),false); +assert.equal(apply({{messages:[{{id:'missing'}}]}}),false); +assert.equal(apply({{session_id:activeSid,_messages_offset:0,message_count:91,messages:[...local,{{id:'cancel'}}]}},true),true); +assert.equal(_oldestIdx,0); assert.equal(_messagesTruncated,false); assert.equal(S.messages[90].id,'cancel'); +assert.equal(S.activeStreamId,null); +assert.equal(apply({{session_id:'rotated',messages:[{{id:'rotated'}}]}}),false); +console.log('ok'); +""" + assert _node(script) == "ok" + + +def test_full_restore_resets_cursor_and_ephemeral_fields_survive(): + restore = extract_function(MESSAGES_SOURCE, "_restoreSettledSession", prefix="async function") + carry = extract_function(MESSAGES_SOURCE, "_carryForwardEphemeralTurnFields") + identity = extract_function(MESSAGES_SOURCE, "_messageIdentityKey") + script = f""" +const assert=require('node:assert/strict'); +(async()=>{{ +const activeSid='same',streamId='stream'; let _oldestIdx=90,_messagesTruncated=true,_streamFinalized=false; +const full=Array.from({{length:91}},(_,i)=>({{id:'full-'+i}})); full[89]={{id:'hidden',recovery_control:true}}; +const api=async()=>({{session:{{session_id:activeSid,message_count:91,_messages_offset:0,_messages_truncated:false,messages:full,tool_calls:[]}}}}); +const S={{session:{{session_id:activeSid,active_stream_id:streamId}},activeStreamId:streamId,messages:[{{id:'window'}}],toolCalls:[]}}; +const source={{close(){{}}}},localStorage={{setItem(){{}}}}; +const _isActiveSession=()=>true,_isSessionCurrentPane=()=>true,_isSessionActivelyViewed=()=>true,_closeSource=s=>s.close(); +const _messageIdentityKey=eval('(' + {json.dumps(identity)} + ')'); +const _carryForwardEphemeralTurnFields=(a,b)=>b,_isTerminalStreamErrorMarkerMessage=()=>false; +const _attachProjectedAnchorSceneToLastAssistant=()=>{{}},_hydrateTodosFromSession=()=>{{}},_replaceMarkerOnlyAssistantWithStreamError=()=>false; +const _mergeSettledToolCallsWithLiveMetadata=x=>x,_clearAnchorProseIncrementalNode=()=>{{}},_cancelThrottledSnapshotTimer=()=>{{}}; +const _cancelAnimationFramePendingStreamRender=()=>{{}},_streamFadeCleanupReduceMotionListener=()=>{{}},_smdEndParser=()=>{{}},finalizeThinkingCard=()=>{{}}; +const _clearOwnerInflightState=()=>{{}},_flushReasoningToAnchor=()=>{{}},_scheduleAnchorRegistryCleanup=()=>{{}},_clearApprovalForOwner=()=>{{}},_clearClarifyForOwner=()=>{{}}; +const clearLiveToolCards=()=>{{}},removeThinking=()=>{{}},_markSessionCompletionUnread=()=>{{}},_markSessionViewed=()=>{{}},syncTopbar=()=>{{}},renderMessages=()=>{{}},renderSessionList=()=>{{}},_setActivePaneIdleIfOwner=()=>{{}},_setActiveSessionUrl=()=>{{}}; +let _queueDrainSid=null,_persistTimer=null; const assistantText=''; +const restore=eval('(' + {json.dumps(restore)} + ')'); +assert.equal(await restore(source),true); assert.equal(_oldestIdx,0); assert.equal(_messagesTruncated,false); +assert.equal(S.messages.length,91); assert.equal(S.messages[89].id,'hidden'); +const fields=['_turnUsage','_turnDuration','_turnTps','_gatewayRouting','_statusCard','_anchor_stream_id','_anchor_activity_scene']; +const _EPHEMERAL_TURN_FIELDS=fields,_isHistoricalAnchorActivityScene=()=>false; +const carry=eval('(' + {json.dumps(carry)} + ')'); +const before={{role:'assistant',content:'answer',_ts:1,_turnUsage:{{x:1}},_turnDuration:2,_turnTps:3,_gatewayRouting:'gw',_statusCard:{{ok:true}},_anchor_stream_id:'s',_anchor_activity_scene:{{id:'a'}}}}; +const after={{role:'assistant',content:'answer',_ts:1}}; carry([before],[after]); +for(const field of fields) assert.deepEqual(after[field],before[field]); +console.log('ok'); }})().catch(e=>{{console.error(e);process.exit(1)}}); +""" + assert _node(script) == "ok" + + +def test_terminal_wiring_reuses_existing_lifecycle_and_coordinate_paths(): + helper = extract_function(MESSAGES_SOURCE, "_applyEmbeddedTerminalSession").lower() + assert all(word not in helper for word in ("fetch", "await", "settimeout", "setinterval", "owner")) + assert "_stream_generations" not in MESSAGES_SOURCE.lower() + assert "_streamgenerationiscurrent" not in MESSAGES_SOURCE.lower() + assert "allowfinalized" not in MESSAGES_SOURCE.lower() + assert "applysession:" not in MESSAGES_SOURCE.lower() + done_start = MESSAGES_SOURCE.index("source.addEventListener('done'") + done_end = MESSAGES_SOURCE.index("source.addEventListener('stream_end'", done_start) + done = MESSAGES_SOURCE[done_start:done_end] + assert ": {session_id:activeSid}" in done + assert "if(!completedSession){" not in done + assert "S.messages=_filterRecoveryControlMessages" not in done + assert "liveDisplayText:typeof _streamDisplay==='function'?_streamDisplay():assistantText" in done + error_start = MESSAGES_SOURCE.index("source.addEventListener('apperror'") + error_end = MESSAGES_SOURCE.index("source.addEventListener('warning'", error_start) + error = MESSAGES_SOURCE[error_start:error_end] + assert "_terminalRecoveryPromise" not in error + assert "S.messages=_filterRecoveryControlMessages" not in error + assert "if(isRecoveryControlMessage){" in error + assert "S.session.message_count != null" in error + cancel_start = MESSAGES_SOURCE.index("source.addEventListener('cancel'") + cancel_end = MESSAGES_SOURCE.index("for(const _runJournalEventName", cancel_start) + cancel = MESSAGES_SOURCE[cancel_start:cancel_end] + assert "payloadSid!==activeSessionSid" in cancel + assert "fullSnapshot=false" in cancel + assert "const data=await api(" in cancel + assert "_applyCancelSessionPayload(data.session,true)" in cancel + assert "const status=await _restoreSettledSession(source,{" not in cancel + assert "parent_session_id" not in cancel + restore = extract_function(MESSAGES_SOURCE, "_restoreSettledSession", prefix="async function") + assert "_filterRecoveryControlMessages" not in restore + assert "_oldestIdx=_restoredOffset" in restore + assert "allowFinalized" not in restore + assert "restoreOwnerIsCurrent" not in restore