diff --git a/backend/open_webui/migrations/env.py b/backend/open_webui/migrations/env.py index 87a1e54608..2bcd45aea7 100644 --- a/backend/open_webui/migrations/env.py +++ b/backend/open_webui/migrations/env.py @@ -9,6 +9,8 @@ from open_webui.env import DATABASE_PASSWORD, DATABASE_URL, LOG_FORMAT from open_webui.internal.db import enable_iam_token_auth, extract_ssl_params_from_url, reattach_ssl_params_to_url from open_webui.models.auths import Auth from open_webui.models.calendar import Calendar, CalendarEvent, CalendarEventAttendee # noqa: F401 +from open_webui.models.chat_messages import ChatMessage # noqa: F401 +from open_webui.models.chats import Chat # noqa: F401 from sqlalchemy import create_engine, engine_from_config, pool alembic_config = alembic.context.config diff --git a/backend/open_webui/migrations/versions/856c5b02fb54_add_chat_message_meta.py b/backend/open_webui/migrations/versions/856c5b02fb54_add_chat_message_meta.py new file mode 100644 index 0000000000..bfbf2751bb --- /dev/null +++ b/backend/open_webui/migrations/versions/856c5b02fb54_add_chat_message_meta.py @@ -0,0 +1,25 @@ +"""add chat message meta + +Revision ID: 856c5b02fb54 +Revises: 42e2978c7933 +Create Date: 2026-07-16 01:39:39.291935 + +""" + +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = '856c5b02fb54' +down_revision: Union[str, None] = '42e2978c7933' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.add_column('chat_message', sa.Column('meta', sa.JSON(), nullable=True)) + + +def downgrade() -> None: + op.drop_column('chat_message', 'meta') diff --git a/backend/open_webui/models/chat_messages.py b/backend/open_webui/models/chat_messages.py index 5579bdbe95..ff05d5fc21 100644 --- a/backend/open_webui/models/chat_messages.py +++ b/backend/open_webui/models/chat_messages.py @@ -100,6 +100,7 @@ class ChatMessage(Base): files = Column(JSON, nullable=True) sources = Column(JSON, nullable=True) embeds = Column(JSON, nullable=True) + meta = Column(JSON, nullable=True) # Status done = Column(Boolean, default=True) @@ -142,6 +143,7 @@ class ChatMessageModel(BaseModel): files: Optional[list] = None sources: Optional[list] = None embeds: Optional[list] = None + meta: Optional[dict] = None done: bool = True status_history: Optional[list] = None error: Optional[dict | str] = None @@ -192,6 +194,8 @@ class ChatMessageTable: existing.sources = data.get('sources') if 'embeds' in data: existing.embeds = data.get('embeds') + if 'meta' in data: + existing.meta = data.get('meta') if 'done' in data: existing.done = data.get('done', True) if 'status_history' in data or 'statusHistory' in data: @@ -225,6 +229,7 @@ class ChatMessageTable: files=data.get('files'), sources=data.get('sources'), embeds=data.get('embeds'), + meta=data.get('meta'), done=data.get('done', True), status_history=data.get('status_history') or data.get('statusHistory'), error=data.get('error'), diff --git a/backend/open_webui/utils/middleware.py b/backend/open_webui/utils/middleware.py index 0c2724b46f..b55d74338f 100644 --- a/backend/open_webui/utils/middleware.py +++ b/backend/open_webui/utils/middleware.py @@ -136,11 +136,25 @@ logging.basicConfig(stream=sys.stdout, level=GLOBAL_LOG_LEVEL) log = logging.getLogger(__name__) -async def publish_chat_finished_event(request: Request, user: UserModel, metadata: dict, title: str, content: str): +def _last_output_text(output: list | None) -> str: + for item in reversed(output or []): + if item.get('type') != 'message': + continue + parts = item.get('content') or [] + text = ''.join(str(part.get('text') or '') for part in parts if part.get('type') == 'output_text') + if text: + return text + return '' + + +async def publish_chat_finished_event( + request: Request, user: UserModel, metadata: dict, title: str, content: str, output: list | None = None +): chat_id = metadata.get('chat_id') if getattr(request.state, 'internal', False) is True or not chat_id or chat_id.startswith(('channel:', 'local:')): return + content = content or _last_output_text(output) webui_url = await Config.get('webui.url') await publish_event( request, @@ -2578,12 +2592,11 @@ async def process_chat_payload(request, form_data, user, metadata, model): files = [*(files or []), *note_files] use_builtin_tools = ( - (chat and (chat.meta or {}).get('internal') is True and (chat.meta or {}).get('type') == 'note') - or ( - bool(metadata.get('session_id')) - and metadata.get('params', {}).get('function_calling') != 'legacy' - and (model.get('info', {}).get('meta', {}).get('capabilities') or {}).get('builtin_tools', True) - ) + chat and (chat.meta or {}).get('internal') is True and (chat.meta or {}).get('type') == 'note' + ) or ( + bool(metadata.get('session_id')) + and metadata.get('params', {}).get('function_calling') != 'legacy' + and (model.get('info', {}).get('meta', {}).get('capabilities') or {}).get('builtin_tools', True) ) if skill_ids: @@ -3610,7 +3623,7 @@ async def non_streaming_chat_response_handler(response, ctx): }, ) - await publish_chat_finished_event(request, user, metadata, title, content) + await publish_chat_finished_event(request, user, metadata, title, content, response_output) ctx['assistant_message'] = { 'content': content, @@ -5339,7 +5352,7 @@ async def streaming_chat_response_handler(response, ctx): touch=False, ) - await publish_chat_finished_event(request, user, metadata, title, content) + await publish_chat_finished_event(request, user, metadata, title, content, output) await event_emitter( { diff --git a/backend/open_webui/utils/subagents.py b/backend/open_webui/utils/subagents.py index efb562f8cc..85da5784a1 100644 --- a/backend/open_webui/utils/subagents.py +++ b/backend/open_webui/utils/subagents.py @@ -101,13 +101,18 @@ async def process_pending_internal_messages( pending = [ message for message in messages.values() + for meta in [message.get('meta') or {}] if message.get('role') == 'user' and not message.get('childrenIds') and ( - (message.get('meta') or {}).get('async_subagent_result') is True + ( + meta.get('internal') is True + and meta.get('type') == 'subagent' + and meta.get('status') in (None, 'pending') + ) or ( - (message.get('meta') or {}).get('internal') is True - and (message.get('meta') or {}).get('type') == 'timer' + meta.get('internal') is True + and meta.get('type') == 'timer' ) ) ] @@ -116,7 +121,7 @@ async def process_pending_internal_messages( first = pending[0] first_meta = first.get('meta') or {} - kind = 'subagent' if first_meta.get('async_subagent_result') is True else 'timer' + kind = 'timer' if first_meta.get('internal') is True and first_meta.get('type') == 'timer' else 'subagent' parent_id = first.get('parentId') if kind == 'timer' and first_meta.get('timer_id'): timer = await Chats.get_chat_by_id(first_meta['timer_id']) @@ -128,9 +133,16 @@ async def process_pending_internal_messages( batch = [ message for message in pending + for meta in [message.get('meta') or {}] if message.get('parentId') == parent_id and (message.get('model') or model_id) == model_id - and (message.get('meta') or {}).get('async_subagent_result') is True + and ( + ( + meta.get('internal') is True + and meta.get('type') == 'subagent' + and meta.get('status') in (None, 'pending') + ) + ) ] combined_content = '\n\n'.join(message.get('content', '') for message in batch if message.get('content')) if kind == 'timer': @@ -153,7 +165,7 @@ async def process_pending_internal_messages( for message in batch if (message.get('meta') or {}).get('subagent_chat_id') ] - combined_meta = {'async_subagent_result': True} + combined_meta = {'internal': True, 'type': 'subagent'} if len(delegation_ids) == 1: combined_meta['delegation_id'] = delegation_ids[0] elif delegation_ids: @@ -163,8 +175,7 @@ async def process_pending_internal_messages( elif subagent_chat_ids: combined_meta['subagent_chat_ids'] = subagent_chat_ids - pending_flag = 'timer_pending' if kind == 'timer' else 'async_subagent_pending' - reuse_message = len(batch) == 1 and not (first.get('meta') or {}).get(pending_flag) + reuse_message = len(batch) == 1 and (first.get('meta') or {}).get('status') != 'pending' user_message_id = first['id'] if reuse_message else str(uuid4()) removed_ids = set() if not reuse_message: @@ -564,7 +575,8 @@ async def delegate( pending_message_id = str(uuid4()) pending_meta = { - 'async_subagent_result': True, + 'internal': True, + 'type': 'subagent', 'delegation_id': delegation_id, 'subagent_chat_id': chat_id, } @@ -587,7 +599,7 @@ async def delegate( raise asyncio.CancelledError return result if await has_active_tasks(request.app.state.redis, parent_chat_id): - pending_message['meta']['async_subagent_pending'] = True + pending_message['meta']['status'] = 'pending' updated_chat = copy.deepcopy(parent.chat) updated_history = updated_chat.setdefault('history', {}) updated_messages = updated_history.setdefault('messages', {}) @@ -604,7 +616,7 @@ async def delegate( data=pending_message, ) - if pending_message['meta'].get('async_subagent_pending') is True: + if pending_message['meta'].get('status') == 'pending': from open_webui.socket.main import sio await sio.emit( diff --git a/backend/open_webui/utils/timers.py b/backend/open_webui/utils/timers.py index bd14841350..e0f603b6a0 100644 --- a/backend/open_webui/utils/timers.py +++ b/backend/open_webui/utils/timers.py @@ -21,6 +21,7 @@ from open_webui.models.chat_messages import ChatMessages from open_webui.models.chats import Chat, ChatForm, Chats from open_webui.models.users import UserModel, Users from open_webui.tasks import has_active_tasks +from open_webui.utils.misc import get_message_list log = logging.getLogger(__name__) @@ -148,7 +149,7 @@ async def create_timer( 'parent_chat_id': parent_chat_id, 'parent_message_id': parent_message_id, 'timer_at': due_at, - 'timer_status': 'pending', + 'status': 'pending', 'timer_model_id': model_id, 'timer_task_message_id': user_message_id, 'cancel_on': selected_events, @@ -175,7 +176,7 @@ async def claim_due_timers(now_ns: int, limit: int = 10) -> list[tuple[str, str] select(Chat) .where(Chat.meta['internal'].as_boolean().is_(True)) .where(Chat.meta['type'].as_string() == 'timer') - .where(Chat.meta['timer_status'].as_string() == 'pending') + .where(Chat.meta['status'].as_string() == 'pending') ) if db.bind.dialect.name == 'postgresql': stmt = stmt.with_for_update(skip_locked=True) @@ -194,7 +195,7 @@ async def claim_due_timers(now_ns: int, limit: int = 10) -> list[tuple[str, str] claim_id = str(uuid4()) row.meta = { **(row.meta or {}), - 'timer_status': 'running', + 'status': 'running', 'timer_started_at': now_ns, 'timer_claim_id': claim_id, } @@ -211,7 +212,7 @@ async def cancel_timers_for_chat(parent_chat_id: str, event: Literal['chat.read' .where(Chat.meta['internal'].as_boolean().is_(True)) .where(Chat.meta['type'].as_string() == 'timer') .where(Chat.meta['parent_chat_id'].as_string() == parent_chat_id) - .where(Chat.meta['timer_status'].as_string() == 'pending') + .where(Chat.meta['status'].as_string() == 'pending') ) now_ns = int(time.time_ns()) for row in result.scalars().all(): @@ -220,7 +221,7 @@ async def cancel_timers_for_chat(parent_chat_id: str, event: Literal['chat.read' continue row.meta = { **meta, - 'timer_status': 'cancelled', + 'status': 'cancelled', 'timer_cancelled_at': now_ns, 'timer_cancelled_by': event, } @@ -232,13 +233,13 @@ async def execute_due_timer(app, timer_id: str, claim_id: str | None = None) -> lock = _timer_locks.setdefault(timer_id, asyncio.Lock()) async with lock: from open_webui.socket.main import sio - from open_webui.utils.subagents import _parent_locks, process_pending_internal_messages + from open_webui.utils.subagents import _parent_locks timer = await Chats.get_chat_by_id(timer_id) if not timer: return meta = timer.meta or {} - if meta.get('timer_status') != 'running': + if meta.get('status') != 'running': return if claim_id is not None and meta.get('timer_claim_id') != claim_id: return @@ -246,24 +247,24 @@ async def execute_due_timer(app, timer_id: str, claim_id: str | None = None) -> parent_chat_id = meta.get('parent_chat_id') or '' parent = await Chats.get_chat_by_id_and_user_id(parent_chat_id, timer.user_id) if not parent: - await _set_timer_status(timer_id, 'error', timer_error='parent chat no longer exists') + await _set_timer_state(timer_id, 'error', timer_error='parent chat no longer exists') return prompt_message_id = meta.get('timer_task_message_id') prompt_message = await Chats.get_message_by_id_and_message_id(timer_id, prompt_message_id) if not prompt_message: - await _set_timer_status(timer_id, 'error', timer_error='timer task message is missing') + await _set_timer_state(timer_id, 'error', timer_error='timer task message is missing') return user = await Users.get_user_by_id(timer.user_id) if not user: - await _set_timer_status(timer_id, 'error', timer_error='timer user no longer exists') + await _set_timer_state(timer_id, 'error', timer_error='timer user no longer exists') return run = meta.get('run') or {} model_id = run.get('model_id') or meta.get('timer_model_id') if not model_id: - await _set_timer_status(timer_id, 'error', timer_error='model context is missing') + await _set_timer_state(timer_id, 'error', timer_error='model context is missing') return prompt = prompt_message.get('content') or '' @@ -273,6 +274,10 @@ async def execute_due_timer(app, timer_id: str, claim_id: str | None = None) -> ) user_message_id = str(uuid4()) + assistant_message_id = str(uuid4()) + user_message = None + assistant_message = None + message_list = [] parent_lock = _parent_locks.setdefault(parent_chat_id, asyncio.Lock()) async with parent_lock: async with get_async_db() as db: @@ -282,7 +287,19 @@ async def execute_due_timer(app, timer_id: str, claim_id: str | None = None) -> result = await db.execute(stmt) parent = result.scalar_one_or_none() if not parent: - await _set_timer_status(timer_id, 'error', timer_error='parent chat no longer exists') + await _set_timer_state(timer_id, 'error', timer_error='parent chat no longer exists') + return + if await has_active_tasks(app.state.redis, parent_chat_id): + timer_row = await db.get(Chat, timer_id) + if timer_row: + timer_row.meta = { + **(timer_row.meta or {}), + 'status': 'pending', + 'timer_claim_id': None, + 'timer_started_at': None, + } + timer_row.updated_at = int(time.time()) + await db.commit() return parent_chat = copy.deepcopy(parent.chat or {}) @@ -298,74 +315,106 @@ async def execute_due_timer(app, timer_id: str, claim_id: str | None = None) -> if done_assistants else meta.get('parent_message_id') ) - pending_meta = {'internal': True, 'type': 'timer', 'timer_id': timer_id} - if await has_active_tasks(app.state.redis, parent_chat_id): - pending_meta['timer_pending'] = True + message_list = get_message_list(messages, parent_id) user_message = { 'id': user_message_id, 'parentId': parent_id, - 'childrenIds': [], + 'childrenIds': [assistant_message_id], 'role': 'user', 'content': prompt, 'model': model_id, - 'meta': pending_meta, + 'meta': {'internal': True, 'type': 'timer', 'timer_id': timer_id}, + 'timestamp': int(time.time()), + } + assistant_message = { + 'id': assistant_message_id, + 'parentId': user_message_id, + 'childrenIds': [], + 'role': 'assistant', + 'content': '', + 'done': False, + 'model': model_id, 'timestamp': int(time.time()), } messages[user_message_id] = user_message + messages[assistant_message_id] = assistant_message if parent_id and parent_id in messages: children = messages[parent_id].setdefault('childrenIds', []) if user_message_id not in children: children.append(user_message_id) parent.chat = parent_chat + history['currentId'] = assistant_message_id parent.updated_at = int(time.time()) timer_row = await db.get(Chat, timer_id) if timer_row: timer_row.meta = { **(timer_row.meta or {}), - 'timer_status': 'dispatched', - 'timer_dispatched_at': int(time.time_ns()), + 'status': 'completed', + 'timer_completed_at': int(time.time_ns()), } timer_row.updated_at = int(time.time()) await db.commit() await ChatMessages.upsert_message(user_message_id, parent_chat_id, timer.user_id, user_message) + await ChatMessages.upsert_message(assistant_message_id, parent_chat_id, timer.user_id, assistant_message) - if user_message['meta'].get('timer_pending') is True: - await sio.emit( - 'events', - { - 'chat_id': parent_chat_id, - 'message_id': user_message_id, - 'data': {'type': 'chat:reload'}, - }, - room=f'user:{timer.user_id}', - ) - elif not await has_active_tasks(app.state.redis, parent_chat_id): - request = Request( - { - 'type': 'http', - 'asgi': {'version': '3.0', 'spec_version': '2.0'}, - 'method': 'POST', - 'path': '/api/v1/timers/internal', - 'query_string': b'', - 'headers': Headers({}).raw, - 'client': ('127.0.0.1', 0), - 'server': ('127.0.0.1', 80), - 'scheme': 'http', - 'app': app, - } - ) - request.state.token = None - request.state.enable_api_keys = False - await process_pending_internal_messages(request, parent_chat_id, user.id, run) + await sio.emit( + 'events', + { + 'chat_id': parent_chat_id, + 'message_id': assistant_message_id, + 'data': {'type': 'chat:reload'}, + }, + room=f'user:{timer.user_id}', + ) + form_data = { + 'model': model_id, + 'messages': [ + *([{'role': 'system', 'content': run.get('system_prompt')}] if run.get('system_prompt') else []), + *message_list, + {'role': 'user', 'content': prompt}, + ], + 'stream': True, + 'chat_id': parent_chat_id, + 'id': assistant_message_id, + 'parent_id': user_message.get('parentId'), + 'user_message': user_message, + 'session_id': run.get('session_id') or f'timer:{parent_chat_id}', + 'background_tasks': {}, + 'tool_ids': run.get('tool_ids') or [], + 'skill_ids': run.get('skill_ids') or [], + 'filter_ids': run.get('filter_ids') or [], + 'features': run.get('features') or {}, + 'files': run.get('files') or [], + 'variables': run.get('variables') or {}, + } + if run.get('terminal_id'): + form_data['terminal_id'] = run['terminal_id'] + request = Request( + { + 'type': 'http', + 'asgi': {'version': '3.0', 'spec_version': '2.0'}, + 'method': 'POST', + 'path': '/api/v1/timers/internal', + 'query_string': b'', + 'headers': Headers({}).raw, + 'client': ('127.0.0.1', 0), + 'server': ('127.0.0.1', 80), + 'scheme': 'http', + 'app': app, + } + ) + request.state.token = None + request.state.enable_api_keys = False + await app.state.CHAT_COMPLETION_HANDLER(request, form_data, user=user) -async def _set_timer_status(timer_id: str, status: str, **fields) -> None: +async def _set_timer_state(timer_id: str, status: str, **fields) -> None: async with get_async_db() as db: row = await db.get(Chat, timer_id) if not row: return - row.meta = {**(row.meta or {}), 'timer_status': status, **fields} + row.meta = {**(row.meta or {}), 'status': status, **fields} row.updated_at = int(time.time()) await db.commit() diff --git a/src/lib/components/chat/Messages/UserMessage.svelte b/src/lib/components/chat/Messages/UserMessage.svelte index 30200ac78d..a5cb0851b8 100644 --- a/src/lib/components/chat/Messages/UserMessage.svelte +++ b/src/lib/components/chat/Messages/UserMessage.svelte @@ -133,7 +133,7 @@ id="message-{message.id}" style="scroll-margin-top: 3rem;" > - {#if !($settings?.chatBubble ?? true) && !message?.meta?.async_subagent_result && !(message?.meta?.internal === true && message?.meta?.type === 'timer')} + {#if !($settings?.chatBubble ?? true) && !(message?.meta?.internal === true && message?.meta?.type === 'subagent') && !(message?.meta?.internal === true && message?.meta?.type === 'timer')}