From 5af4a481e012543cbb4cf3c215b35fb83e8ff2bd Mon Sep 17 00:00:00 2001 From: Nameless Traveler <2675664993@qq.com> Date: Fri, 24 Jul 2026 00:53:33 +0800 Subject: [PATCH] perf: lazily load branched chat history --- backend/open_webui/models/chats.py | 353 ++++++++-- backend/open_webui/routers/chats.py | 247 ++++++- backend/open_webui/utils/chat_history.py | 238 +++++++ src/lib/apis/chats/index.ts | 152 ++++- src/lib/components/chat/Chat.svelte | 224 +++++-- src/lib/components/chat/Messages.svelte | 606 ++++++++++++------ .../components/chat/Messages/Message.svelte | 4 + .../Messages/MultiResponseMessages.svelte | 160 ++--- .../chat/Messages/ResponseMessage.svelte | 229 +++---- .../chat/Messages/UserMessage.svelte | 6 +- src/lib/utils/chatHistoryWindow.ts | 277 ++++++++ 11 files changed, 1957 insertions(+), 539 deletions(-) create mode 100644 backend/open_webui/utils/chat_history.py create mode 100644 src/lib/utils/chatHistoryWindow.ts diff --git a/backend/open_webui/models/chats.py b/backend/open_webui/models/chats.py index d29b623b44..42012318c9 100644 --- a/backend/open_webui/models/chats.py +++ b/backend/open_webui/models/chats.py @@ -5,6 +5,7 @@ from __future__ import annotations import logging import time import uuid +from copy import deepcopy # local imports from open_webui.internal.db import Base, JSONField, get_async_db_context @@ -12,6 +13,14 @@ from open_webui.models.automations import AutomationRun from open_webui.models.chat_messages import ChatMessage, ChatMessages from open_webui.models.folders import Folders from open_webui.models.tags import Tag, TagModel, Tags +from open_webui.utils.chat_history import ( + MESSAGE_LOADED_KEY, + MESSAGE_WINDOW_KEY, + build_window_chat, + create_message_list, + create_message_window, + prepare_messages_for_storage, +) from open_webui.utils.misc import get_output_text, sanitize_data_for_db, sanitize_text_for_db from pydantic import BaseModel, ConfigDict, field_validator from sqlalchemy import ( @@ -598,7 +607,7 @@ class ChatTable: """Persist updated chat content, sanitizing null bytes.""" try: # load the chat record for in-place mutation async with get_async_db_context(db) as session: - chat_item = await session.get(Chat, id) + chat_item = await session.get(Chat, id, with_for_update=True) if chat_item is None: return None @@ -613,7 +622,8 @@ class ChatTable: await session.commit() return ChatModel.model_validate(chat_item) - except Exception: + except Exception as exc: + log.exception('Failed to update chat %s: %s', id, exc) return async def update_chat_variables_by_id( @@ -639,6 +649,132 @@ class ChatTable: except Exception: return None + async def update_chat_window_by_id( + self, + id: str, + chat: dict, + db: AsyncSession | None = None, + *, + touch: bool = True, + ) -> ChatModel | None: + """Merge a window save without replacing already-persisted message bodies.""" + try: + async with get_async_db_context(db) as session: + chat_item = await session.get(Chat, id, with_for_update=True) + if chat_item is None: + return None + + existing_chat = deepcopy(chat_item.chat or {}) + incoming_chat = self._clean_null_bytes(chat) + merged_chat = {**existing_chat, **incoming_chat} + + existing_history = existing_chat.get('history') or {} + incoming_history = incoming_chat.get('history') or {} + existing_messages = prepare_messages_for_storage(existing_history.get('messages')) + incoming_messages = prepare_messages_for_storage(incoming_history.get('messages')) + + # Existing messages are authoritative. Window saves may append messages, + # while edits to existing messages must use the single-message PATCH API. + merged_messages = { + **existing_messages, + **{ + message_id: message + for message_id, message in incoming_messages.items() + if message_id not in existing_messages + }, + } + + for message in merged_messages.values(): + message['childrenIds'] = [] + for message_id, message in merged_messages.items(): + parent_id = message.get('parentId') + if parent_id in merged_messages: + merged_messages[parent_id]['childrenIds'].append(message_id) + + current_id = incoming_history.get('currentId') + if current_id not in merged_messages: + current_id = existing_history.get('currentId') + if current_id not in merged_messages: + current_id = None + + merged_history = {**existing_history, **incoming_history} + merged_history.pop(MESSAGE_WINDOW_KEY, None) + merged_history['messages'] = merged_messages + merged_history['currentId'] = current_id + merged_chat['history'] = merged_history + + # Rebuild the canonical branch list from authoritative message bodies; + # the top-level list in a window response is only a partial snapshot. + merged_chat['messages'] = create_message_list(merged_messages, current_id) + + chat_item.chat = merged_chat + flag_modified(chat_item, 'chat') + if 'title' in incoming_chat: + chat_item.title = self._clean_null_bytes(incoming_chat['title']) + chat_item.current_message_id = current_id + if touch: + chat_item.updated_at = int(time.time()) + + await session.commit() + await session.refresh(chat_item) + return ChatModel.model_validate(chat_item) + except Exception as exc: + log.exception('Failed to update chat window %s: %s', id, exc) + return None + + async def patch_message_by_chat_id_and_message_id( + self, + id: str, + message_id: str, + patch: dict, + *, + touch: bool = True, + db: AsyncSession | None = None, + ) -> dict | None: + """Patch one existing embedded message under the chat row lock.""" + protected_fields = { + MESSAGE_LOADED_KEY, + 'childrenIds', + 'id', + 'parentId', + 'role', + 'timestamp', + } + clean_patch = { + key: value for key, value in self._clean_null_bytes(patch).items() if key not in protected_fields + } + + try: + async with get_async_db_context(db) as session: + chat_item = await session.get(Chat, id, with_for_update=True) + if chat_item is None: + return None + + chat_data = deepcopy(chat_item.chat or {}) + history = chat_data.get('history') or {} + messages = history.get('messages') or {} + message = messages.get(message_id) + if not isinstance(message, dict): + return None + + for key, value in clean_patch.items(): + if value is None: + message.pop(key, None) + else: + message[key] = value + + chat_item.chat = chat_data + chat_item.current_message_id = history.get('currentId') + flag_modified(chat_item, 'chat') + if clean_patch and touch: + chat_item.updated_at = int(time.time()) + + await session.commit() + return deepcopy(message) + except Exception as exc: + log.exception('Failed to patch message %s for chat %s: %s', message_id, id, exc) + return None + async def update_chat_last_read_at_by_id( self, id: str, user_id: str, db: AsyncSession | None = None ) -> tuple[int, bool] | None: @@ -966,7 +1102,13 @@ class ChatTable: return chat.chat.get('history', {}).get('messages', {}).get(message_id, {}) async def upsert_message_to_chat_by_id_and_message_id( - self, id: str, message_id: str, message: dict, *, touch: bool = True + self, + id: str, + message_id: str, + message: dict, + *, + touch: bool = True, + include_messages: bool = False, ) -> ChatModel | None: if not message.get('content'): output_text = get_output_text(message.get('output')) @@ -977,88 +1119,158 @@ class ChatTable: if isinstance(message.get('content'), str): message['content'] = sanitize_text_for_db(message['content']) + clean_message = self._clean_null_bytes(message) + try: async with get_async_db_context() as session: - chat_item = await session.get(Chat, id) + chat_item = await session.get(Chat, id, with_for_update=True) if chat_item is None: return None self._sanitize_chat_row(chat_item) - chat = chat_item.chat or {} - self._repair_chat_current_id(chat) + chat_data = deepcopy(chat_item.chat or {}) + self._repair_chat_current_id(chat_data) + history = chat_data.setdefault('history', {}) + messages = history.setdefault('messages', {}) + existing_message = messages.get(message_id) + is_new_message = not isinstance(existing_message, dict) - history = chat.get('history', {}) - saved_message = self.upsert_message_to_history(history, message_id, message) - chat['history'] = history - clean_chat = self._clean_null_bytes(chat) - chat_item.chat = clean_chat - chat_item.title = self._clean_null_bytes(clean_chat['title']) if 'title' in clean_chat else 'New Chat' - chat_item.current_message_id = self.get_current_message_id(clean_chat) + old_parent_id = existing_message.get('parentId') if not is_new_message else None + payload = dict(clean_message) + if is_new_message: + parent_id = payload.get('parentId') + if parent_id is None: + parent_id = next( + ( + candidate_id + for candidate_id, candidate in messages.items() + if message_id in (candidate.get('childrenIds') or []) + ), + None, + ) + + parent = messages.get(parent_id) if parent_id else None + output = payload.get('output') or [] + output_role = next( + (item.get('role') for item in output if isinstance(item, dict) and item.get('role')), + None, + ) + role = payload.get('role') or output_role + if not role: + parent_role = parent.get('role') if parent else None + role = ( + 'assistant' + if parent_role == 'user' + else 'user' + if parent_role == 'assistant' + else 'assistant' + ) + + payload = { + **payload, + 'id': message_id, + 'parentId': parent_id, + 'childrenIds': ( + payload.get('childrenIds') if isinstance(payload.get('childrenIds'), list) else [] + ), + 'role': role, + 'timestamp': payload.get('timestamp') or int(time.time()), + } + messages[message_id] = payload + else: + messages[message_id] = {**existing_message, **payload} + + new_parent_id = messages[message_id].get('parentId') + if old_parent_id and old_parent_id != new_parent_id and old_parent_id in messages: + messages[old_parent_id]['childrenIds'] = [ + child_id + for child_id in (messages[old_parent_id].get('childrenIds') or []) + if child_id != message_id + ] + if new_parent_id and (is_new_message or old_parent_id != new_parent_id) and new_parent_id in messages: + children_ids = [ + child_id + for child_id in (messages[new_parent_id].get('childrenIds') or []) + if child_id != message_id + ] + messages[new_parent_id]['childrenIds'] = [*children_ids, message_id] + + if is_new_message: + history['currentId'] = message_id + chat_item.current_message_id = history.get('currentId') + + chat_item.chat = chat_data + chat_item.title = self._clean_null_bytes(chat_data['title']) if 'title' in chat_data else 'New Chat' flag_modified(chat_item, 'chat') - if touch: chat_item.updated_at = int(time.time()) await session.commit() - updated_chat = ChatModel.model_validate(chat_item) - user_id = chat_item.user_id + await session.refresh(chat_item) + result = ChatModel.model_validate(chat_item) - # Dual-write to chat_message table try: await ChatMessages.upsert_message( message_id=message_id, chat_id=id, - user_id=user_id, - data=saved_message, + user_id=result.user_id, + data=messages[message_id], ) - except Exception as e: - log.warning(f'Failed to write to chat_message table: {e}') + except Exception as exc: + log.warning('Failed to mirror message %s for chat %s: %s', message_id, id, exc) - return updated_chat - except Exception: + if not include_messages: + return result + return result + except Exception as exc: + log.exception('Failed to upsert message %s for chat %s: %s', message_id, id, exc) return None - async def delete_message_from_chat_by_id_and_message_id(self, id: str, message_id: str) -> ChatModel | None: + async def delete_message_from_chat_by_id_and_message_id( + self, + id: str, + message_id: str, + *, + include_messages: bool = True, + db: AsyncSession | None = None, + ) -> ChatModel | None: + del include_messages try: - async with get_async_db_context() as session: - chat_item = await session.get(Chat, id) + async with get_async_db_context(db) as session: + chat_item = await session.get(Chat, id, with_for_update=True) if chat_item is None: return None self._sanitize_chat_row(chat_item) - chat = chat_item.chat or {} - self._repair_chat_current_id(chat) - - history = chat.get('history', {}) + chat_data = deepcopy(chat_item.chat or {}) + self._repair_chat_current_id(chat_data) + history = chat_data.get('history') or {} deleted_ids = self.delete_message_from_history(history, message_id) if not deleted_ids: - clean_chat = self._clean_null_bytes(chat) - chat_item.chat = clean_chat - chat_item.title = ( - self._clean_null_bytes(clean_chat['title']) if 'title' in clean_chat else 'New Chat' - ) - chat_item.current_message_id = self.get_current_message_id(clean_chat) + chat_item.chat = self._clean_null_bytes(chat_data) + chat_item.current_message_id = self.get_current_message_id(chat_data) flag_modified(chat_item, 'chat') await session.commit() return ChatModel.model_validate(chat_item) - messages = history.get('messages') or {} - chat['history'] = history - clean_chat = self._clean_null_bytes(chat) + chat_data['history'] = history + clean_chat = self._clean_null_bytes(chat_data) chat_item.chat = clean_chat chat_item.title = self._clean_null_bytes(clean_chat['title']) if 'title' in clean_chat else 'New Chat' chat_item.current_message_id = self.get_current_message_id(clean_chat) - flag_modified(chat_item, 'chat') chat_item.updated_at = int(time.time()) + flag_modified(chat_item, 'chat') await session.commit() - updated_chat = ChatModel.model_validate(chat_item) + await session.refresh(chat_item) + result = ChatModel.model_validate(chat_item) user_id = chat_item.user_id + messages = history.get('messages') or {} await self.backfill_messages_by_chat_id(id, user_id, messages) await ChatMessages.delete_message_ids_by_chat_id(id, deleted_ids) - - return updated_chat - except Exception: + return result + except Exception as exc: + log.exception('Failed to delete message %s from chat %s: %s', message_id, id, exc) return None async def add_message_status_to_chat_by_id_and_message_id( @@ -1492,13 +1704,46 @@ class ChatTable: 'total': total, } + async def get_chat_window( + self, + chat: ChatModel, + limit: int = 32, + current_id: str | None = None, + before_id: str | None = None, + db: AsyncSession | None = None, + ) -> ChatModel: + """Return full branch topology with only one bounded page of message bodies.""" + del db + history = chat.chat.get('history') or {} + messages = history.get('messages') or {} + current_id = current_id or history.get('currentId') + window = create_message_window( + messages, + current_id=current_id, + limit=limit, + before_id=before_id, + ) + window_chat = build_window_chat( + chat.chat, + topology=window['topology'], + loaded_messages=window['messages'], + loaded_ids=window['loaded_ids'], + has_more=window['has_more'], + limit=limit, + current_id=window['current_id'], + ) + return chat.model_copy(update={'chat': window_chat}) + # retrieve conversation async def get_chat_by_id( self, id: str, db: AsyncSession | None = None, + *, + include_messages: bool = True, ) -> ChatModel | None: """Fetch a chat by PK, auto-sanitizing null bytes on read.""" + del include_messages try: async with get_async_db_context(db) as session: chat_item = await session.get(Chat, id) @@ -1508,7 +1753,11 @@ class ChatTable: repaired_history = self._repair_chat_current_id(chat_item.chat or {}) if repaired_history: flag_modified(chat_item, 'chat') - if self._sanitize_chat_row(chat_item) or repaired_history: + current_message_id = self.get_current_message_id(chat_item.chat or {}) + repaired_column = chat_item.current_message_id != current_message_id + if repaired_column: + chat_item.current_message_id = current_message_id + if self._sanitize_chat_row(chat_item) or repaired_history or repaired_column: await session.commit() return ChatModel.model_validate(chat_item) @@ -1537,8 +1786,14 @@ class ChatTable: return None async def get_chat_by_id_and_user_id( - self, id: str, user_id: str, db: AsyncSession | None = None + self, + id: str, + user_id: str, + db: AsyncSession | None = None, + *, + include_messages: bool = True, ) -> ChatModel | None: + del include_messages try: async with get_async_db_context(db) as session: result = await session.execute(select(Chat).filter_by(id=id, user_id=user_id)) @@ -1549,7 +1804,11 @@ class ChatTable: repaired_history = self._repair_chat_current_id(chat.chat or {}) if repaired_history: flag_modified(chat, 'chat') - if self._sanitize_chat_row(chat) or repaired_history: + current_message_id = self.get_current_message_id(chat.chat or {}) + repaired_column = chat.current_message_id != current_message_id + if repaired_column: + chat.current_message_id = current_message_id + if self._sanitize_chat_row(chat) or repaired_history or repaired_column: await session.commit() return ChatModel.model_validate(chat) diff --git a/backend/open_webui/routers/chats.py b/backend/open_webui/routers/chats.py index 0c46f5bbbf..34dc3ac3bd 100644 --- a/backend/open_webui/routers/chats.py +++ b/backend/open_webui/routers/chats.py @@ -1301,35 +1301,115 @@ async def compact_chat_by_id( ############################ -@router.get('/{id}', response_model=ChatResponse | None) -async def get_chat_by_id(id: str, user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session)): - chat = await Chats.get_chat_by_id_and_user_id(id, user.id, db=db) +async def get_accessible_chat_by_id( + id: str, + user, + db: AsyncSession, + *, + include_messages: bool, +): + chat = await Chats.get_chat_by_id_and_user_id( + id, + user.id, + db=db, + include_messages=include_messages, + ) - if not chat and user.role == 'admin': - candidate = await Chats.get_chat_by_id(id, db=db) + if chat: + return chat + + candidate = None + if user.role == 'admin': + candidate = await Chats.get_chat_by_id(id, db=db, include_messages=include_messages) if ENABLE_ADMIN_CHAT_ACCESS or (candidate and is_internal_chat(candidate.meta)): - chat = candidate + return candidate - # Access explicitly granted to this user applies to admins too, so an admin - # does not lose a chat shared with them when ENABLE_ADMIN_CHAT_ACCESS is off. + has_grant = await AccessGrants.has_access( + user_id=user.id, + resource_type='shared_chat', + resource_id=id, + permission='read', + db=db, + ) + if has_grant: + return await Chats.get_chat_by_id(id, db=db, include_messages=include_messages) + + if candidate is None: + candidate = await Chats.get_chat_by_id(id, db=db, include_messages=include_messages) + if candidate and candidate.folder_id: + folder = await Folders.get_folder_by_id(candidate.folder_id, db=db) + if folder and await has_folder_access(user.id, folder, 'read', db): + return candidate + + return None + + +@router.get('/{id}/window', response_model=ChatResponse | None) +async def get_chat_window_by_id( + id: str, + limit: int = 32, + current_id: str | None = None, + user=Depends(get_verified_user), + db: AsyncSession = Depends(get_async_session), +): + limit = max(1, min(limit, 128)) + chat = await get_accessible_chat_by_id(id, user, db, include_messages=False) if not chat: - has_grant = await AccessGrants.has_access( - user_id=user.id, - resource_type='shared_chat', - resource_id=id, - permission='read', + raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=ERROR_MESSAGES.NOT_FOUND) + + try: + chat = await Chats.get_chat_window(chat, limit=limit, current_id=current_id, db=db) + except ValueError as exc: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc + + data = ChatResponse(**chat.model_dump()).model_dump() + data['context_usage'] = await get_chat_context_usage(chat) + return data + + +@router.get('/{id}/history/window') +async def get_chat_history_window_by_id( + id: str, + current_id: str, + limit: int = 32, + before_id: str | None = None, + user=Depends(get_verified_user), + db: AsyncSession = Depends(get_async_session), +): + limit = max(1, min(limit, 128)) + chat = await get_accessible_chat_by_id(id, user, db, include_messages=False) + if not chat: + raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=ERROR_MESSAGES.NOT_FOUND) + + try: + window_chat = await Chats.get_chat_window( + chat, + current_id=current_id, + limit=limit, + before_id=before_id, db=db, ) - if has_grant: - chat = await Chats.get_chat_by_id(id, db=db) + except ValueError as exc: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc - # Check folder-based access (shared folders) - if not chat: - candidate = await Chats.get_chat_by_id(id, db=db) - if candidate and candidate.folder_id: - folder = await Folders.get_folder_by_id(candidate.folder_id, db=db) - if folder and await has_folder_access(user.id, folder, 'read', db): - chat = candidate + history = window_chat.chat.get('history') or {} + metadata = history.get('messageWindow') or {} + loaded_ids = metadata.get('loadedIds') or [] + return { + 'messages': { + message_id: history['messages'][message_id] + for message_id in loaded_ids + if message_id in (history.get('messages') or {}) + }, + 'loadedIds': loaded_ids, + 'hasMore': metadata.get('hasMore', False), + 'currentId': history.get('currentId'), + } + + +@router.get('/{id}', response_model=ChatResponse | None) +async def get_chat_by_id(id: str, user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session)): + chat = await get_accessible_chat_by_id(id, user, db, include_messages=True) if chat: data = ChatResponse.model_validate(chat, from_attributes=True).model_dump() @@ -1344,6 +1424,39 @@ async def get_chat_by_id(id: str, user=Depends(get_verified_user), db: AsyncSess ############################ +@router.post('/{id}/window', response_model=ChatResponse | None) +async def update_chat_window_by_id( + request: Request, + id: str, + form_data: ChatForm, + limit: int = 32, + user=Depends(get_verified_user), + db: AsyncSession = Depends(get_async_session), +): + limit = max(1, min(limit, 128)) + if not await Chats.is_chat_owner(id, user.id, db=db): + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail=ERROR_MESSAGES.ACCESS_PROHIBITED, + ) + + chat = await Chats.update_chat_window_by_id(id, form_data.chat, db=db) + if not chat: + raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=ERROR_MESSAGES.DEFAULT()) + + chat = await Chats.get_chat_window(chat, limit=limit, db=db) + await publish_event( + request, + EVENTS.CHAT_UPDATED, + actor=user, + subject_id=id, + data={'title': chat.title}, + ) + data = ChatResponse(**chat.model_dump()).model_dump() + data['context_usage'] = await get_chat_context_usage(chat) + return data + + @router.post('/{id}', response_model=ChatResponse | None) async def update_chat_by_id( request: Request, @@ -1402,6 +1515,78 @@ class MessageForm(BaseModel): content: str +class MessagePatchForm(BaseModel): + message: dict + + +@router.patch('/{id}/messages/{message_id}', response_model=dict | None) +async def patch_chat_message_by_id( + request: Request, + id: str, + message_id: str, + form_data: MessagePatchForm, + user=Depends(get_verified_user), + db: AsyncSession = Depends(get_async_session), +): + chat = await Chats.get_chat_by_id(id, db=db, include_messages=False) + + if not chat or (chat.user_id != user.id and user.role != 'admin'): + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail=ERROR_MESSAGES.ACCESS_PROHIBITED, + ) + + message = await Chats.patch_message_by_chat_id_and_message_id( + id, + message_id, + form_data.message, + db=db, + ) + if message is None: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=ERROR_MESSAGES.NOT_FOUND, + ) + + event_emitter = await get_event_emitter( + { + 'user_id': chat.user_id, + 'chat_id': id, + 'message_id': message_id, + }, + False, + ) + protected_fields = {'__loaded', 'childrenIds', 'id', 'parentId', 'role', 'timestamp'} + response_patch = { + key: message.get(key) for key in form_data.message if key not in protected_fields and key in message + } + + if event_emitter: + await event_emitter( + { + 'type': 'chat:message', + 'data': { + 'chat_id': id, + 'message_id': message_id, + **response_patch, + }, + } + ) + + content = form_data.message.get('content') + await publish_event( + request, + EVENTS.MESSAGE_UPDATED, + actor=user, + subject_id=message_id, + data={ + 'chat_id': id, + 'content_preview': content[:300] if isinstance(content, str) else '', + }, + ) + return {'id': message_id, **response_patch} + + @router.post('/{id}/messages/{message_id}', response_model=ChatResponse | None) async def update_chat_message_by_id( request: Request, @@ -1469,10 +1654,11 @@ async def delete_chat_message_by_id( request: Request, id: str, message_id: str, + compact: bool = False, user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session), ): - chat = await Chats.get_chat_by_id(id, db=db) + chat = await Chats.get_chat_by_id(id, db=db, include_messages=False) if not chat: raise HTTPException( @@ -1486,7 +1672,18 @@ async def delete_chat_message_by_id( detail=ERROR_MESSAGES.ACCESS_PROHIBITED, ) - chat = await Chats.delete_message_from_chat_by_id_and_message_id(id, message_id) + if await has_active_tasks(request.app.state.redis, id): + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail='Wait for the current response to finish before deleting messages.', + ) + + chat = await Chats.delete_message_from_chat_by_id_and_message_id( + id, + message_id, + include_messages=not compact, + db=db, + ) if not chat: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, @@ -1500,6 +1697,8 @@ async def delete_chat_message_by_id( subject_id=message_id, data={'chat_id': id}, ) + if compact: + chat = await Chats.get_chat_window(chat, limit=32, db=db) return ChatResponse.model_validate(chat, from_attributes=True) diff --git a/backend/open_webui/utils/chat_history.py b/backend/open_webui/utils/chat_history.py new file mode 100644 index 0000000000..3212407ef5 --- /dev/null +++ b/backend/open_webui/utils/chat_history.py @@ -0,0 +1,238 @@ +from __future__ import annotations + +from copy import deepcopy +from typing import Any + +MESSAGE_STORAGE_VERSION = 1 +MESSAGE_STORAGE_KEY = 'messageStorage' +MESSAGE_WINDOW_KEY = 'messageWindow' +MESSAGE_LOADED_KEY = '__loaded' +TOPOLOGY_KEYS = ( + 'id', + 'parentId', + 'childrenIds', + 'role', + 'timestamp', + 'model', + 'modelIdx', + 'modelName', + 'models', + 'done', +) + + +def _message_map(value: Any) -> dict[str, dict]: + if isinstance(value, dict): + return {str(message_id): message for message_id, message in value.items() if isinstance(message, dict)} + + if isinstance(value, list): + return {str(message['id']): message for message in value if isinstance(message, dict) and message.get('id')} + + return {} + + +def has_embedded_messages(chat: dict | None) -> bool: + if not isinstance(chat, dict): + return False + history = chat.get('history') + return (isinstance(history, dict) and 'messages' in history) or 'messages' in chat + + +def uses_normalized_message_storage(chat: dict | None) -> bool: + if not isinstance(chat, dict): + return False + history = chat.get('history') + storage = history.get(MESSAGE_STORAGE_KEY) if isinstance(history, dict) else None + return isinstance(storage, dict) and storage.get('version') == MESSAGE_STORAGE_VERSION + + +def prepare_messages_for_storage(messages: Any) -> dict[str, dict]: + """Return lossless message copies without response-only window markers.""" + prepared: dict[str, dict] = {} + for message_id, message in _message_map(messages).items(): + clean_message = deepcopy(message) + clean_message.pop(MESSAGE_LOADED_KEY, None) + clean_message.setdefault('id', message_id) + prepared[message_id] = clean_message + return prepared + + +def copy_chat_without_messages(chat: dict | None) -> dict: + """Copy chat metadata without adding storage-specific response markers.""" + compact = deepcopy(chat or {}) + history = compact.get('history') + if not isinstance(history, dict): + history = {} + else: + history = deepcopy(history) + + history.pop('messages', None) + history.pop(MESSAGE_WINDOW_KEY, None) + compact.pop('messages', None) + compact['history'] = history + return compact + + +def split_chat_messages(chat: dict | None) -> tuple[dict, dict[str, dict]]: + """Separate embedded messages from compact chat metadata.""" + source = deepcopy(chat or {}) + source_history = source.get('history') if isinstance(source.get('history'), dict) else {} + messages = _message_map(source_history.get('messages')) + top_level_messages = source.get('messages') + if not messages: + messages = _message_map(top_level_messages) + + compact = copy_chat_without_messages(source) + history = compact['history'] + history[MESSAGE_STORAGE_KEY] = {'version': MESSAGE_STORAGE_VERSION} + return compact, prepare_messages_for_storage(messages) + + +def merge_compact_chat(existing: dict | None, incoming: dict | None) -> tuple[dict, dict[str, dict]]: + """Merge chat metadata while returning only messages supplied by the caller.""" + existing_compact, _ = split_chat_messages(existing) + incoming_compact, incoming_messages = split_chat_messages(incoming) + + merged = {**existing_compact, **incoming_compact} + existing_history = existing_compact.get('history') or {} + incoming_history = incoming_compact.get('history') or {} + merged['history'] = {**existing_history, **incoming_history} + merged['history'][MESSAGE_STORAGE_KEY] = {'version': MESSAGE_STORAGE_VERSION} + return merged, incoming_messages + + +def create_message_list(messages: dict[str, dict], current_id: str | None) -> list[dict]: + message_list: list[dict] = [] + visited: set[str] = set() + + while current_id and current_id not in visited: + visited.add(current_id) + message = messages.get(current_id) + if not isinstance(message, dict): + break + message_list.append(message) + current_id = message.get('parentId') + + message_list.reverse() + return message_list + + +def create_message_window( + messages: dict[str, dict], + current_id: str | None, + limit: int, + before_id: str | None = None, +) -> dict: + """Create a branch window from a legacy in-memory message map.""" + if limit < 1: + raise ValueError('limit must be at least 1') + + clean_messages = prepare_messages_for_storage(messages) + if current_id is not None and current_id not in clean_messages: + raise ValueError('current_id does not exist in this chat') + + topology = { + message_id: {key: message[key] for key in TOPOLOGY_KEYS if key in message} + for message_id, message in clean_messages.items() + } + + branch_ids = [message.get('id') for message in create_message_list(clean_messages, current_id) if message.get('id')] + if before_id is not None: + if before_id not in branch_ids: + raise ValueError('before_id is not an ancestor of current_id') + branch_ids = branch_ids[: branch_ids.index(before_id)] + + loaded_ids = branch_ids[-limit:] + return { + 'topology': topology, + 'messages': {message_id: clean_messages[message_id] for message_id in loaded_ids}, + 'loaded_ids': loaded_ids, + 'has_more': len(branch_ids) > len(loaded_ids), + 'current_id': current_id, + } + + +def hydrate_chat(chat: dict | None, messages: dict[str, dict]) -> dict: + """Rebuild the legacy full-chat shape for compatibility endpoints.""" + hydrated, _ = split_chat_messages(chat) + history = hydrated.setdefault('history', {}) + clean_messages = prepare_messages_for_storage(messages) + history['messages'] = clean_messages + hydrated['messages'] = create_message_list(clean_messages, history.get('currentId')) + return hydrated + + +def build_window_chat( + chat: dict | None, + topology: dict[str, dict], + loaded_messages: dict[str, dict], + loaded_ids: list[str], + has_more: bool, + limit: int, + current_id: str | None = None, +) -> dict: + """Build a response containing full topology and only a window of message bodies.""" + window_chat = copy_chat_without_messages(chat) + history = window_chat.setdefault('history', {}) + if current_id is not None: + history['currentId'] = current_id + + clean_loaded = prepare_messages_for_storage(loaded_messages) + messages: dict[str, dict] = {} + + for message_id, topology_message in topology.items(): + topology_copy = deepcopy(topology_message) + if message_id in clean_loaded: + messages[message_id] = { + **clean_loaded[message_id], + **topology_copy, + MESSAGE_LOADED_KEY: True, + } + else: + messages[message_id] = { + **topology_copy, + MESSAGE_LOADED_KEY: False, + } + + for message_id, message in clean_loaded.items(): + if message_id not in messages: + messages[message_id] = {**message, MESSAGE_LOADED_KEY: True} + + unique_loaded_ids = list(dict.fromkeys(loaded_ids)) + history['messages'] = messages + history[MESSAGE_WINDOW_KEY] = { + 'loadedIds': [message_id for message_id in unique_loaded_ids if message_id in messages], + 'hasMore': has_more, + 'limit': limit, + } + return window_chat + + +def inline_message_images_from_files(messages: list[dict]) -> list[dict]: + """Move attached images into message content and remove storage-only files.""" + for message in messages: + raw_files = message.get('files') + files = raw_files if isinstance(raw_files, list) else [] + image_files = [ + file + for file in files + if isinstance(file, dict) + and (file.get('type') == 'image' or (file.get('content_type') or '').startswith('image/')) + ] + if message.get('role') == 'user' and image_files: + text_content = message.get('content', '') + if isinstance(text_content, str): + message['content'] = [ + {'type': 'text', 'text': text_content}, + *[ + { + 'type': 'image_url', + 'image_url': {'url': file['url']}, + } + for file in image_files + if file.get('url') + ], + ] + message.pop('files', None) + + return messages diff --git a/src/lib/apis/chats/index.ts b/src/lib/apis/chats/index.ts index b55df4b72d..af83e6bfb3 100644 --- a/src/lib/apis/chats/index.ts +++ b/src/lib/apis/chats/index.ts @@ -1,5 +1,9 @@ import { WEBUI_API_BASE_URL } from '$lib/constants'; import { getTimeRange } from '$lib/utils'; +import { + DEFAULT_CHAT_MESSAGE_WINDOW, + type ChatHistoryWindowResponse +} from '$lib/utils/chatHistoryWindow'; export const getChatConfig = async (token: string) => { let error = null; @@ -777,6 +781,87 @@ export const getChatById = async (token: string, id: string) => { return res; }; +export const getChatByIdWindow = async ( + token: string, + id: string, + limit: number = DEFAULT_CHAT_MESSAGE_WINDOW, + currentId?: string +) => { + let error = null; + const searchParams = new URLSearchParams(); + searchParams.set('limit', `${limit}`); + if (currentId) { + searchParams.set('current_id', currentId); + } + + const res = await fetch(`${WEBUI_API_BASE_URL}/chats/${id}/window?${searchParams.toString()}`, { + method: 'GET', + headers: { + Accept: 'application/json', + 'Content-Type': 'application/json', + ...(token && { authorization: `Bearer ${token}` }) + } + }) + .then(async (res) => { + if (!res.ok) throw await res.json(); + return res.json(); + }) + .catch((err) => { + error = err?.detail ?? err; + console.error(err); + return null; + }); + + if (error) { + throw error; + } + + return res; +}; + +export const getChatHistoryWindow = async ( + token: string, + id: string, + currentId: string, + limit: number = DEFAULT_CHAT_MESSAGE_WINDOW, + beforeId?: string +): Promise => { + let error = null; + const searchParams = new URLSearchParams(); + searchParams.set('current_id', currentId); + searchParams.set('limit', `${limit}`); + if (beforeId) { + searchParams.set('before_id', beforeId); + } + + const res = await fetch( + `${WEBUI_API_BASE_URL}/chats/${id}/history/window?${searchParams.toString()}`, + { + method: 'GET', + headers: { + Accept: 'application/json', + 'Content-Type': 'application/json', + ...(token && { authorization: `Bearer ${token}` }) + } + } + ) + .then(async (res) => { + if (!res.ok) throw await res.json(); + return res.json(); + }) + .catch((err) => { + error = err?.detail ?? err; + console.error(err); + return null; + }); + + if (error) { + throw error; + } + + return res; +}; + export const getChatByShareId = async (token: string, share_id: string) => { let error = null; @@ -1300,6 +1385,37 @@ export const updateChatById = async ( return res; }; +export const updateChatByIdWindow = async (token: string, id: string, chat: object) => { + let error = null; + + const res = await fetch(`${WEBUI_API_BASE_URL}/chats/${id}/window`, { + method: 'POST', + headers: { + Accept: 'application/json', + 'Content-Type': 'application/json', + ...(token && { authorization: `Bearer ${token}` }) + }, + body: JSON.stringify({ + chat + }) + }) + .then(async (res) => { + if (!res.ok) throw await res.json(); + return res.json(); + }) + .catch((err) => { + error = err?.detail ?? err; + console.error(err); + return null; + }); + + if (error) { + throw error; + } + + return res; +}; + export const compactChatById = async (token: string, id: string, model?: string | null) => { let error = null; @@ -1333,10 +1449,44 @@ export const compactChatById = async (token: string, id: string, model?: string return res; }; -export const deleteChatMessageById = async (token: string, id: string, messageId: string) => { +export const updateChatMessageById = async ( + token: string, + id: string, + messageId: string, + message: object +) => { let error = null; const res = await fetch(`${WEBUI_API_BASE_URL}/chats/${id}/messages/${messageId}`, { + method: 'PATCH', + headers: { + Accept: 'application/json', + 'Content-Type': 'application/json', + ...(token && { authorization: `Bearer ${token}` }) + }, + body: JSON.stringify({ message }) + }) + .then(async (res) => { + if (!res.ok) throw await res.json(); + return res.json(); + }) + .catch((err) => { + error = err?.detail ?? err; + console.error(err); + return null; + }); + + if (error) { + throw error; + } + + return res; +}; + +export const deleteChatMessageById = async (token: string, id: string, messageId: string) => { + let error = null; + + const res = await fetch(`${WEBUI_API_BASE_URL}/chats/${id}/messages/${messageId}?compact=true`, { method: 'DELETE', headers: { Accept: 'application/json', diff --git a/src/lib/components/chat/Chat.svelte b/src/lib/components/chat/Chat.svelte index a57fbc8cf0..9d11ca0780 100644 --- a/src/lib/components/chat/Chat.svelte +++ b/src/lib/components/chat/Chat.svelte @@ -76,11 +76,20 @@ deleteChatById, forkChatById, getAllTags, - getChatById, + getChatByIdWindow, + getChatHistoryWindow, getTagsById, - updateChatById, + updateChatByIdWindow, updateChatFolderIdById } from '$lib/apis/chats'; + import { + DEFAULT_CHAT_MESSAGE_WINDOW, + isChatMessageLoaded, + mergeChatMessageEvent, + mergeChatHistoryWindow, + resolveLoadedCurrentMessageId, + serializeChatForWindowSave + } from '$lib/utils/chatHistoryWindow'; import { generateOpenAIChatCompletion } from '$lib/apis/openai'; import { processWeb, processWebSearch, processYoutubeVideo } from '$lib/apis/retrieval'; import { getAndUpdateUserLocation, getUserSettings } from '$lib/apis/users'; @@ -121,6 +130,17 @@ import EmbeddedChatHistoryDropdown from './EmbeddedChatHistoryDropdown.svelte'; import InputVariablesModal from './MessageInput/InputVariablesModal.svelte'; + type ChatHistoryState = { + currentId: string | null; + messages: Record; + messageWindow?: { + loadedIds?: string[]; + hasMore?: boolean; + [key: string]: unknown; + }; + [key: string]: unknown; + }; + export let chatIdProp = ''; export let embedded = false; export let embeddedTitle = ''; @@ -138,6 +158,84 @@ null; let loading = true; + let chatDestroyed = false; + let loadChatGeneration = 0; + const historyWindowRequests = new Map>(); + let showMessageNavigationGeneration = 0; + onDestroy(() => { + chatDestroyed = true; + loadChatGeneration += 1; + showMessageNavigationGeneration += 1; + historyWindowRequests.clear(); + }); + + const getWindowSavePayload = (payload: any) => { + const serialized = serializeChatForWindowSave(payload); + + if (serialized.history && Array.isArray(serialized.messages)) { + const loadedMessageIds = new Set(Object.keys(serialized.history.messages ?? {})); + serialized.messages = serialized.messages.filter( + (message: any) => !message?.id || loadedMessageIds.has(message.id) + ); + } + + return serialized; + }; + + const updateChatWindow = async (id: string, payload: any) => + await updateChatByIdWindow(localStorage.token, id, getWindowSavePayload(payload)); + + const hydrateBranchWindow = async (messageId: string, navigationGeneration: number) => { + if ( + !messageId || + !history?.messageWindow || + $temporaryChatEnabled || + !$chatId || + $chatId.startsWith('local:') + ) { + return true; + } + + const requestChatId = $chatId; + const requestKey = `${navigationGeneration}:${requestChatId}:${messageId}`; + const pendingRequest = historyWindowRequests.get(requestKey); + if (pendingRequest) return await pendingRequest; + + const request = (async () => { + try { + const window = await getChatHistoryWindow( + localStorage.token, + requestChatId, + messageId, + DEFAULT_CHAT_MESSAGE_WINDOW + ); + if ( + !window || + chatDestroyed || + $chatId !== requestChatId || + navigationGeneration !== showMessageNavigationGeneration + ) { + return false; + } + + const activeCurrentId = history.currentId; + const mergedHistory = mergeChatHistoryWindow(history, window) as ChatHistoryState; + mergedHistory.currentId = activeCurrentId; + history = mergedHistory; + return true; + } catch (error) { + console.error('Failed to load chat history branch', error); + toast.error($i18n.t('Failed to load chat')); + return false; + } finally { + historyWindowRequests.delete(requestKey); + } + })(); + + historyWindowRequests.set(requestKey, request); + return await request; + }; + $: chatContainerId = embedded ? 'note-chat-container' : 'chat-container'; $: messageInputDropzoneId = embedded ? 'note-chat-input-dropzone' : 'chat-pane'; @@ -375,7 +473,7 @@ let chatTasks = []; - let history = { + let history: ChatHistoryState = { messages: {}, currentId: null }; @@ -853,6 +951,7 @@ }; const showMessage = async (message, scroll = true, save = true) => { + const navigationGeneration = ++showMessageNavigationGeneration; const _chatId = JSON.parse(JSON.stringify($chatId)); let _messageId = JSON.parse(JSON.stringify(message.id)); @@ -870,6 +969,9 @@ messageChildrenIds = history.messages[_messageId].childrenIds; } + if (_messageId && !(await hydrateBranchWindow(_messageId, navigationGeneration))) return; + if (navigationGeneration !== showMessageNavigationGeneration) return; + history.currentId = _messageId; await tick(); @@ -885,8 +987,8 @@ await tick(); await tick(); - if (save) { - saveChatHandler(_chatId, history); + if (save && !readOnly) { + await saveChatHandler(_chatId, history); } }; @@ -997,15 +1099,29 @@ message.done = true; } } else if (type === 'chat:message:delta' || type === 'message') { - message.content += data.content; - } else if (type === 'chat:message' || type === 'replace') { - message.content = data.content; + message.content = `${message.content ?? ''}${data?.content ?? ''}`; + } else if (type === 'chat:message') { + message = mergeChatMessageEvent(message, data); + } else if (type === 'replace') { + message = mergeChatMessageEvent(message, data, true); } else if (type === 'chat:message:files' || type === 'files') { - message.files = data.files; + const files = data?.files ?? []; + message.files = data?.replace + ? files + : [...(message.files ?? []), ...files].filter( + (item, index, array) => + array.findIndex((candidate) => equal(candidate, item)) === index + ); } else if (type === 'chat:message:tasks') { chatTasks = data.tasks; } else if (type === 'chat:message:embeds' || type === 'embeds') { - message.embeds = data.embeds; + const embeds = data?.embeds ?? []; + message.embeds = data?.replace + ? embeds + : [...(message.embeds ?? []), ...embeds].filter( + (item, index, array) => + array.findIndex((candidate) => equal(candidate, item)) === index + ); // Auto-scroll to the embed once it's rendered in the DOM await tick(); @@ -1050,7 +1166,7 @@ } await refreshChatList(localStorage.token); } else if (type === 'chat:tags') { - chat = await getChatById(localStorage.token, $chatId); + chat = await getChatByIdWindow(localStorage.token, $chatId, DEFAULT_CHAT_MESSAGE_WINDOW); allTags.set(await getAllTags(localStorage.token)); } else if (type === 'source' || type === 'citation') { if (data?.type === 'code_execution') { @@ -1940,9 +2056,11 @@ }; const loadChat = async () => { + const requestGeneration = ++loadChatGeneration; + const requestChatId = chatIdProp || $chatId; noteChatDebug('loadChat start'); // chatIdProp is empty for chats started from the home page (URL set via replaceState) - chatId.set(chatIdProp || $chatId); + chatId.set(requestChatId); noteChatDebug('loadChat set active chat id'); if ($temporaryChatEnabled) { @@ -1950,18 +2068,26 @@ temporaryChatEnabled.set(false); } - chat = await getChatById(localStorage.token, $chatId).catch(async (error) => { - console.error('[note-chat] getChatById failed', { + const loadedChat = await getChatByIdWindow( + localStorage.token, + requestChatId, + DEFAULT_CHAT_MESSAGE_WINDOW + ).catch(async (error) => { + console.error('[note-chat] getChatByIdWindow failed', { chatIdProp, activeChatId: $chatId, error }); - if (!embedded) { + if (!embedded && requestGeneration === loadChatGeneration) { await goto('/'); } return null; }); - noteChatDebug('getChatById completed', { + if (chatDestroyed || requestGeneration !== loadChatGeneration || $chatId !== requestChatId) { + return false; + } + chat = loadedChat; + noteChatDebug('getChatByIdWindow completed', { found: !!chat, chatId: chat?.id, hasChatPayload: !!chat?.chat, @@ -1969,7 +2095,7 @@ }); if (chat) { - tags = await getTagsById(localStorage.token, $chatId).catch(async (error) => { + tags = await getTagsById(localStorage.token, requestChatId).catch(async (error) => { console.warn('[note-chat] getTagsById failed; continuing without tags', { chatIdProp, activeChatId: $chatId, @@ -1977,6 +2103,9 @@ }); return []; }); + if (chatDestroyed || requestGeneration !== loadChatGeneration || $chatId !== requestChatId) { + return false; + } noteChatDebug('getTagsById completed', { tagCount: tags?.length ?? 0 }); const chatContent = chat.chat; @@ -2011,9 +2140,7 @@ (chatContent?.history ?? undefined) !== undefined ? chatContent.history : convertMessagesToHistory(chatContent.messages); - if (chat?.current_message_id && history?.messages?.[chat.current_message_id]) { - history.currentId = chat.current_message_id; - } + history.currentId = resolveLoadedCurrentMessageId(history, chat?.current_message_id); // Sanitize history: repair orphaned references and structurally-malformed // nodes from failed regenerations (#24424, #24157, #20474) @@ -2037,6 +2164,7 @@ for (const message of Object.values(history.messages)) { if ( message && + isChatMessageLoaded(history, message.id) && message.role === 'assistant' && message.id !== history.currentId && message.done !== false @@ -2051,7 +2179,7 @@ // work (follow-ups, title gen) that shouldn't block the input. const activeTaskIds = taskIds; const currentMessage = history.currentId ? history.messages[history.currentId] : null; - const pendingTaskIds = await getTaskIdsByChatId(localStorage.token, $chatId) + const pendingTaskIds = await getTaskIdsByChatId(localStorage.token, requestChatId) .then((res) => res?.task_ids ?? []) .catch((error) => { console.warn('[note-chat] getTaskIdsByChatId failed; continuing without tasks', { @@ -2061,6 +2189,13 @@ }); return []; }); + if ( + chatDestroyed || + requestGeneration !== loadChatGeneration || + $chatId !== requestChatId + ) { + return false; + } noteChatDebug('task reconciliation completed', { pendingTaskCount: pendingTaskIds.length, hasCurrentMessage: !!currentMessage @@ -2093,7 +2228,7 @@ return null; } } - console.warn('[note-chat] no chat returned from getChatById', { + console.warn('[note-chat] no chat returned from getChatByIdWindow', { chatIdProp, activeChatId: $chatId }); @@ -2223,7 +2358,7 @@ if ($chatId == _chatId) { if (!$temporaryChatEnabled) { - chat = await updateChatById(localStorage.token, _chatId, { + chat = await updateChatWindow(_chatId, { models: selectedModels, messages: messages, history: history, @@ -2772,14 +2907,14 @@ const sendMessage = async ( _history, - parentId: string, + parentId: string | null, { - messages = null, + messagesFromId = null, modelId = null, modelIdx = null, regenerationPrompt = null }: { - messages?: any[] | null; + messagesFromId?: string | null; modelId?: string | null; modelIdx?: number | null; regenerationPrompt?: string | null; @@ -2790,7 +2925,9 @@ } let _chatId = JSON.parse(JSON.stringify($chatId)); - _history = structuredClone(_history); + _history = structuredClone(history); + const messages = + $temporaryChatEnabled && messagesFromId ? createMessagesList(_history, messagesFromId) : null; const responseMessageIds: Record = {}; // If modelId is provided, use it, else use selected model @@ -2925,7 +3062,8 @@ // regenerations in a duplicate-model chat, which would otherwise lose their // column identity and collapse on reload. messageIdsList: messageIdsList.length > 0 ? messageIdsList : undefined, - regenerationPrompt + regenerationPrompt, + regenerationSourceMessageId: regenerationPrompt ? messagesFromId : null } ); } finally { @@ -2980,25 +3118,28 @@ { messageIdsList, regenerationPrompt, + regenerationSourceMessageId, continueResponse = false }: { messageIdsList?: Array<{ model_id: string; message_id: string }>; regenerationPrompt?: string | null; + regenerationSourceMessageId?: string | null; continueResponse?: boolean; } = {} ) => { const responseMessage = _history.messages[responseMessageId]; const userMessage = _history.messages[responseMessage.parentId]; - const chatMessageFiles = _messages - .filter((message) => message.files) - .flatMap((message) => message.files); + if ($temporaryChatEnabled) { + const chatMessageFiles = _messages + .filter((message) => message.files) + .flatMap((message) => message.files); - // Filter chatFiles to only include files that are in the chatMessageFiles - chatFiles = chatFiles.filter((item) => { - const fileExists = chatMessageFiles.some((messageFile) => messageFile.id === item.id); - return fileExists; - }); + // Temporary chats have no server-side branch to recover attachment state from. + chatFiles = chatFiles.filter((item) => + chatMessageFiles.some((messageFile) => messageFile.id === item.id) + ); + } let files = structuredClone(chatFiles); files.push( @@ -3167,6 +3308,9 @@ parent_id: userMessage?.parentId ?? null, user_message: userMessage, ...(regenerationPrompt ? { regeneration_prompt: regenerationPrompt } : {}), + ...(regenerationSourceMessageId + ? { regeneration_source_message_id: regenerationSourceMessageId } + : {}), ...(continueResponse ? { assistant_message_id: responseMessageId } : {}), background_tasks: { @@ -3244,7 +3388,7 @@ // chat completion request. Files are now persisted // by the backend at chat creation time. if (Object.keys(params).length > 0) { - await updateChatById(localStorage.token, res.chat_id, { + await updateChatWindow(res.chat_id, { params: params }); } @@ -3396,7 +3540,7 @@ await sendMessage(history, userMessage.id, { ...(suggestionPrompt ? { - messages: createMessagesList(history, message.id), + messagesFromId: message.id, regenerationPrompt: suggestionPrompt } : {}), @@ -3541,7 +3685,7 @@ const saveChatHandler = async (_chatId, history) => { if ($chatId == _chatId) { if (!$temporaryChatEnabled) { - chat = await updateChatById(localStorage.token, _chatId, { + chat = await updateChatWindow(_chatId, { models: selectedModels, history: history, messages: createMessagesList(history, history.currentId), @@ -3557,7 +3701,7 @@ const loaded = chat?.chat ?? {}; if (equal(params, loaded.params ?? {}) && equal(chatFiles, loaded.files ?? [])) return; - const res = await updateChatById(localStorage.token, $chatId, { + const res = await updateChatWindow($chatId, { params, files: chatFiles }).catch((err) => { diff --git a/src/lib/components/chat/Messages.svelte b/src/lib/components/chat/Messages.svelte index 2f4ad5fec9..00caa4074b 100644 --- a/src/lib/components/chat/Messages.svelte +++ b/src/lib/components/chat/Messages.svelte @@ -6,8 +6,20 @@ const dispatch = createEventDispatcher(); import { toast } from 'svelte-sonner'; - import { deleteChatMessageById, updateChatById } from '$lib/apis/chats'; + import { + deleteChatMessageById, + getChatHistoryWindow, + updateChatByIdWindow, + updateChatMessageById + } from '$lib/apis/chats'; import { copyToClipboard, extractCurlyBraceWords } from '$lib/utils'; + import { + createMessagePatch, + DEFAULT_CHAT_MESSAGE_WINDOW, + isChatMessageLoaded, + mergeChatHistoryWindow, + serializeChatForWindowSave + } from '$lib/utils/chatHistoryWindow'; import Message from './Messages/Message.svelte'; import Loader from '../common/Loader.svelte'; @@ -15,7 +27,18 @@ import ChatPlaceholder from './ChatPlaceholder.svelte'; - const i18n = getContext('i18n'); + const i18n: any = getContext('i18n'); + + type ChatHistoryState = { + currentId: string | null; + messages: Record; + messageWindow?: { + loadedIds?: string[]; + hasMore?: boolean; + [key: string]: unknown; + }; + [key: string]: unknown; + }; export let className = 'h-full flex pt-18'; @@ -23,11 +46,11 @@ export let user = $_user; export let prompt; - export let history = {}; + export let history: ChatHistoryState = { messages: {}, currentId: null }; export let selectedModels; export let atSelectedModel; - let messages = []; + let messages: any[] = []; export let setInputText: Function = () => {}; @@ -49,45 +72,199 @@ export let topPadding = false; export let bottomPadding = false; - export let autoScroll; + export let autoScroll: boolean; export let messagesContainerId = 'messages-container'; - export let onSelect = (e) => {}; + export let onSelect = (e: any) => {}; export let onInsertToNote: ((content: string) => void) | null = null; export let messagesCount: number | null = 8; let messagesLoading = false; + let destroyed = false; + const historyWindowRequests = new Map>(); + const branchHasMoreByCurrentId = new Map(); + let pendingRebuild: number | null = null; + let lastCurrentId: string | null = null; + let branchNavigationGeneration = 0; + let historyGeneration = 0; + let observedHistory = history; + let observedChatId = chatId; const getMessagesContainer = () => document.getElementById(messagesContainerId); - onDestroy(() => { - cancelAnimationFrame(pendingRebuild); - }); - - const loadMoreMessages = async () => { - // scroll slightly down to disable continuous loading - const element = getMessagesContainer(); - if (element) { - element.scrollTop = element.scrollTop + 100; - } - - messagesLoading = true; - messagesCount += 8; - - buildMessages(); - - await tick(); - + const resetLazyPaginationState = () => { + messagesCount = 8; messagesLoading = false; + branchNavigationGeneration += 1; + historyGeneration += 1; + historyWindowRequests.clear(); + branchHasMoreByCurrentId.clear(); + lastCurrentId = null; + if (pendingRebuild !== null) cancelAnimationFrame(pendingRebuild); + pendingRebuild = null; }; - let pendingRebuild = null; - let lastCurrentId = null; + const setMergedHistory = (nextHistory: ChatHistoryState) => { + observedHistory = nextHistory; + history = nextHistory; + }; + + const replaceHistory = (nextHistory: ChatHistoryState) => { + resetLazyPaginationState(); + observedHistory = nextHistory; + observedChatId = chatId; + history = nextHistory; + }; + + const observeHistoryReplacement = (nextHistory: ChatHistoryState, nextChatId: string) => { + if (nextHistory === observedHistory && nextChatId === observedChatId) return; + + resetLazyPaginationState(); + observedHistory = nextHistory; + observedChatId = nextChatId; + }; + + onDestroy(() => { + destroyed = true; + branchNavigationGeneration += 1; + historyGeneration += 1; + historyWindowRequests.clear(); + branchHasMoreByCurrentId.clear(); + if (pendingRebuild !== null) cancelAnimationFrame(pendingRebuild); + }); + + $: observeHistoryReplacement(history, chatId); + + const getBranchLoadState = (currentId = history?.currentId) => { + let loadedCount = 0; + let earliestLoadedId: string | null = null; + let blockedId: string | null = null; + let reachedRoot = false; + let messageId = currentId; + const visited = new Set(); + + while (messageId) { + if (visited.has(messageId)) { + reachedRoot = true; + break; + } + visited.add(messageId); + + const message = history?.messages?.[messageId]; + if (!message || !isChatMessageLoaded(history, messageId)) { + blockedId = messageId; + break; + } + + loadedCount += 1; + earliestLoadedId = messageId; + if (message.parentId === null || message.parentId === undefined) { + reachedRoot = true; + break; + } + messageId = message.parentId; + } + + return { loadedCount, earliestLoadedId, blockedId, reachedRoot }; + }; + + const hydrateHistoryWindow = async (currentId: string, beforeId?: string) => { + if ( + !currentId || + !history?.messageWindow || + $temporaryChatEnabled || + !chatId || + chatId.startsWith('local:') + ) { + return true; + } + + const requestChatId = chatId; + const requestGeneration = historyGeneration; + const requestKey = `${requestGeneration}:${requestChatId}:${currentId}:${beforeId ?? ''}`; + const pendingRequest = historyWindowRequests.get(requestKey); + if (pendingRequest) return await pendingRequest; + + const request = (async () => { + try { + const historyWindow = await getChatHistoryWindow( + localStorage.token, + requestChatId, + currentId, + DEFAULT_CHAT_MESSAGE_WINDOW, + beforeId + ); + if ( + !historyWindow || + destroyed || + chatId !== requestChatId || + historyGeneration !== requestGeneration + ) { + return false; + } + + const activeCurrentId = history.currentId; + branchHasMoreByCurrentId.set(currentId, historyWindow.hasMore); + const mergedHistory = mergeChatHistoryWindow(history, historyWindow) as ChatHistoryState; + mergedHistory.currentId = activeCurrentId; + if (currentId !== activeCurrentId && mergedHistory.messageWindow) { + mergedHistory.messageWindow.hasMore = history.messageWindow?.hasMore; + } + setMergedHistory(mergedHistory); + return true; + } catch (error) { + console.error('Failed to load chat history window', error); + toast.error($i18n.t('Failed to load chat')); + return false; + } finally { + historyWindowRequests.delete(requestKey); + } + })(); + + historyWindowRequests.set(requestKey, request); + return await request; + }; + + const ensureLoadedAncestors = async (targetCount: number) => { + while (true) { + const state = getBranchLoadState(); + if (state.loadedCount >= targetCount || state.reachedRoot) return true; + if (!state.earliestLoadedId || history?.messageWindow?.hasMore === false) return false; + + const currentId = history.currentId; + if (!currentId) return false; + const loaded = await hydrateHistoryWindow(currentId, state.earliestLoadedId); + if (!loaded) return false; + + const nextState = getBranchLoadState(); + if (nextState.loadedCount <= state.loadedCount) return false; + } + }; + + const loadMoreMessages = async () => { + if (messagesLoading || messagesCount === null) return; + + // scroll slightly down to disable continuous loading + const element = getMessagesContainer(); + if (element) element.scrollTop += 100; + + messagesLoading = true; + try { + const nextCount = messagesCount + 8; + if (!(await ensureLoadedAncestors(nextCount))) return; + + messagesCount = nextCount; + buildMessages(); + await tick(); + } finally { + messagesLoading = false; + } + }; const buildMessages = () => { let _messages = []; - let message = history.messages[history.currentId]; + let message = history.currentId ? history.messages[history.currentId] : null; const visitedMessageIds = new Set(); while (message && (messagesCount !== null ? _messages.length < messagesCount : true)) { @@ -97,6 +274,7 @@ } visitedMessageIds.add(message.id); + if (!isChatMessageLoaded(history, message.id)) break; _messages.push(message); message = message.parentId !== null ? history.messages[message.parentId] : null; } @@ -106,7 +284,7 @@ // Throttle message list rebuilds to once per animation frame during streaming. // Structural changes (currentId change) always rebuild immediately. - const handleHistoryChange = (currentId, _messages) => { + const handleHistoryChange = (currentId: string | null, _messages: Record) => { if (!currentId) { messages = []; return; @@ -117,7 +295,7 @@ if (currentIdChanged) { // Structural change: new chat, navigation, new message — rebuild immediately - cancelAnimationFrame(pendingRebuild); + if (pendingRebuild !== null) cancelAnimationFrame(pendingRebuild); pendingRebuild = null; buildMessages(); } else if (_messages) { @@ -155,14 +333,35 @@ }; export const scrollToTop = async () => { - messagesCount = null; - buildMessages(); - await tick(); - if (messages.length > 0) { - const firstMessageEl = document.getElementById(`message-${messages[0].id}`); - if (firstMessageEl) { - firstMessageEl.scrollIntoView({ behavior: 'smooth', block: 'start' }); + if (messagesLoading) return; + + messagesLoading = true; + try { + while (true) { + const state = getBranchLoadState(); + if (state.reachedRoot) break; + if (!state.earliestLoadedId || history?.messageWindow?.hasMore === false) return; + + const currentId = history.currentId; + if (!currentId) return; + const loaded = await hydrateHistoryWindow(currentId, state.earliestLoadedId); + if (!loaded) return; + + const nextState = getBranchLoadState(); + if (nextState.loadedCount <= state.loadedCount) return; } + + messagesCount = null; + buildMessages(); + await tick(); + if (messages.length > 0) { + const firstMessageEl = document.getElementById(`message-${messages[0].id}`); + if (firstMessageEl) { + firstMessageEl.scrollIntoView({ behavior: 'smooth', block: 'start' }); + } + } + } finally { + messagesLoading = false; } }; @@ -170,181 +369,113 @@ if (!$temporaryChatEnabled) { history = history; await tick(); - const res = await updateChatById(localStorage.token, chatId, { + const payload = serializeChatForWindowSave({ history: history, messages: messages }); - - // Keep local plain-content edits aligned with the saved chat response. - if (res?.chat?.history?.messages) { - for (const [id, msg] of Object.entries(res.chat.history.messages)) { - if (history.messages[id] && (msg as any).content) { - history.messages[id].content = (msg as any).content; - } - } - history = history; - } + await updateChatByIdWindow(localStorage.token, chatId, payload); await refreshChatList(localStorage.token); } }; - const gotoMessage = async (message, idx) => { - // Determine the correct sibling list (either parent's children or root messages) - let siblings; - if (message.parentId !== null) { - siblings = history.messages[message.parentId].childrenIds; - } else { - siblings = Object.values(history.messages) - .filter((msg) => msg.parentId === null) - .map((msg) => msg.id); + const getSiblingIds = (message: any): string[] => + message.parentId !== null + ? (history.messages[message.parentId]?.childrenIds ?? []) + : Object.values(history.messages) + .filter((candidate) => candidate.parentId === null) + .map((candidate) => candidate.id); + + const getDeepestLeafId = (messageId: string) => { + const visited = new Set(); + let leafId = messageId; + + while (leafId && !visited.has(leafId)) { + visited.add(leafId); + const childrenIds = history.messages[leafId]?.childrenIds ?? []; + if (childrenIds.length === 0) break; + leafId = childrenIds.at(-1); } - // Clamp index to a valid range - idx = Math.max(0, Math.min(idx, siblings.length - 1)); - - let messageId = siblings[idx]; - - // If we're navigating to a different message - if (message.id !== messageId) { - // Drill down to the deepest child of that branch - let messageChildrenIds = history.messages[messageId].childrenIds; - while (messageChildrenIds.length !== 0) { - messageId = messageChildrenIds.at(-1); - messageChildrenIds = history.messages[messageId].childrenIds; - } - - history.currentId = messageId; - } - - await tick(); - - // Optional auto-scroll - if ($settings?.scrollOnBranchChange ?? true) { - const element = getMessagesContainer(); - autoScroll = element - ? element.scrollHeight - element.scrollTop <= element.clientHeight + 50 - : false; - - setTimeout(() => { - scrollToBottom(); - }, 100); - } + return leafId; }; - const showPreviousMessage = async (message) => { - if (message.parentId !== null) { - let messageId = - history.messages[message.parentId].childrenIds[ - Math.max(history.messages[message.parentId].childrenIds.indexOf(message.id) - 1, 0) - ]; - - if (message.id !== messageId) { - let messageChildrenIds = history.messages[messageId].childrenIds; - - while (messageChildrenIds.length !== 0) { - messageId = messageChildrenIds.at(-1); - messageChildrenIds = history.messages[messageId].childrenIds; - } - - history.currentId = messageId; - } - } else { - let childrenIds = Object.values(history.messages) - .filter((message) => message.parentId === null) - .map((message) => message.id); - let messageId = childrenIds[Math.max(childrenIds.indexOf(message.id) - 1, 0)]; - - if (message.id !== messageId) { - let messageChildrenIds = history.messages[messageId].childrenIds; - - while (messageChildrenIds.length !== 0) { - messageId = messageChildrenIds.at(-1); - messageChildrenIds = history.messages[messageId].childrenIds; - } - - history.currentId = messageId; - } - } - - await tick(); - - if ($settings?.scrollOnBranchChange ?? true) { - const element = getMessagesContainer(); - autoScroll = element - ? element.scrollHeight - element.scrollTop <= element.clientHeight + 50 - : false; - - setTimeout(() => { - scrollToBottom(); - }, 100); - } + const ensureMessageLoaded = async (messageId: string) => { + if (!messageId || isChatMessageLoaded(history, messageId)) return true; + return await hydrateHistoryWindow(messageId); }; - const showNextMessage = async (message) => { - if (message.parentId !== null) { - let messageId = - history.messages[message.parentId].childrenIds[ - Math.min( - history.messages[message.parentId].childrenIds.indexOf(message.id) + 1, - history.messages[message.parentId].childrenIds.length - 1 - ) - ]; - - if (message.id !== messageId) { - let messageChildrenIds = history.messages[messageId].childrenIds; - - while (messageChildrenIds.length !== 0) { - messageId = messageChildrenIds.at(-1); - messageChildrenIds = history.messages[messageId].childrenIds; - } - - history.currentId = messageId; - } - } else { - let childrenIds = Object.values(history.messages) - .filter((message) => message.parentId === null) - .map((message) => message.id); - let messageId = - childrenIds[Math.min(childrenIds.indexOf(message.id) + 1, childrenIds.length - 1)]; - - if (message.id !== messageId) { - let messageChildrenIds = history.messages[messageId].childrenIds; - - while (messageChildrenIds.length !== 0) { - messageId = messageChildrenIds.at(-1); - messageChildrenIds = history.messages[messageId].childrenIds; - } - - history.currentId = messageId; - } - } - + const scrollAfterBranchChange = async () => { await tick(); + if (!($settings?.scrollOnBranchChange ?? true)) return; - if ($settings?.scrollOnBranchChange ?? true) { - const element = getMessagesContainer(); - autoScroll = element - ? element.scrollHeight - element.scrollTop <= element.clientHeight + 50 - : false; - - setTimeout(() => { - scrollToBottom(); - }, 100); + const element = getMessagesContainer(); + if (element) { + autoScroll = element.scrollHeight - element.scrollTop <= element.clientHeight + 50; } + + setTimeout(() => { + scrollToBottom(); + }, 100); + }; + + const switchToBranchLeaf = async (messageId: string) => { + if (!messageId) return false; + const navigationGeneration = ++branchNavigationGeneration; + if (messageId === history.currentId) return true; + if (!(await ensureMessageLoaded(messageId))) return false; + if (navigationGeneration !== branchNavigationGeneration) return false; + + messagesCount = 8; + history.currentId = messageId; + const targetHasMore = branchHasMoreByCurrentId.get(messageId); + if (typeof targetHasMore === 'boolean' && history.messageWindow) { + history.messageWindow.hasMore = targetHasMore; + } + history = history; + await scrollAfterBranchChange(); + return true; + }; + + const gotoMessage = async (message: any, idx: number) => { + const siblings = getSiblingIds(message); + if (siblings.length === 0) return; + + const siblingIndex = Math.max(0, Math.min(idx, siblings.length - 1)); + const leafId = getDeepestLeafId(siblings[siblingIndex]); + await switchToBranchLeaf(leafId); + }; + + const showPreviousMessage = async (message: any) => { + const siblings = getSiblingIds(message); + if (siblings.length === 0) return; + + const siblingIndex = Math.max(siblings.indexOf(message.id) - 1, 0); + const leafId = getDeepestLeafId(siblings[siblingIndex]); + await switchToBranchLeaf(leafId); + }; + + const showNextMessage = async (message: any) => { + const siblings = getSiblingIds(message); + if (siblings.length === 0) return; + + const siblingIndex = Math.min(siblings.indexOf(message.id) + 1, siblings.length - 1); + const leafId = getDeepestLeafId(siblings[siblingIndex]); + await switchToBranchLeaf(leafId); }; const rateMessage = async (messageId, rating) => { - history.messages[messageId].annotation = { - ...history.messages[messageId].annotation, - rating: rating - }; - - await updateChat(); + await saveMessage(messageId, { + ...history.messages[messageId], + annotation: { + ...history.messages[messageId].annotation, + rating: rating + } + }); }; const editMessage = async (messageId, { content, files, output = undefined }, submit = true) => { - if ((selectedModels ?? []).filter((id) => id).length === 0) { + if (submit && (selectedModels ?? []).filter((id) => id).length === 0) { toast.error($i18n.t('Model not selected')); return; } @@ -381,9 +512,11 @@ await sendMessage(history, userMessageId); } else { // Edit user message - history.messages[messageId].content = content; - history.messages[messageId].files = files; - await updateChat(); + await saveMessage(messageId, { + ...history.messages[messageId], + content, + files + }); } } else { if (submit) { @@ -417,15 +550,16 @@ await updateChat(); } else { // Edit response message + const updatedMessage = { ...history.messages[messageId] }; if (content !== undefined) { - history.messages[messageId].originalContent = history.messages[messageId].content; - history.messages[messageId].content = content; + updatedMessage.originalContent = history.messages[messageId].content; + updatedMessage.content = content; } if (output !== undefined) { - history.messages[messageId].output = output; - history.messages[messageId].content = ''; + updatedMessage.output = output; + updatedMessage.content = ''; } - await updateChat(); + await saveMessage(messageId, updatedMessage); } } }; @@ -436,15 +570,78 @@ const saveMessage = async (messageId, message) => { if (!history.messages?.[messageId]) { - return; + return null; } - history.messages[messageId] = message; - await updateChat(); + const previousMessage = structuredClone(history.messages[messageId]); + const messagePatch = createMessagePatch(previousMessage, message); + history.messages[messageId] = { ...structuredClone(message), __loaded: true }; + history = history; + await tick(); + + if ($temporaryChatEnabled) { + return history.messages[messageId]; + } + if (Object.keys(messagePatch).length === 0) { + return history.messages[messageId]; + } + + try { + const savedMessage = await updateChatMessageById( + localStorage.token, + chatId, + messageId, + messagePatch + ); + + if (savedMessage) { + history.messages[messageId] = { + ...history.messages[messageId], + ...savedMessage, + __loaded: true + }; + history = history; + await tick(); + } + + void refreshChatList(localStorage.token); + return savedMessage; + } catch (error) { + history.messages[messageId] = previousMessage; + history = history; + await tick(); + toast.error(`${error}`); + throw error; + } }; const deleteMessage = async (messageId) => { const messageToDelete = history.messages[messageId]; + if (!messageToDelete) return; + + if (!$temporaryChatEnabled) { + try { + const response = await deleteChatMessageById(localStorage.token, chatId, messageId); + let nextHistory = response?.chat?.history as ChatHistoryState | undefined; + + if (nextHistory?.currentId && !isChatMessageLoaded(nextHistory, nextHistory.currentId)) { + const historyWindow = await getChatHistoryWindow( + localStorage.token, + chatId, + nextHistory.currentId, + DEFAULT_CHAT_MESSAGE_WINDOW + ); + nextHistory = mergeChatHistoryWindow(nextHistory, historyWindow) as ChatHistoryState; + } + + if (nextHistory) replaceHistory(nextHistory); + void refreshChatList(localStorage.token); + } catch (error) { + toast.error(`${error}`); + } + return; + } + const parentMessageId = messageToDelete.parentId; const childMessageIds = messageToDelete.childrenIds ?? []; @@ -484,15 +681,6 @@ } history.currentId = nextMessageId; history = history; - - if (!$temporaryChatEnabled) { - const res = await deleteChatMessageById(localStorage.token, chatId, messageId); - if (res?.chat?.history) { - history = res.chat.history; - } - - await refreshChatList(localStorage.token); - } }; const triggerScroll = () => { @@ -544,6 +732,8 @@ {gotoMessage} {showPreviousMessage} {showNextMessage} + navigateToMessageId={switchToBranchLeaf} + {ensureMessageLoaded} {updateChat} {editMessage} {deleteMessage} diff --git a/src/lib/components/chat/Messages/Message.svelte b/src/lib/components/chat/Messages/Message.svelte index c899c9dccd..aa2ac05b68 100644 --- a/src/lib/components/chat/Messages/Message.svelte +++ b/src/lib/components/chat/Messages/Message.svelte @@ -25,6 +25,8 @@ export let gotoMessage; export let showPreviousMessage; export let showNextMessage; + export let navigateToMessageId; + export let ensureMessageLoaded; export let updateChat; export let editMessage; @@ -120,6 +122,8 @@ {chatId} {messageId} {selectedModels} + {navigateToMessageId} + {ensureMessageLoaded} isLastMessage={messageId === history?.currentId} {setInputText} {updateChat} diff --git a/src/lib/components/chat/Messages/MultiResponseMessages.svelte b/src/lib/components/chat/Messages/MultiResponseMessages.svelte index 4edc7d1c94..4eafea0b29 100644 --- a/src/lib/components/chat/Messages/MultiResponseMessages.svelte +++ b/src/lib/components/chat/Messages/MultiResponseMessages.svelte @@ -1,12 +1,12 @@