diff --git a/api/gateway_chat.py b/api/gateway_chat.py index 00183dd1..4072d39e 100644 --- a/api/gateway_chat.py +++ b/api/gateway_chat.py @@ -757,14 +757,19 @@ def _settle_gateway_terminal_error(session_id, stream_id, workspace, model, mode session.workspace = str(workspace) session.model = model session.model_provider = model_provider + terminal_session_persisted = False try: session.save() + 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=[]) ) error_payload["session_id"] = session.session_id + error_payload["terminal_session_persisted"] = terminal_session_persisted + if terminal_session_persisted: + error_payload["terminal_session_persisted_session_id"] = session.session_id return error_payload diff --git a/api/models.py b/api/models.py index 0540a932..dd3b5b33 100644 --- a/api/models.py +++ b/api/models.py @@ -2477,17 +2477,221 @@ def _run_journal_has_visible_output(session, stream_id: str | None) -> bool: return False +def _run_journal_event_owns_run( + event, + session_id: str, + stream_id: str | None, +) -> bool: + if not isinstance(event, dict) or not stream_id: + return False + seq = event.get('seq') + if isinstance(seq, bool) or not isinstance(seq, int) or seq < 1: + return False + return ( + event.get('session_id') == session_id + and event.get('run_id') == stream_id + and event.get('event_id') == f"{stream_id}:{seq}" + ) + + def _run_journal_terminal_state(session, stream_id: str | None) -> str | None: if not stream_id: return None try: - from api.run_journal import latest_run_summary - summary = latest_run_summary(session.session_id, stream_id) + from api.run_journal import ( + read_run_events, + select_authoritative_terminal_event, + ) + journal = read_run_events(session.session_id, stream_id) + terminal = select_authoritative_terminal_event(journal.get('events') or []) except Exception: return None - if not summary.get('terminal'): + if ( + not _run_journal_event_owns_run( + terminal, session.session_id, stream_id, + ) + or terminal.get('terminal') is not True + ): return None - return str(summary.get('terminal_state') or '') or None + return str(terminal.get('terminal_state') or '') or None + + +def _recoverable_unsaved_gateway_terminal_error( + session, + stream_id: str | None, +) -> dict | None: + """Return one validated current-turn terminal error from the run journal.""" + if not stream_id: + return None + try: + from api.run_journal import ( + read_run_events, + select_authoritative_terminal_event, + ) + journal = read_run_events(session.session_id, stream_id) + except Exception: + logger.debug( + "Session %s: failed to read terminal error journal for stream %s", + getattr(session, 'session_id', '?'), + stream_id, + exc_info=True, + ) + return None + + event = select_authoritative_terminal_event(journal.get('events') or []) + if ( + not isinstance(event, dict) + or event.get('event') != 'apperror' + or event.get('type') != 'apperror' + or event.get('terminal') is not True + ): + return None + if ( + not _run_journal_event_owns_run( + event, session.session_id, stream_id, + ) + ): + return None + expected_event_id = event['event_id'] + + payload = event.get('payload') + if not isinstance(payload, dict) or payload.get('session_id') != session.session_id: + return None + embedded_session = payload.get('session') + if ( + not isinstance(embedded_session, dict) + or embedded_session.get('session_id') != session.session_id + ): + return None + persisted_id = payload.get('terminal_session_persisted_session_id') + if ( + payload.get('terminal_session_persisted') is True + and persisted_id == session.session_id + ): + return None + + embedded_messages = embedded_session.get('messages') + if not isinstance(embedded_messages, list): + return None + current_user_idx = next( + ( + idx + for idx in range(len(embedded_messages) - 1, -1, -1) + if isinstance(embedded_messages[idx], dict) + and embedded_messages[idx].get('role') == 'user' + ), + None, + ) + if current_user_idx is None: + return None + candidate = next( + ( + embedded_messages[idx] + for idx in range(len(embedded_messages) - 1, current_user_idx, -1) + if isinstance(embedded_messages[idx], dict) + and embedded_messages[idx].get('role') == 'assistant' + ), + None, + ) + if ( + not isinstance(candidate, dict) + or candidate.get('_error') is not True + or not isinstance(candidate.get('content'), str) + or not candidate.get('content').strip() + ): + return None + return { + 'event_id': expected_event_id, + 'stream_id': stream_id, + 'message': dict(candidate), + } + + +def _pending_recovery_turn_start(session) -> int | None: + pending_text = getattr(session, 'pending_user_message', None) + if not pending_text: + return None + for idx in range(len(session.messages or []) - 1, -1, -1): + message = session.messages[idx] + if _message_matches_pending_checkpoint( + message, + pending_text, + session.pending_started_at, + session.pending_user_source, + session.pending_attachments, + ) or _message_matches_pending_text(message, pending_text): + return idx + return None + + +def _materialize_unsaved_gateway_terminal_error( + session, + stream_id: str | None, + recovery: dict | None = None, +) -> bool: + """Place the validated current-turn gateway error at the transcript tail.""" + recovery = recovery or _recoverable_unsaved_gateway_terminal_error( + session, stream_id, + ) + if not isinstance(recovery, dict): + return False + event_id = recovery.get('event_id') + candidate = recovery.get('message') + if not event_id or not isinstance(candidate, dict): + return False + + for existing in session.messages or []: + if ( + isinstance(existing, dict) + and existing.get('_recovered_event_id') == event_id + ): + return True + + turn_start = _pending_recovery_turn_start(session) + if turn_start is not None: + for existing in reversed((session.messages or [])[turn_start + 1:]): + if not isinstance(existing, dict): + continue + existing_stream = existing.get('_recovered_stream_id') + if existing_stream and existing_stream != stream_id: + continue + if ( + existing.get('role') == 'assistant' + and existing.get('_error') is True + and existing.get('content') == candidate.get('content') + ): + existing['_recovered_from_run_journal'] = True + existing['_recovered_stream_id'] = stream_id + existing['_recovered_event_id'] = event_id + return True + + recovered = dict(candidate) + recovered['_recovered_from_run_journal'] = True + recovered['_recovered_stream_id'] = stream_id + recovered['_recovered_event_id'] = event_id + session.messages.append(recovered) + return True + + +def _recover_journaled_output_and_terminal_error( + session, + stream_id: str | None, + *, + dedupe_existing: bool = False, + terminal_recovery: dict | None = None, +) -> tuple[bool, bool]: + """Recover readable activity first, then append its authoritative terminal error.""" + recovered_output = _append_journaled_partial_output( + session, + stream_id, + dedupe_existing=dedupe_existing, + ) + terminal_error_recovered = _materialize_unsaved_gateway_terminal_error( + session, + stream_id, + terminal_recovery, + ) + return recovered_output, terminal_error_recovered def _journal_is_still_arriving(session, stream_id: str | None) -> bool: @@ -2836,11 +3040,11 @@ def _append_journaled_partial_output( # onto the marker: `_journal_retry_stream_id`, `_journal_retry_attempts`, # `_journal_retry_first_seen_ts`. # * Every `get_session()` call that returns the full session checks the -# latest assistant marker; if the flag is set it re-runs -# `_append_journaled_partial_output` with `dedupe_existing=True`. On -# success the marker is promoted in place to the recovered-output -# wording, the journaled rows are reordered to sit above the marker, -# and all retry meta is stripped. If the journal is still missing or +# latest assistant marker; if the flag is set it re-runs journaled output +# and terminal-error recovery with `dedupe_existing=True`. On success, +# journaled rows move above the marker. A specific gateway terminal error +# replaces the marker; otherwise the marker is promoted to recovered-output +# wording and its retry meta is stripped. If the journal is still missing or # zero-byte, the retry is a no-op and does not consume attempt budget. # Terminal/non-useful journals consume attempt budget and can demote # immediately at the max-attempt cap. @@ -2979,7 +3183,7 @@ def _retry_journal_recovery_in_place( ) -> bool: """Re-attempt run-journal recovery for the most recent pending marker. - Returns True if the marker was promoted to the recovered-output wording. + Returns True if journal output or a specific terminal error resolved the marker. Never raises — caller is best-effort. """ try: @@ -3033,29 +3237,39 @@ def _retry_journal_recovery_in_place( exc_info=True, ) return False - tail_len_before = len(session.messages) - ok = _append_journaled_partial_output( - session, stream_id, dedupe_existing=True, + recovered_output, terminal_error_recovered = ( + _recover_journaled_output_and_terminal_error( + session, + stream_id, + dedupe_existing=True, + ) ) - if ok: - msg['content'] = _INTERRUPTED_RECOVERED_WORDING - _strip_journal_retry_meta(msg) + if recovered_output or terminal_error_recovered: + if not terminal_error_recovered: + msg['content'] = _INTERRUPTED_RECOVERED_WORDING + _strip_journal_retry_meta(msg) # The journaled rows were appended at the end of messages; - # only the rows past the previous tail count as "newly - # journaled" and need to move above the marker. - _ = tail_len_before # informational; helper below scans + # move them above the marker before either retaining its + # interrupted wording or replacing it with a specific terminal + # error from that same stream. _reorder_journal_tail_above_marker(session, idx) + if terminal_error_recovered: + session.messages = [ + message + for message in session.messages + if message is not msg + ] try: session.save(touch_updated_at=False) except Exception: logger.debug( - "save() failed while promoting marker for session %s", + "save() failed while applying lazy journal recovery for session %s", getattr(session, 'session_id', '?'), exc_info=True, ) logger.info( - "Session %s: lazy journal-recovery promoted marker for " - "stream %s after %d attempts", + "Session %s: lazy journal-recovery applied stream %s " + "after %d attempts", getattr(session, 'session_id', '?'), stream_id, attempts, @@ -3132,6 +3346,10 @@ def _apply_core_sync_or_error_marker( return False if require_stream_dead and session.active_stream_id in _active_stream_ids(): return False + _stream_id = stream_id_for_recheck or session.active_stream_id + _terminal_recovery = _recoverable_unsaved_gateway_terminal_error( + session, _stream_id, + ) # When messages is already non-empty, do not overwrite history from any core # transcript. The pending user turn may still be the only durable copy of a @@ -3152,7 +3370,6 @@ def _apply_core_sync_or_error_marker( session.messages[-1], session.pending_user_message, ) - _stream_id = stream_id_for_recheck or session.active_stream_id _pending_started_at = session.pending_started_at if _run_journal_terminal_state(session, _stream_id) == 'completed': if not (_already_checkpointed or _latest_user_matches_pending_text(session.messages, session.pending_user_message)): @@ -3189,22 +3406,26 @@ def _apply_core_sync_or_error_marker( if session.pending_attachments: recovered['attachments'] = list(session.pending_attachments) _append_recovered_turn_to_context(session, recovered) - recovered_output = _append_journaled_partial_output( - session, - _stream_id, + recovered_output, terminal_error_recovered = ( + _recover_journaled_output_and_terminal_error( + session, + _stream_id, + terminal_recovery=_terminal_recovery, + ) ) session.active_stream_id = None session.pending_user_message = None session.pending_attachments = [] session.pending_started_at = None session.pending_user_source = None - session.messages.append( - _build_recovery_marker_with_retry_hook( - recovered_output=recovered_output, - stream_id=_stream_id, - pending_started_at=_pending_started_at, + if not terminal_error_recovered: + session.messages.append( + _build_recovery_marker_with_retry_hook( + recovered_output=recovered_output, + stream_id=_stream_id, + pending_started_at=_pending_started_at, + ) ) - ) session.save(touch_updated_at=touch_updated_at) logger.info( "Session %s: recovered pending user turn (messages non-empty), added error marker", @@ -3219,7 +3440,6 @@ def _apply_core_sync_or_error_marker( core = json.load(f) core_messages = core.get('messages', []) if core_messages: - _stream_id = stream_id_for_recheck or session.active_stream_id session.messages = core_messages session.tool_calls = core.get('tool_calls', []) for field in ('input_tokens', 'output_tokens', 'estimated_cost'): @@ -3243,13 +3463,19 @@ def _apply_core_sync_or_error_marker( if ( _pending_text and not _tail_user_already_checkpointed - and _run_journal_has_visible_output(session, _stream_id) + and ( + _run_journal_has_visible_output(session, _stream_id) + or _terminal_recovery is not None + ) ): _append_recovered_pending_turn(session, timestamp=_recovered_ts) - recovered_output = _append_journaled_partial_output( - session, - _stream_id, - dedupe_existing=True, + recovered_output, terminal_error_recovered = ( + _recover_journaled_output_and_terminal_error( + session, + _stream_id, + dedupe_existing=True, + terminal_recovery=_terminal_recovery, + ) ) _pending_started_at = session.pending_started_at session.active_stream_id = None @@ -3257,7 +3483,7 @@ def _apply_core_sync_or_error_marker( session.pending_attachments = [] session.pending_started_at = None session.pending_user_source = None - if recovered_output: + if recovered_output and not terminal_error_recovered: session.messages.append( _interrupted_recovery_marker( recovered_output=True, @@ -3294,24 +3520,27 @@ def _apply_core_sync_or_error_marker( if isinstance(session.pending_started_at, (int, float)) and session.pending_started_at > 0: _recovered_ts = int(session.pending_started_at) _append_recovered_pending_turn(session, timestamp=_recovered_ts) - recovered_output = _append_journaled_partial_output( - session, - stream_id_for_recheck or session.active_stream_id, + recovered_output, terminal_error_recovered = ( + _recover_journaled_output_and_terminal_error( + session, + _stream_id, + terminal_recovery=_terminal_recovery, + ) ) - _stream_id = stream_id_for_recheck or session.active_stream_id _pending_started_at = session.pending_started_at session.active_stream_id = None session.pending_user_message = None session.pending_attachments = [] session.pending_started_at = None session.pending_user_source = None - session.messages.append( - _build_recovery_marker_with_retry_hook( - recovered_output=recovered_output, - stream_id=_stream_id, - pending_started_at=_pending_started_at, + if not terminal_error_recovered: + session.messages.append( + _build_recovery_marker_with_retry_hook( + recovered_output=recovered_output, + stream_id=_stream_id, + pending_started_at=_pending_started_at, + ) ) - ) session.save(touch_updated_at=touch_updated_at) logger.info("Session %s: no core transcript found, added error marker", sid) return True diff --git a/api/run_journal.py b/api/run_journal.py index cabdacfa..ab8d0a7e 100644 --- a/api/run_journal.py +++ b/api/run_journal.py @@ -484,14 +484,32 @@ def read_run_events( } +def select_authoritative_terminal_event(events: Iterable[dict]) -> dict | None: + """Return the terminal event that owns the run's settled outcome. + + ``stream_end`` is transport closure, so a preceding semantic terminal event + (done, cancel, or error) remains authoritative. Among semantic terminal + events, the latest journal row wins. + """ + terminal_events = [ + event + for event in events + if isinstance(event, dict) and event.get("terminal") + ] + return next( + ( + event + for event in reversed(terminal_events) + if event.get("event") != "stream_end" + ), + terminal_events[-1] if terminal_events else None, + ) + + def _summary_from_events(session_id: str, run_id: str, events: Iterable[dict]) -> dict: ordered = [event for event in events if isinstance(event, dict)] last = ordered[-1] if ordered else None - terminal_events = [event for event in ordered if event.get("terminal")] - terminal = next( - (event for event in reversed(terminal_events) if event.get("event") != "stream_end"), - terminal_events[-1] if terminal_events else None, - ) + terminal = select_authoritative_terminal_event(ordered) status = terminal.get("terminal_state") if terminal else ("running" if ordered else "unknown") return { "session_id": str(session_id), diff --git a/api/streaming.py b/api/streaming.py index f9dc14eb..f83a6d35 100644 --- a/api/streaming.py +++ b/api/streaming.py @@ -566,8 +566,11 @@ def _is_agent_compression_start_status(kind: str, message: str) -> bool: auto-compression. Positive markers below match the agent emitters in hermes-agent - (``turn_context`` preflight, ``conversation_loop`` pre-API / 413 / too-large, - ``conversation_compression`` compaction status). Explicitly reject skip / + (``conversation_loop`` pre-API / 413 / too-large, + ``conversation_compression`` compaction status). Preflight compression + (``turn_context``) is intentionally excluded — the later authoritative + ``Compacting context`` marker is the signal that compression actually + proceeded. Explicitly reject skip / defer notices so "Skipping preflight compression…" never surfaces as a running compress divider. """ @@ -587,8 +590,7 @@ def _is_agent_compression_start_status(kind: str, message: str) -> bool: if 'compressed' in m and 'compressing' not in m and 'compression attempt' not in m: return False return ( - 'preflight compression:' in m - or 'pre-api compression:' in m + 'pre-api compression:' in m or 'compacting context' in m or 'context too large' in m or '— compressing (' in m diff --git a/static/ui.js b/static/ui.js index eada54bb..3b51b5a4 100644 --- a/static/ui.js +++ b/static/ui.js @@ -961,6 +961,7 @@ function _captureMessageViewportAnchor(){ // anchor's topOffset is stale and realigning to it would yank a still reader // backward (issue #5637). scrollHeightAtCapture:container.scrollHeight, + inputGeneration:typeof _messageScrollInputGeneration==='number' ? _messageScrollInputGeneration : 0, }; } } @@ -5680,6 +5681,9 @@ let _lastMessageClientHeight=null; // #4702: track scroller height to ignore i // window after load as suppressed. let _lastNonMessageScrollIntentMs=-Infinity; let _messageUserUnpinned=false; +// A monotonic ownership token lets delayed restores distinguish reader input +// that happened after a snapshot from input that merely happened recently. +let _messageScrollInputGeneration=0; let _bottomSettleToken=0; let _settleRAF=0; let _settleRO=null; @@ -5763,6 +5767,9 @@ function _recordNonMessageScrollIntent(e){ const target=e&&e.target; if(!el||!target) return; if(!el.contains(target)){ _lastNonMessageScrollIntentMs=performance.now(); return; } + if(e.type==='touchmove'||(typeof e.deltaY==='number'&&e.deltaY!==0)){ + if(typeof _messageScrollInputGeneration==='number') _messageScrollInputGeneration++; + } // #4970: record ANY upward message-pane wheel motion as recent wheel intent, // including gentle low-delta trackpad wheels (e.g. deltaY:-5) that never reach // the decisive -30 sticky-unpin threshold below. The post-render artifact @@ -5966,7 +5973,10 @@ if(typeof window!=='undefined'){ const el=document.getElementById('messages'); if(!el) return; el.addEventListener('pointerdown',(e)=>{ - if(e.target===el&&e.offsetX>=el.clientWidth) _scrollbarDragActive=true; + if(e.target===el&&e.offsetX>=el.clientWidth){ + _scrollbarDragActive=true; + if(typeof _messageScrollInputGeneration==='number') _messageScrollInputGeneration++; + } },{passive:true}); window.addEventListener('pointerup',()=>{ if(!_scrollbarDragActive) return; @@ -6012,6 +6022,7 @@ if(typeof window!=='undefined'){ // contains the focus, or the pointer is over it (keyboard scroll w/o focus). if(a===el||el.contains(a)||el.matches(':hover')){ const now=performance.now(); + if(typeof _messageScrollInputGeneration==='number') _messageScrollInputGeneration++; _lastMessageKeyScrollIntentMs=now; const bottomDistance=el.scrollHeight-el.scrollTop-el.clientHeight; if(bottomDistance>120) _lastMessageScrollIntentMs=now; @@ -12714,6 +12725,23 @@ function _prepareLiveAnchorScrollRebuildGuard(scrollSnapshot){ }, }; } +function _restoreLiveAnchorScrollSnapshotAfterRebuild(scrollSnapshot, scrollRebuildGuard){ + if(!scrollSnapshot) return; + const hasHeightGuard=!!(scrollRebuildGuard&&scrollRebuildGuard.release); + // Pinned renders have no height guard to release, but still need the same + // queued ownership check: a reader can provide real input before the frame + // settles, and that input must win over the captured tail position. + if(!hasHeightGuard&&scrollSnapshot.pinned!==true) return; + requestAnimationFrame(()=>{ + if(hasHeightGuard) scrollRebuildGuard.release(); + // Only re-restore the unpinned snapshot if the reader is STILL unpinned at + // rAF time. If they re-pinned between guard-engage and this frame, the + // stale re-restore would yank them back off the bottom (Opus gate finding). + if(scrollSnapshot.pinned===true||_messageUserUnpinned){ + _restoreMessageScrollSnapshotSameFrame(scrollSnapshot); + } + }); +} function _resetMismatchedLiveAssistantTurnForSession(turn, sessionId){ const sid=String(sessionId||''); if(!turn||!sid||!turn.dataset) return false; @@ -12863,15 +12891,7 @@ function renderLiveAnchorActivityScene(streamId, scene, opts){ _dedupeLiveProcessedWorklogAnchors(turn); if(typeof _moveLiveRunStatusToTurnEnd==='function') _moveLiveRunStatusToTurnEnd(); _restoreMessageScrollSnapshotSameFrame(scrollSnapshot); - if(scrollRebuildGuard&&scrollRebuildGuard.release){ - requestAnimationFrame(()=>{ - scrollRebuildGuard.release(); - // Only re-restore the unpinned snapshot if the reader is STILL unpinned at - // rAF time. If they re-pinned between guard-engage and this frame, the - // stale re-restore would yank them back off the bottom (Opus gate finding). - if(_messageUserUnpinned) _restoreMessageScrollSnapshotSameFrame(scrollSnapshot); - }); - } + _restoreLiveAnchorScrollSnapshotAfterRebuild(scrollSnapshot,scrollRebuildGuard); if(!scrollRebuildGuard.readerAwayFromBottom&&typeof scrollIfPinned==='function') scrollIfPinned(); return true; } @@ -12977,12 +12997,7 @@ function _renderLiveAnchorActivitySceneTransparent(streamId, scene, opts){ if(renderedRows.length) _syncTransparentEventControls(turn); if(typeof _moveLiveRunStatusToTurnEnd==='function') _moveLiveRunStatusToTurnEnd(); _restoreMessageScrollSnapshotSameFrame(scrollSnapshot); - if(scrollRebuildGuard&&scrollRebuildGuard.release){ - requestAnimationFrame(()=>{ - scrollRebuildGuard.release(); - if(_messageUserUnpinned) _restoreMessageScrollSnapshotSameFrame(scrollSnapshot); - }); - } + _restoreLiveAnchorScrollSnapshotAfterRebuild(scrollSnapshot,scrollRebuildGuard); if(!scrollRebuildGuard.readerAwayFromBottom&&typeof scrollIfPinned==='function') scrollIfPinned(); return !!renderedRows.length; } @@ -14985,10 +15000,41 @@ function _captureMessageScrollSnapshot(){ top:el.scrollTop, bottom, scrollHeight:el.scrollHeight, + inputGeneration:typeof _messageScrollInputGeneration==='number' ? _messageScrollInputGeneration : 0, pinned:readerAwayFromBottom?false:_shouldFollowMessagesOnDomReplace(), userUnpinned:readerAwayFromBottom?true:_messageUserUnpinned, }; } +function _messageScrollSnapshotInputChanged(snapshot){ + if(!snapshot) return false; + const captured=Number(snapshot.inputGeneration); + const current=typeof _messageScrollInputGeneration==='number' ? _messageScrollInputGeneration : captured; + return Number.isFinite(captured)&&Number.isFinite(current)&¤t!==captured; +} +function _abandonMessageScrollSnapshot(){ + const el=$('messages'); + if(!el){ + _messageUserUnpinned=true; + _scrollPinned=false; + _nearBottomCount=0; + return; + } + _lastScrollTop=el.scrollTop||0; + _lastMessageClientHeight=el.clientHeight||0; + // A generation mismatch abandons only the stale snapshot write. Reconcile + // ownership from the live viewport so a reader who moved down to the true + // bottom is immediately re-pinned instead of being stranded sticky-unpinned. + const bottomDistance=el.scrollHeight-el.scrollTop-el.clientHeight; + if(bottomDistance<=80){ + _messageUserUnpinned=false; + _scrollPinned=true; + _nearBottomCount=2; + }else{ + _messageUserUnpinned=true; + _scrollPinned=false; + _nearBottomCount=0; + } +} function _restorePinnedMessageScrollSnapshot(snapshot){ const el=$('messages'); if(!el||!snapshot||snapshot.pinned!==true||snapshot.userUnpinned===true) return false; @@ -15015,6 +15061,10 @@ function _restoreMessageScrollSnapshot(snapshot){ // activity rebuilds can remount an older top-of-viewport anchor and yank a // pinned streaming transcript upward. Semantic anchors remain for manual // unpinned reading positions below. + if(typeof _messageScrollSnapshotInputChanged==='function'&&_messageScrollSnapshotInputChanged(snapshot)){ + if(typeof _abandonMessageScrollSnapshot==='function') _abandonMessageScrollSnapshot(); + return; + } if(_restorePinnedMessageScrollSnapshot(snapshot)) return; let restoredViaAnchor=(snapshot.anchor&&typeof _restoreMessageViewportAnchor==='function') ? _restoreMessageViewportAnchor(snapshot.anchor,0) @@ -15233,7 +15283,14 @@ function _restoreMessageScrollSnapshotSameFrame(snapshot){ // Same-frame live DOM updates (tool/worklog/activity rows) are the hot path for // streaming. Pinned followers must stay tail-relative here too; restoring the // semantic viewport anchor is only safe for explicitly unpinned readers. + if(typeof _messageScrollSnapshotInputChanged==='function'&&_messageScrollSnapshotInputChanged(snapshot)){ + if(typeof _abandonMessageScrollSnapshot==='function') _abandonMessageScrollSnapshot(); + return; + } if(_restorePinnedMessageScrollSnapshot(snapshot)) return; + // A delayed rAF restore must not overwrite a position the reader changed + // after capture. Recent-intent timestamps are lossy; the generation is + // monotonic and therefore preserves snapshot ownership exactly. let restoredViaAnchor=(snapshot.anchor&&typeof _restoreMessageViewportAnchor==='function') ? _restoreMessageViewportAnchor(snapshot.anchor,0) : false; @@ -15245,6 +15302,10 @@ function _restoreMessageScrollSnapshotSameFrame(snapshot){ if(!restoredViaAnchor){ const maxTop=Math.max(0,el.scrollHeight-el.clientHeight); const bottom=Number(snapshot.bottom); + // Mobile/touch viewports have native overflow anchoring to hold an + // unpinned reader across a rebuild. Desktop deliberately disables that + // browser behavior, so it must continue into the explicit fallback below. + const _fbTouchHold=(typeof _isTouchLikeMessageViewport==='function' && _isTouchLikeMessageViewport(el)); // #5637: when the reader has scrolled UP into history (userUnpinned) and the // semantic anchor restore failed, do NOT snap scrollTop to the captured // ABSOLUTE snapshot.top. During streaming, the live activity-scene refresh @@ -15254,7 +15315,7 @@ function _restoreMessageScrollSnapshotSameFrame(snapshot){ // untouched lets the browser's own scroll anchoring hold the reader's // position. Pinned / near-bottom readers still get the tail-relative restore // below (that path is correct and must run). - if(snapshot.userUnpinned===true&&snapshot.pinned!==true){ + if(snapshot.userUnpinned===true&&snapshot.pinned!==true&&_fbTouchHold){ _lastScrollTop=el.scrollTop;_lastMessageClientHeight=el.clientHeight; _messageUserUnpinned=true; _scrollPinned=false; @@ -15285,7 +15346,6 @@ function _restoreMessageScrollSnapshotSameFrame(snapshot){ const _grewSinceSnap=Number.isFinite(_snapSH)&&_snapSH>0&&(el.scrollHeight-_snapSH)>4; const _fbActiveIntent=(typeof _recentMessageScrollIntent==='function' && _recentMessageScrollIntent()) || (typeof _recentMessageTouchScrollIntent==='function' && _recentMessageTouchScrollIntent()); - const _fbTouchHold=(typeof _isTouchLikeMessageViewport==='function' && _isTouchLikeMessageViewport(el)); if(_fbTouchHold && snapshot.pinned!==true && _grewSinceSnap && !_fbActiveIntent && Math.abs((Math.max(0,Math.min(target,maxTop)))-el.scrollTop)>8){ _lastScrollTop=el.scrollTop;_lastMessageClientHeight=el.clientHeight; diff --git a/tests/test_auto_compression_card.py b/tests/test_auto_compression_card.py index a6bfa8fd..837ff474 100644 --- a/tests/test_auto_compression_card.py +++ b/tests/test_auto_compression_card.py @@ -473,7 +473,13 @@ def test_agent_status_callback_emits_compressing_and_warning_events(): def test_agent_compression_start_status_matches_real_emitters_only(): # Real start notices from hermes-agent emitters - assert _is_agent_compression_start_status( + # Preflight compression is intentionally excluded — the later + # authoritative ``Compacting context`` marker from + # conversation_compression is the signal that compression + # actually proceeded, and the preflight status can fire even + # when compression exits before compaction (e.g. durable + # guard refresh, Codex-Hermes mode with no active thread). + assert not _is_agent_compression_start_status( "lifecycle", "📦 Preflight compression: ~101,000 tokens >= 96,000 threshold. This may take a moment.", ) diff --git a/tests/test_issue3479_ios_stream_scroll_jump.py b/tests/test_issue3479_ios_stream_scroll_jump.py index c803abf3..1458f361 100644 --- a/tests/test_issue3479_ios_stream_scroll_jump.py +++ b/tests/test_issue3479_ios_stream_scroll_jump.py @@ -74,7 +74,7 @@ def test_live_anchor_worklog_rebuild_restores_snapshot_before_follow_settle(): dedupe_idx = body.index("_dedupeLiveProcessedWorklogAnchors(turn);") move_status_idx = body.index("_moveLiveRunStatusToTurnEnd();") restore_idx = body.index("_restoreMessageScrollSnapshotSameFrame(scrollSnapshot);") - release_idx = body.index("if(scrollRebuildGuard&&scrollRebuildGuard.release)") + release_idx = body.index("_restoreLiveAnchorScrollSnapshotAfterRebuild(scrollSnapshot,scrollRebuildGuard);") settle_idx = body.index("if(!scrollRebuildGuard.readerAwayFromBottom&&typeof scrollIfPinned==='function') scrollIfPinned();") assert capture_idx < guard_idx < remove_idx < restore_detail_idx < dedupe_idx < move_status_idx < restore_idx < release_idx < settle_idx diff --git a/tests/test_issue5367_transparent_live_row_reconcile.py b/tests/test_issue5367_transparent_live_row_reconcile.py index 7017d05d..04b4f16d 100644 --- a/tests/test_issue5367_transparent_live_row_reconcile.py +++ b/tests/test_issue5367_transparent_live_row_reconcile.py @@ -311,6 +311,8 @@ eval(extractFunc('_bindTransparentFadeCleanup')); eval(extractFunc('_appendTransparentFadeText')); eval(extractFunc('_refreshTransparentFadeProseRow')); eval(extractFunc('_refreshTransparentLiveRow')); +eval(extractFunc('_restoreLiveAnchorScrollSnapshotAfterRebuild')); +eval(extractFunc('_restoreLiveAnchorScrollSnapshotAfterRebuild')); eval(extractFunc('_renderLiveAnchorActivitySceneTransparent')); const firstScene = {{ @@ -1264,6 +1266,7 @@ eval(extractFunc('_transparentLiveRowInteractiveState')); eval(extractFunc('_rehydrateTransparentLiveRow')); eval(extractFunc('_refreshTransparentThinkingLiveRow')); eval(extractFunc('_refreshTransparentLiveRow')); +eval(extractFunc('_restoreLiveAnchorScrollSnapshotAfterRebuild')); eval(extractFunc('_renderLiveAnchorActivitySceneTransparent')); const firstScene = { version:'activity_scene_v1', activity_rows:[{ row_id:'row-tool', role:'tool', source_event_type:'tool_delta', toolName:'old_tool' }] }; @@ -1601,6 +1604,7 @@ eval(extractFunc('_transparentLiveRowInteractiveState')); eval(extractFunc('_rehydrateTransparentLiveRow')); eval(extractFunc('_refreshTransparentThinkingLiveRow')); eval(extractFunc('_refreshTransparentLiveRow')); +eval(extractFunc('_restoreLiveAnchorScrollSnapshotAfterRebuild')); eval(extractFunc('_renderLiveAnchorActivitySceneTransparent')); global._copyEventToClipboard = (row) => {{ const tc = row && row._tcData || {{}}; diff --git a/tests/test_issue5637_mobile_scroll_clamp_jumpback.py b/tests/test_issue5637_mobile_scroll_clamp_jumpback.py index d69c63f9..c999d597 100644 --- a/tests/test_issue5637_mobile_scroll_clamp_jumpback.py +++ b/tests/test_issue5637_mobile_scroll_clamp_jumpback.py @@ -94,17 +94,19 @@ def test_content_visibility_scoped_to_user_rows(): ) -def test_restore_same_frame_holds_position_for_unpinned_reader(): - """The absolute-top fallback must be skipped for an unpinned reader. +def test_restore_same_frame_holds_position_for_unpinned_touch_reader(): + """The absolute-top fallback must be skipped when native anchoring can hold. When the reader is scrolled up (userUnpinned) and the anchor restore failed, _restoreMessageScrollSnapshotSameFrame must NOT write an absolute snapshot.top (it goes stale and nudges the viewport backward as scrollHeight grows, #5637). - It should hold position and return before the el.scrollTop write. + It should hold position and return before the el.scrollTop write. Desktop + intentionally falls through to the explicit fallback because it has no + native overflow-anchor layer. """ fn = _restore_same_frame_fn() - assert "snapshot.userUnpinned===true&&snapshot.pinned!==true" in fn, ( - "The fallback must guard the unpinned/anchor-failed case (#5637)." + assert "snapshot.userUnpinned===true&&snapshot.pinned!==true&&_fbTouchHold" in fn, ( + "The fallback must guard the unpinned/anchor-failed touch case (#5637)." ) # The guard must sit BEFORE the absolute-top scrollTop write and short-circuit it. guard_idx = fn.index("snapshot.userUnpinned===true&&snapshot.pinned!==true") diff --git a/tests/test_issue5637_stale_anchor_guard.py b/tests/test_issue5637_stale_anchor_guard.py index 3b83c094..c6d76cc8 100644 --- a/tests/test_issue5637_stale_anchor_guard.py +++ b/tests/test_issue5637_stale_anchor_guard.py @@ -52,6 +52,185 @@ def _run_node(source: str) -> str: return result.stdout.strip() +def _extract_js_function(js: str, name: str) -> str: + """Extract a production function without runtime eval/source execution.""" + marker = f"function {name}(" + start = js.index(marker) + brace = js.index("{", start) + depth = 0 + quote = None + escaped = False + line_comment = False + block_comment = False + index = brace + while index < len(js): + char = js[index] + next_char = js[index + 1] if index + 1 < len(js) else "" + if line_comment: + if char == "\n": + line_comment = False + index += 1 + continue + if block_comment: + if char == "*" and next_char == "/": + block_comment = False + index += 2 + continue + index += 1 + continue + if quote: + if escaped: + escaped = False + elif char == "\\": + escaped = True + elif char == quote: + quote = None + index += 1 + continue + if char == "/" and next_char == "/": + line_comment = True + index += 2 + continue + if char == "/" and next_char == "*": + block_comment = True + index += 2 + continue + if char in "'\"`": + quote = char + index += 1 + continue + if char == "{": + depth += 1 + elif char == "}": + depth -= 1 + if depth == 0: + return js[start : index + 1] + index += 1 + raise AssertionError(f"JavaScript function {name} did not close") + + +def _live_render_ownership_harness(*, move_to_bottom: bool) -> str: + """Compose real input production with the production queued live-render restore.""" + js = UI_JS_PATH.read_text(encoding="utf-8") + functions = [ + "renderLiveAnchorActivityScene", + "_recordNonMessageScrollIntent", + "_captureMessageScrollSnapshot", + "_messageScrollSnapshotInputChanged", + "_abandonMessageScrollSnapshot", + "_restorePinnedMessageScrollSnapshot", + "_restoreMessageScrollSnapshotSameFrame", + "_prepareLiveAnchorScrollRebuildGuard", + "_restoreLiveAnchorScrollSnapshotAfterRebuild", + ] + production = "\n".join(_extract_js_function(js, name) for name in functions) + target_delta = 100 if move_to_bottom else -100 + final_top = 90026 if move_to_bottom else 89000 + return production + f""" +let writes=[]; +let _messageScrollInputGeneration=0; +let _messageUserUnpinned={str(move_to_bottom).lower()}; +let _scrollPinned={str(not move_to_bottom).lower()}; +let _nearBottomCount=0; +let _lastScrollTop=null, _lastMessageClientHeight=null; +let _programmaticScroll=false, _programmaticScrollSetAt=0; +let recordWrites=true; +const callbacks=[]; +const performance={{now(){{return 1;}}}}; +const el={{ + scrollHeight:90453, clientHeight:427, + get scrollTop(){{return this._top;}}, + set scrollTop(value){{if(recordWrites) writes.push(Math.round(value)); this._top=value;}}, + _top:{'89577' if move_to_bottom else '90026'}, + contains(target){{return target===this;}}, + querySelector(){{return null;}}, +}}; +const document={{getElementById(id){{return id==='messages'?el:null;}}}}; +const msgInner={{style:{{}},dataset:{{}}}}; +const emptyState={{style:{{display:''}}}}; +const turn={{ + dataset:{{}}, + setAttribute(){{}}, + querySelectorAll(){{return [];}}, +}}; +const blocks={{querySelectorAll(){{return [];}}}}; +function $(id){{ + if(id==='messages') return el; + if(id==='msgInner') return msgInner; + if(id==='emptyState') return emptyState; + if(id==='liveAssistantTurn') return turn; + return null; +}} +function _captureMessageViewportAnchor(){{return null;}} +function _shouldFollowMessagesOnDomReplace(){{return true;}} +function _deferClearProgrammaticScroll(){{}} +function _cancelBottomSettle(){{}} +function _restoreMessageViewportAnchor(){{return false;}} +function _remountMessageViewportAnchor(){{return false;}} +function _desktopAnchorRealignDelta(){{return null;}} +function _isTouchLikeMessageViewport(){{return false;}} +function _recentMessageScrollIntent(){{return false;}} +function _recentMessageTouchScrollIntent(){{return false;}} +function _restoreMessageScrollSnapshotSameFrameFallback(){{}} +function requestAnimationFrame(callback){{callbacks.push(callback);}} +function chatActivityMode(){{return 'compact_worklog';}} +function isSimplifiedToolCalling(){{return true;}} +const S={{session:{{session_id:'sid-1',pending_started_at:1}},activeStreamId:'stream-1'}}; +function _anchorSceneRowsForRendering(){{return [];}} +function _assistantTurnBlocks(){{return blocks;}} +function _captureWorklogDetailDisclosureState(){{return null;}} +function _anchorSceneWorklogGroup(){{return {{}};}} +function _renderAnchorSceneRowsIntoWorklog(){{return true;}} +function _restoreWorklogDetailDisclosureState(){{}} +function _startActivityElapsedTimer(){{}} +function _dedupeLiveProcessedWorklogAnchors(){{}} +function _moveLiveRunStatusToTurnEnd(){{}} +function scrollIfPinned(){{}} +const rendered=renderLiveAnchorActivityScene('stream-1',{{activity_rows:[]}},{{sessionId:'sid-1'}}); +const initialWrites=writes.slice(); +writes=[]; +const guardQueued=callbacks.length>0; +recordWrites=false; +_recordNonMessageScrollIntent({{target:el,deltaY:{target_delta}}}); +el.scrollTop={final_top}; +recordWrites=true; +callbacks.shift()(); +console.log(JSON.stringify({{rendered,guardQueued,initialWrites,writes, finalScrollTop:el.scrollTop, + messageUserUnpinned:_messageUserUnpinned, scrollPinned:_scrollPinned, + generation:_messageScrollInputGeneration}})); +""" + + +def test_live_render_queue_preserves_real_upward_input_from_pinned_capture(): + """The real producer and queued live-render callback preserve upward input.""" + result = json.loads(_run_node(_live_render_ownership_harness(move_to_bottom=False))) + assert result == { + "rendered": True, + "guardQueued": True, + "initialWrites": [90026], + "writes": [], + "finalScrollTop": 89000, + "messageUserUnpinned": True, + "scrollPinned": False, + "generation": 1, + } + + +def test_live_render_queue_repins_real_downward_input_at_bottom(): + """The real producer and queued live-render callback re-pin at true bottom.""" + result = json.loads(_run_node(_live_render_ownership_harness(move_to_bottom=True))) + assert result == { + "rendered": True, + "guardQueued": True, + "initialWrites": [89577], + "writes": [], + "finalScrollTop": 90026, + "messageUserUnpinned": False, + "scrollPinned": True, + "generation": 1, + } + + def _extract_func_script(js: str) -> str: prelude = "const src = " + json.dumps(js) + ";\n" body = r""" @@ -175,8 +354,12 @@ def test_realign_allows_on_desktop_no_native_anchor(): def _fallback_harness(*, snapshot_scroll_height, cur_scroll_height, snapshot_top, active_intent: bool = False, touch_like: bool = True, init_scroll_top: int = 90030, + snapshot_user_unpinned: bool = False, + snapshot_pinned: bool = False, anchor=None, anchor_row_content_pos=None, container_top: int = 0, - top_pad_now=None) -> str: + top_pad_now=None, input_generation: int = 0, + delayed_input_generation=None, delayed_scroll_top: int = 89000, + restore_fn: str = "_restoreMessageScrollSnapshotSameFrame") -> str: """Node harness for the absolute-fallback path of _restoreMessageScrollSnapshotSameFrame. SCROLL-DEPENDENT geometry (round-3 gate-cert requirement): the anchor row's @@ -202,6 +385,7 @@ def _fallback_harness(*, snapshot_scroll_height, cur_scroll_height, snapshot_top return _extract_func_script(js) + f""" let writes = []; let stTop = {init_scroll_top}; +let _messageScrollInputGeneration = {input_generation}; const CONTAINER_TOP = {container_top}; const ROW_CONTENT_POS = {row_pos_js}; const ROW_PRESENT = {"true" if row_present else "false"}; @@ -230,7 +414,12 @@ function _recentMessageScrollIntent(){{ return {intent_js}; }} function _recentMessageTouchScrollIntent(){{ return {intent_js}; }} function _isTouchLikeMessageViewport(){{ return {touch_js}; }} // realign path fails (no anchor restore) so execution reaches the absolute fallback -function _restorePinnedMessageScrollSnapshot(){{ return false; }} +function _restorePinnedMessageScrollSnapshot(snapshot){{ + if(snapshot.pinned!==true) return false; + const pinnedTarget=el.scrollHeight-el.clientHeight-(Number(snapshot.bottom)||0); + el.scrollTop=pinnedTarget; + return true; +}} function _restoreMessageViewportAnchor(){{ return false; }} function _remountMessageViewportAnchor(){{ return false; }} let _messageUserUnpinned = false; let _scrollPinned = true; let _nearBottomCount = 5; @@ -241,10 +430,13 @@ function _deferClearProgrammaticScroll(){{}} function requestAnimationFrame(cb){{ cb(); }} function setTimeout(cb){{ cb(); return 1; }} const snapshot = {{ anchor: {anchor_js}, top: {snapshot_top}, bottom: 40, - scrollHeight: {sh}, pinned: false, userUnpinned: false }}; + scrollHeight: {sh}, inputGeneration: {input_generation}, pinned: {str(snapshot_pinned).lower()}, userUnpinned: {str(snapshot_user_unpinned).lower()} }}; eval(extractFunc('_desktopAnchorRealignDelta')); -eval(extractFunc('_restoreMessageScrollSnapshotSameFrame')); -_restoreMessageScrollSnapshotSameFrame(snapshot); +eval(extractFunc('_messageScrollSnapshotInputChanged')); +eval(extractFunc('_abandonMessageScrollSnapshot')); +eval(extractFunc({json.dumps(restore_fn)})); +{restore_fn}(snapshot); +{f"stTop = {delayed_scroll_top}; _messageScrollInputGeneration = {delayed_input_generation}; {restore_fn}(snapshot);" if delayed_input_generation is not None else ""} console.log(JSON.stringify({{ wrote: writes.length, writes, messageUserUnpinned: _messageUserUnpinned, scrollPinned: _scrollPinned, finalScrollTop: Math.round(stTop) }})); """ @@ -312,6 +504,75 @@ def test_fallback_allows_snapshot_top_on_desktop_no_native_anchor(): assert m["messageUserUnpinned"] is False and m["scrollPinned"] is True +def test_unpinned_anchor_miss_restores_on_desktop_without_native_anchor(): + """A desktop reader who deliberately scrolled up cannot be left to native + anchoring: desktop disables it. If the semantic row is temporarily absent, + the rebuild may clamp scrollTop toward zero before this function runs, so the + fallback must restore the snapshot rather than taking the mobile no-write exit. + """ + m = json.loads(_run_node(_fallback_harness( + snapshot_scroll_height=90000, cur_scroll_height=90453, snapshot_top=89577, + active_intent=False, touch_like=False, snapshot_user_unpinned=True, + init_scroll_top=0, + ))) + assert m["writes"] == [89577] + assert m["messageUserUnpinned"] is True and m["scrollPinned"] is False + + +def test_delayed_restore_cannot_overwrite_new_desktop_input(): + """The synchronous restore owns generation 0, then the reader moves to 89000 + before the delayed rAF restore. The second call must abandon the stale snapshot + instead of writing its old 89577 target.""" + m = json.loads(_run_node(_fallback_harness( + snapshot_scroll_height=90453, cur_scroll_height=90453, snapshot_top=89577, + active_intent=False, touch_like=False, snapshot_user_unpinned=True, + init_scroll_top=89577, input_generation=0, delayed_input_generation=1, + ))) + assert m["writes"] == [89577] + assert m["finalScrollTop"] == 89000 + assert m["messageUserUnpinned"] is True and m["scrollPinned"] is False + + +@pytest.mark.parametrize("restore_fn", [ + "_restoreMessageScrollSnapshot", + "_restoreMessageScrollSnapshotSameFrame", +]) +def test_delayed_pinned_restore_cannot_overwrite_new_upward_input(restore_fn): + """A pinned-at-capture snapshot must lose ownership before its tail write. + + The reader moves upward before the delayed restore; the stale pinned snapshot + must not put them back at its captured tail-relative position. + """ + m = json.loads(_run_node(_fallback_harness( + snapshot_scroll_height=90453, cur_scroll_height=90453, snapshot_top=89577, + active_intent=False, touch_like=False, snapshot_pinned=True, + init_scroll_top=89577, input_generation=0, delayed_input_generation=1, + delayed_scroll_top=89000, + restore_fn=restore_fn, + ))) + assert m["writes"] == [89986] + assert m["finalScrollTop"] == 89000 + assert m["messageUserUnpinned"] is True and m["scrollPinned"] is False + + +@pytest.mark.parametrize("restore_fn", [ + "_restoreMessageScrollSnapshot", + "_restoreMessageScrollSnapshotSameFrame", +]) +def test_delayed_unpinned_restore_repins_reader_who_reached_bottom(restore_fn): + """Abandoning a stale unpinned snapshot must reconcile a live bottom position.""" + m = json.loads(_run_node(_fallback_harness( + snapshot_scroll_height=90453, cur_scroll_height=90453, snapshot_top=89577, + active_intent=False, touch_like=False, snapshot_user_unpinned=True, + init_scroll_top=89577, input_generation=0, delayed_input_generation=1, + delayed_scroll_top=90026, + restore_fn=restore_fn, + ))) + assert m["writes"] == [89577] + assert m["finalScrollTop"] == 90026 + assert m["messageUserUnpinned"] is False and m["scrollPinned"] is True + + # ---- round-3 desktop anchor-realign (PR #5742): app's own scrollTop += delta idiom ----- # These use SCROLL-DEPENDENT rect mocks (rect.top = rowContentPos - scrollTop), so the # certified hold numbers are physically realizable in a real browser — the round-2 @@ -479,6 +740,7 @@ def test_predicate_stays_true_on_touch_when_inline_anchor_clobbered_to_none(): assert m["touchLike"] is True + def test_predicate_false_on_desktop_fine_pointer(): """Desktop (fine pointer): matchMedia('(pointer:coarse)') is false and the resting computed overflow-anchor is 'none' -> predicate reports NOT touch, so the guards do @@ -546,4 +808,3 @@ def test_predicate_true_on_android_not_ios(): platform="Linux armv8l", max_touch_points=5, ))) assert m["touchLike"] is True - diff --git a/tests/test_issue5720_reasoning_owner.py b/tests/test_issue5720_reasoning_owner.py index 8a0223e5..6cc31e99 100644 --- a/tests/test_issue5720_reasoning_owner.py +++ b/tests/test_issue5720_reasoning_owner.py @@ -476,6 +476,7 @@ for(const name of [ '_liveAnchorReasoningRowForFallback','_updateLiveAnchorReasoningRowForFallback', '_anchorSceneNodeForRow','_anchorSceneWorklogGroup','_renderAnchorSceneRowsIntoWorklog', 'isLiveAnchorActivitySceneOwner','_projectLiveAnchorActivitySceneForStream', + '_restoreLiveAnchorScrollSnapshotAfterRebuild', '_renderLiveAnchorActivitySceneTransparent','renderLiveAnchorActivityScene', '_renderLiveAnchorActivitySceneForStream','appendThinking','updateThinking', ]) eval(extractFunc(uiSrc,name)); diff --git a/tests/test_live_to_final_anchor_visible_order.py b/tests/test_live_to_final_anchor_visible_order.py index ff830a6c..589ed37b 100644 --- a/tests/test_live_to_final_anchor_visible_order.py +++ b/tests/test_live_to_final_anchor_visible_order.py @@ -1414,6 +1414,7 @@ global._syncToolCallGroupSummary=()=>{{}}; eval(extractFunc('_anchorSceneLiveTokenFinalPrefix')); eval(extractFunc('_anchorSceneTransparentNodeForRow')); eval(extractFunc('renderLiveAnchorActivityScene')); + eval(extractFunc('_restoreLiveAnchorScrollSnapshotAfterRebuild')); eval(extractFunc('_transparentLiveRowKey')); eval(extractFunc('_transparentLiveRowsCompatible')); eval(extractFunc('_transparentLiveRowAttributePairs')); diff --git a/tests/test_run_journal_routes.py b/tests/test_run_journal_routes.py index 5e9beed1..b87f54cb 100644 --- a/tests/test_run_journal_routes.py +++ b/tests/test_run_journal_routes.py @@ -10,6 +10,73 @@ ROOT = Path(__file__).resolve().parents[1] ROUTES_SRC = (ROOT / "api" / "routes.py").read_text(encoding="utf-8") +def test_gateway_terminal_error_save_failure_is_marked_unsaved(monkeypatch, tmp_path): + import api.gateway_chat as gateway_chat + import api.models as models + import api.streaming as streaming + + session = models.Session( + session_id="gateway_terminal_error_save_failed", + workspace=str(tmp_path), + model="gpt-4o", + model_provider="openai", + messages=[{"role": "user", "content": "prompt"}], + context_messages=[], + ) + session.active_stream_id = "gateway_terminal_error_stream" + + def fail_save(*_args, **_kwargs): + raise OSError("forced gateway terminal error save failure") + + session.save = fail_save + monkeypatch.setattr(gateway_chat, "get_session", lambda _sid: session) + monkeypatch.setattr(gateway_chat, "_stream_writeback_is_current", lambda *_args: True) + monkeypatch.setattr(streaming, "_snapshot_and_append_partial_on_error", lambda *_args: None) + + payload = gateway_chat._settle_gateway_terminal_error( + session.session_id, + session.active_stream_id, + str(tmp_path), + "gpt-4o", + "openai", + "gateway exploded", + ) + + assert payload["terminal_session_persisted"] is False + assert "terminal_session_persisted_session_id" not in payload + + +def test_gateway_terminal_error_successful_save_is_marked_persisted(monkeypatch, tmp_path): + import api.gateway_chat as gateway_chat + import api.models as models + import api.streaming as streaming + + session = models.Session( + session_id="gateway_terminal_error_save_succeeds", + workspace=str(tmp_path), + model="gpt-4o", + model_provider="openai", + messages=[{"role": "user", "content": "prompt"}], + context_messages=[], + ) + session.active_stream_id = "gateway_terminal_error_stream" + monkeypatch.setattr(gateway_chat, "get_session", lambda _sid: session) + monkeypatch.setattr(gateway_chat, "_stream_writeback_is_current", lambda *_args: True) + monkeypatch.setattr(streaming, "_snapshot_and_append_partial_on_error", lambda *_args: None) + + payload = gateway_chat._settle_gateway_terminal_error( + session.session_id, + session.active_stream_id, + str(tmp_path), + "gpt-4o", + "openai", + "gateway exploded", + ) + + assert payload["terminal_session_persisted"] is True + assert payload["terminal_session_persisted_session_id"] == session.session_id + + def test_stream_status_exposes_replay_summary(): status_pos = ROUTES_SRC.index('parsed.path == "/api/chat/stream/status"') block = ROUTES_SRC[status_pos : status_pos + 900] diff --git a/tests/test_session_sidecar_repair.py b/tests/test_session_sidecar_repair.py index 9a8d7d4a..e8bb6a6d 100644 --- a/tests/test_session_sidecar_repair.py +++ b/tests/test_session_sidecar_repair.py @@ -21,7 +21,7 @@ from api.models import ( import api.config as config import api.streaming as streaming import api.profiles as profiles -from api.run_journal import append_run_event +from api.run_journal import RunJournalWriter, append_run_event # ── Fixtures ──────────────────────────────────────────────────────────────── @@ -118,6 +118,66 @@ def _register_active_run(stream_id): config.register_active_run(stream_id, session_id="stale_sid", phase="running") +def _journal_gateway_terminal_save_failure( + monkeypatch, + stale_session, + *, + live_messages, + partial_text="Current partial output.", + terminal_error="gateway exploded", +): + """Compose the real gateway producer payload and append it through the real journal.""" + import api.gateway_chat as gateway_chat + + stream_id = stale_session.active_stream_id + live_session = Session( + session_id=stale_session.session_id, + title=stale_session.title, + workspace=stale_session.workspace, + model=stale_session.model, + model_provider=stale_session.model_provider, + messages=[dict(message) for message in live_messages], + context_messages=[dict(message) for message in live_messages], + ) + live_session.pending_user_message = stale_session.pending_user_message + live_session.pending_started_at = stale_session.pending_started_at + live_session.pending_attachments = list(stale_session.pending_attachments) + live_session.pending_user_source = stale_session.pending_user_source + live_session.active_stream_id = stream_id + + def fail_save(*_args, **_kwargs): + raise OSError("forced gateway terminal error save failure") + + monkeypatch.setattr(live_session, "save", fail_save) + monkeypatch.setattr(gateway_chat, "get_session", lambda _sid: live_session) + monkeypatch.setitem(config.STREAM_PARTIAL_TEXT, stream_id, partial_text) + monkeypatch.setitem(config.STREAM_REASONING_TEXT, stream_id, "") + monkeypatch.setitem(config.STREAM_LIVE_TOOL_CALLS, stream_id, []) + + payload = gateway_chat._settle_gateway_terminal_error( + live_session.session_id, + stream_id, + live_session.workspace, + live_session.model, + live_session.model_provider, + terminal_error, + ) + assert payload["terminal_session_persisted"] is False + + writer = RunJournalWriter(live_session.session_id, stream_id) + if partial_text: + writer.append_sse_event("token", {"text": partial_text}) + terminal_event = writer.append_sse_event("apperror", payload) + terminal_message = next( + message + for message in reversed(payload["session"]["messages"]) + if isinstance(message, dict) + and message.get("role") == "assistant" + and message.get("_error") is True + ) + return payload, terminal_event, terminal_message + + class TestRepairStalePendingNoDeadlock: """_repair_stale_pending uses non-blocking lock acquire so callers that already hold the per-session lock (retry_last, undo_last, cancel_stream) @@ -763,6 +823,265 @@ class TestNonEmptyMessagesPendingCleared: assert "partial output above was recovered" in error_msgs[0]["content"] assert "no agent output was recovered" not in error_msgs[0]["content"] + @pytest.mark.parametrize("sidecar_shape", ["non_empty", "core", "empty"]) + def test_gateway_terminal_error_cold_recovery_keeps_turn_identity_and_order( + self, hermes_home, monkeypatch, tmp_path, sidecar_shape, + ): + """The exact producer payload must cold-recover after partial output in every branch.""" + sid = f"gateway_terminal_cold_{sidecar_shape}" + stream_id = f"{sid}_stream" + repeated_error = { + "role": "assistant", + "content": "**Error:** gateway exploded", + "timestamp": int(time.time()) - 300, + "_error": True, + } + history = [ + {"role": "user", "content": "Earlier gateway request"}, + repeated_error, + ] if sidecar_shape != "empty" else [] + stale = _make_session( + session_id=sid, + workspace=str(tmp_path), + messages=history if sidecar_shape == "non_empty" else [], + ) + stale.pending_user_message = "Current gateway request" + stale.pending_started_at = time.time() - 120 + stale.active_stream_id = stream_id + stale.save() + if sidecar_shape == "core": + _write_core_transcript(hermes_home, sid, history) + + _payload, terminal_event, terminal_message = _journal_gateway_terminal_save_failure( + monkeypatch, + stale, + live_messages=history, + ) + models.SESSIONS.pop(sid, None) + + recovered = models.get_session(sid) + current_errors = [ + message for message in recovered.messages + if message.get("_recovered_event_id") == terminal_event["event_id"] + ] + assert [message["content"] for message in current_errors] == [ + terminal_message["content"] + ] + current_error_idx = recovered.messages.index(current_errors[0]) + current_user_idx = max( + idx for idx, message in enumerate(recovered.messages) + if message.get("role") == "user" + and message.get("content") == "Current gateway request" + ) + partial_idx = next( + idx for idx, message in enumerate(recovered.messages) + if message.get("_recovered_stream_id") == stream_id + and message.get("content") == "Current partial output." + ) + assert current_user_idx < partial_idx < current_error_idx + assert current_error_idx == len(recovered.messages) - 1 + + same_text_errors = [ + message for message in recovered.messages + if message.get("_error") is True + and message.get("content") == terminal_message["content"] + ] + expected_count = 1 if sidecar_shape == "empty" else 2 + assert len(same_text_errors) == expected_count + + @pytest.mark.parametrize( + "malformed_field", + [ + "session_id", + "run_id", + "seq", + "event_id", + "payload_session_id", + "embedded_session_id", + "later_done_event_id", + ], + ) + def test_gateway_terminal_error_rejects_foreign_or_malformed_identity( + self, hermes_home, monkeypatch, tmp_path, malformed_field, + ): + """A journal path is not proof that the event envelope or payload owns the turn.""" + sid = f"gateway_terminal_bad_{malformed_field}" + stream_id = f"{sid}_stream" + stale = _make_session( + session_id=sid, + workspace=str(tmp_path), + messages=[ + {"role": "user", "content": "Earlier request"}, + {"role": "assistant", "content": "Earlier answer"}, + ], + ) + stale.pending_user_message = "Current gateway request" + stale.pending_started_at = time.time() - 120 + stale.active_stream_id = stream_id + stale.save() + + _payload, _terminal_event, terminal_message = _journal_gateway_terminal_save_failure( + monkeypatch, + stale, + live_messages=stale.messages, + ) + if malformed_field == "later_done_event_id": + RunJournalWriter(sid, stream_id).append_sse_event( + "done", {"session_id": sid}, + ) + journal_path = ( + models.SESSION_DIR + / "_run_journal" + / sid + / f"{stream_id}.jsonl" + ) + events = [ + json.loads(line) + for line in journal_path.read_text(encoding="utf-8").splitlines() + ] + terminal = events[-1] + if malformed_field == "session_id": + terminal["session_id"] = "foreign_session" + elif malformed_field == "run_id": + terminal["run_id"] = "foreign_run" + elif malformed_field == "seq": + terminal["seq"] = str(terminal["seq"]) + elif malformed_field == "event_id": + terminal["event_id"] = "foreign_run:999" + elif malformed_field == "payload_session_id": + terminal["payload"].pop("session_id", None) + elif malformed_field == "embedded_session_id": + terminal["payload"]["session"].pop("session_id", None) + else: + terminal["event_id"] = "foreign_run:999" + journal_path.write_text( + "".join( + json.dumps(event, ensure_ascii=False, separators=(",", ":")) + "\n" + for event in events + ), + encoding="utf-8", + ) + models.SESSIONS.pop(sid, None) + + recovered = models.get_session(sid) + assert terminal_message["content"] not in [ + message.get("content") for message in recovered.messages + ] + assert any( + message.get("type") == "interrupted" + for message in recovered.messages + ) + + @pytest.mark.parametrize( + ("later_event_name", "expected_error"), + [ + ("apperror", "**Error:** later gateway failure"), + ("done", None), + ("cancel", None), + ("stream_end", "**Error:** gateway exploded"), + ], + ) + def test_gateway_terminal_error_uses_authoritative_latest_terminal_event( + self, hermes_home, monkeypatch, tmp_path, later_event_name, expected_error, + ): + """Later terminal evidence supersedes an older error, except transport-only stream_end.""" + sid = f"gateway_terminal_latest_{later_event_name}" + stream_id = f"{sid}_stream" + stale = _make_session( + session_id=sid, + workspace=str(tmp_path), + messages=[ + {"role": "user", "content": "Earlier request"}, + {"role": "assistant", "content": "Earlier answer"}, + ], + ) + stale.pending_user_message = "Current gateway request" + stale.pending_started_at = time.time() - 120 + stale.active_stream_id = stream_id + stale.save() + payload, first_terminal, first_message = _journal_gateway_terminal_save_failure( + monkeypatch, + stale, + live_messages=stale.messages, + ) + + writer = RunJournalWriter(sid, stream_id) + later_payload = {"session_id": sid} + later_terminal = None + if later_event_name == "apperror": + later_payload = json.loads(json.dumps(payload)) + later_message = next( + message + for message in reversed(later_payload["session"]["messages"]) + if message.get("_error") is True + ) + later_message["content"] = expected_error + later_terminal = writer.append_sse_event("apperror", later_payload) + else: + writer.append_sse_event(later_event_name, later_payload) + models.SESSIONS.pop(sid, None) + + recovered = models.get_session(sid) + recovered_error_contents = [ + message.get("content") + for message in recovered.messages + if message.get("_error") is True + and message.get("type") != "interrupted" + ] + if expected_error is None: + assert first_message["content"] not in recovered_error_contents + else: + assert recovered_error_contents == [expected_error] + expected_event_id = ( + later_terminal["event_id"] + if later_terminal is not None + else first_terminal["event_id"] + ) + assert recovered.messages[-1].get("_recovered_event_id") == expected_event_id + + def test_late_gateway_terminal_journal_replaces_pending_retry_marker( + self, hermes_home, monkeypatch, tmp_path, + ): + """A journal that appears after first repair must still recover its specific error.""" + sid = "gateway_terminal_late_journal" + stream_id = f"{sid}_stream" + stale = _make_session( + session_id=sid, + workspace=str(tmp_path), + messages=[ + {"role": "user", "content": "Earlier request"}, + {"role": "assistant", "content": "Earlier answer"}, + ], + ) + stale.pending_user_message = "Current gateway request" + stale.pending_started_at = time.time() - 120 + stale.active_stream_id = stream_id + stale.save() + models.SESSIONS.pop(sid, None) + + first_recovery = models.get_session(sid) + assert any( + message.get("_pending_journal_recovery") is True + for message in first_recovery.messages + ) + _payload, terminal_event, terminal_message = _journal_gateway_terminal_save_failure( + monkeypatch, + stale, + live_messages=stale.messages, + ) + + recovered = models.get_session(sid) + assert all( + message.get("_pending_journal_recovery") is not True + for message in recovered.messages + ) + assert all( + message.get("type") != "interrupted" + for message in recovered.messages + ) + assert recovered.messages[-1].get("_recovered_event_id") == terminal_event["event_id"] + assert recovered.messages[-1]["content"] == terminal_message["content"] + def test_journal_recovery_restores_reasoning_only_as_display_metadata( self, hermes_home, monkeypatch, ):