open-webui/backend/open_webui/utils/middleware.py

import asyncio
import base64
import copy
import hashlib
import html
import inspect
import json
import logging
import mimetypes
import os
import random
import re
import sys
import textwrap
import time
from concurrent.futures import ThreadPoolExecutor
from typing import Any, Optional
from urllib.parse import unquote
from uuid import uuid4

from aiocache import cached
from fastapi import HTTPException, Request
from fastapi.responses import HTMLResponse, JSONResponse
from open_webui.config import (
    CACHE_DIR,
    CODE_INTERPRETER_BLOCKED_MODULES,
    CODE_INTERPRETER_PYODIDE_PROMPT,
    DEFAULT_CODE_INTERPRETER_PROMPT,
    DEFAULT_TOOLS_FUNCTION_CALLING_PROMPT_TEMPLATE,
    DEFAULT_VOICE_MODE_PROMPT_TEMPLATE,
)
from open_webui.constants import TASKS
from open_webui.env import (
    BYPASS_MODEL_ACCESS_CONTROL,
    CHAT_RESPONSE_MAX_TOOL_CALL_ITERATIONS,
    CHAT_RESPONSE_STREAM_DELTA_CHUNK_SIZE,
    ENABLE_API_OUTLET_FILTERS,
    ENABLE_CHAT_RESPONSE_BASE64_IMAGE_URL_CONVERSION,
    ENABLE_CHAT_RESPONSE_STREAM_INPLACE_APPEND,
    ENABLE_FUNCTIONS,
    ENABLE_QUERIES_CACHE,
    ENABLE_REALTIME_CHAT_SAVE,
    ENABLE_RESPONSES_API_STATEFUL,
    ENABLE_TOOL_SERVERS,
    GLOBAL_LOG_LEVEL,
    RAG_SYSTEM_CONTEXT,
)
from open_webui.events import EVENTS, build_event, publish_event
from open_webui.models.access_grants import AccessGrants
from open_webui.models.chats import Chats
from open_webui.models.config import Config
from open_webui.models.folders import Folders
from open_webui.models.models import Models
from open_webui.models.users import UserModel, Users
from open_webui.retrieval.utils import filter_source_metadata, get_sources_from_items
from open_webui.routers.images import (
    CreateImageForm,
    EditImageForm,
    image_edits,
    image_generations,
)
from open_webui.routers.pipelines import (
    get_sorted_filters,
    process_pipeline_inlet_filter,
    process_pipeline_outlet_filter,
)
from open_webui.routers.retrieval import (
    SearchForm,
    process_web_search,
)
from open_webui.routers.tasks import (
    generate_chat_tags,
    generate_follow_ups,
    generate_image_prompt,
    generate_queries,
    generate_title,
)
from open_webui.socket.main import (
    get_event_call,
    get_event_emitter,
)
from open_webui.tasks import append_response_stream, clear_response_stream, save_response_stream
from open_webui.utils.access_control import has_permission
from open_webui.utils.access_control.files import get_owner_accessible_folder_files
from open_webui.utils.access_control.folders import has_folder_access
from open_webui.utils.ask_user import stage_ask_user_tool_calls
from open_webui.utils.audit import log_model_usage
from open_webui.utils.auth import get_verified_user_by_id
from open_webui.utils.chat import generate_chat_completion
from open_webui.utils.chat_id import is_saved_chat_id, is_temporary_chat_id
from open_webui.utils.code_interpreter import execute_code_jupyter
from open_webui.utils.context_compaction import apply_summary_checkpoint, compact_messages_for_request
from open_webui.utils.files import (
    convert_markdown_base64_images,
    get_file_url_from_base64,
    get_image_base64_from_url,
    get_image_url_from_base64,
)
from open_webui.utils.filter import (
    FilterContext,
    get_filter_context,
    get_filter_functions,
    process_filter_functions,
)
from open_webui.utils.json_codec import JSONCodec, dumps_bytes
from open_webui.utils.mcp.apps import MCPAppResult
from open_webui.utils.memory import add_memory_context, recall_memories, review_memory_after_turn
from open_webui.utils.misc import (
    add_or_update_system_message,
    add_or_update_user_message,
    convert_output_to_messages,
    extract_urls,
    get_content_from_message,
    get_last_assistant_message,
    get_last_user_message,
    get_last_user_message_item,
    get_message_list,
    get_output_text,
    get_paired_tool_call_ids,
    get_reasoning_details,
    get_response_error_detail,
    get_system_message,
    is_raster_image_content_type,
    merge_system_messages,
    prepend_to_first_user_message_content,
    set_last_user_message_content,
    strip_empty_content_blocks,
)
from open_webui.utils.oauth import get_system_oauth_token
from open_webui.utils.payload import apply_params_to_form_data, apply_system_prompt_to_body
from open_webui.utils.plugin import load_function_module_by_id
from open_webui.utils.response import (
    convert_responses_stream_to_chat_completions,
    get_response_delta_key,
    handle_responses_streaming_event,
    merge_usage,
    normalize_usage,
    update_usage_snapshot,
)
from open_webui.utils.sanitize import sanitize_code
from open_webui.utils.skills import (
    apply_skills_create_prompt,
    extract_skill_ids_from_messages,
    has_prior_user_message,
    replace_skill_mentions_with_labels,
    strip_skill_mentions,
)
from open_webui.utils.task import (
    get_task_model_id,
    parse_follow_ups,
    rag_template,
    tools_function_calling_generation_template,
)
from open_webui.utils.terminals import filter_user_terminal_tools, upload_files_to_terminal
from open_webui.utils.tool_approval import complete_tool_call, pending_tool_calls
from open_webui.utils.tool_permissions import (
    ToolApprovalRequired,
    get_tool_permissions_auto_config,
    requires_tool_approval,
    set_tool_approval_prompt,
)
from open_webui.utils.tool_calling import (
    INCOMPLETE_CALL_ERROR,
    add_tool_call_metadata,
    final_answer_payload,
    prepare_tool_call,
    recover_tool_calls,
    set_tool_call_metadata,
    split_tool_calls,
    tool_repetition_error,
)
from open_webui.utils.tool_search import apply_tool_search, resolve_tool_call, split_tool_call_batches
from open_webui.utils.tools import (
    connect_mcp_server,
    display_file,
    get_attached_knowledge,
    get_builtin_tool_spec,
    get_builtin_tools,
    get_terminal_tools,
    get_tools,
    get_updated_tool_function,
    is_file_context_enabled,
)
from starlette.responses import JSONResponse, Response, StreamingResponse

logging.basicConfig(stream=sys.stdout, level=GLOBAL_LOG_LEVEL)
log = logging.getLogger(__name__)


def _is_tool_result_error(value: Any) -> bool:
    if isinstance(value, str):
        text = value.strip().lower()
        if (
            text.startswith('error:')
            or text.startswith('exception:')
            or text.startswith('traceback')
            or text.startswith('http error!')
        ):
            return True

    parsed = value
    while isinstance(parsed, str):
        try:
            parsed = JSONCodec.loads(parsed)
        except (JSONCodec.JSONDecodeError, TypeError, ValueError):
            break

    if not isinstance(parsed, dict):
        return False

    error = parsed.get('error')
    if isinstance(error, str):
        has_error = bool(error.strip())
    else:
        has_error = isinstance(error, (dict, list)) and bool(error)
    if has_error:
        return True

    status = parsed.get('status')
    if isinstance(status, str) and status.strip().lower() in {'error', 'failed'}:
        return True

    if parsed.get('success') is False or parsed.get('ok') is False:
        message = parsed.get('message')
        return has_error or (
            bool(message.strip()) if isinstance(message, str) else isinstance(message, (dict, list)) and bool(message)
        )

    return False


def normalize_messages_for_model(form_data: dict) -> dict:
    form_data['messages'] = strip_empty_content_blocks(form_data.get('messages', []))
    form_data['messages'] = merge_system_messages(form_data.get('messages', []))
    return form_data


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 is_saved_chat_id(chat_id):
        return

    content = content or get_output_text(output)
    webui_url = await Config.get('webui.url')
    model = (getattr(request.app.state, 'MODELS', {}) or {}).get(metadata.get('model_id'), {})
    event = build_event(
        request,
        EVENTS.CHAT_FINISHED,
        actor=user,
        subject_id=chat_id,
        subject_type='chat',
        data={
            'user_id': user.id,
            'chat_id': chat_id,
            'message_id': metadata.get('message_id'),
            'model_id': metadata.get('model_id'),
            'model_name': model.get('name'),
            'title': title,
            'url': f'{webui_url}/c/{chat_id}' if webui_url else f'/c/{chat_id}',
            'message': content,
        },
        message=title or 'Chat finished',
    )
    try:
        from open_webui.internal.db import get_async_db_context
        from open_webui.models.notifications import emit_changes
        from open_webui.utils.inbox import consume_event

        async with get_async_db_context() as db:
            recipients = await consume_event(event, db=db)
            await db.commit()
        await emit_changes(recipients)
    except Exception:
        log.exception('Chat mention notification failed for %s', chat_id)
    await publish_event(request, event)
    event_emitter = await get_event_emitter(metadata, update_db=False)
    if event_emitter:
        folder_id = metadata.get('folder_id') or await Chats.get_chat_folder_id(chat_id, metadata.get('user_id'))
        await event_emitter({'type': 'chat:list', 'data': {'chat_id': chat_id, 'folder_id': folder_id}})


async def publish_chat_failed_event(request: Request, user: UserModel, metadata: dict, error: str):
    chat_id = metadata.get('chat_id')
    if getattr(request.state, 'internal', False) is True or not is_saved_chat_id(chat_id):
        return

    webui_url = await Config.get('webui.url')
    await publish_event(
        request,
        EVENTS.CHAT_FAILED,
        actor=user,
        subject_id=chat_id,
        subject_type='chat',
        data={
            'user_id': user.id,
            'chat_id': chat_id,
            'message_id': metadata.get('message_id'),
            'model_id': metadata.get('model_id'),
            'url': f'{webui_url}/c/{chat_id}' if webui_url else f'/c/{chat_id}',
            'message': error,
        },
        message='Chat failed',
    )


# Let every wandering thought find its closing tag,
# and let the answer arrive before we forget the question.
DEFAULT_REASONING_TAGS = [
    ('<think>', '</think>'),
    ('<thinking>', '</thinking>'),
    ('<reason>', '</reason>'),
    ('<reasoning>', '</reasoning>'),
    ('<thought>', '</thought>'),
    ('<Thought>', '</Thought>'),
    ('<|begin_of_thought|>', '<|end_of_thought|>'),
    ('◁think▷', '◁/think▷'),
]

DEFAULT_SOLUTION_TAGS = [('<|begin_of_solution|>', '<|end_of_solution|>')]
DEFAULT_CODE_INTERPRETER_TAGS = [('<code_interpreter>', '</code_interpreter>')]


def _start_tag_pattern(start_tag: str) -> str:
    if start_tag.startswith('<') and start_tag.endswith('>'):
        return rf'<{re.escape(start_tag[1:-1])}(\s.*?)?>'
    return re.escape(start_tag)


def output_id(prefix: str) -> str:
    """Generate OR-style ID: prefix + 24-char hex UUID."""
    return f'{prefix}_{uuid4().hex[:24]}'


def build_terminal_file_tool_result(
    tool_function_name: str,
    tool_function_params: dict,
    tool_result: Any,
    tool: dict | None,
    metadata: dict | None,
) -> dict | None:
    if isinstance(tool_result, (list, tuple)) and tool_result and isinstance(tool_result[0], dict):
        tool_result = tool_result[0]

    if tool_function_name != 'display_file' or not isinstance(tool_result, dict) or tool_result.get('exists') is False:
        return None

    tool_id = (tool or {}).get('tool_id', '')
    terminal_id = metadata.get('terminal_id') if metadata else None
    if isinstance(tool_id, str) and tool_id.startswith('terminal:'):
        terminal_id = tool_id.split(':', 1)[1]

    server_url = ((tool or {}).get('server') or {}).get('url')
    terminal_selector = terminal_id or server_url
    path = tool_result.get('path') or tool_function_params.get('path')
    if not terminal_selector or not path:
        return None
    mime_type, _ = mimetypes.guess_type(path)
    mime_type = mime_type or 'application/octet-stream'
    page = tool_result.get('page') or tool_function_params.get('page')

    return {
        **tool_result,
        'type': 'file',
        'source': 'open_terminal',
        'exists': True,
        'displayed': tool_function_params.get('inline') is True,
        'terminal_selector': terminal_selector,
        **({'terminal_id': terminal_id} if terminal_id else {}),
        **({'terminal_url': server_url} if server_url and not terminal_id else {}),
        'session_id': metadata.get('chat_id') if metadata else None,
        'path': path,
        'full_path': tool_result.get('full_path') or path,
        'name': tool_result.get('name') or os.path.basename(path),
        'mime_type': tool_result.get('mime_type') or tool_result.get('content_type') or mime_type,
        'content_type': tool_result.get('content_type') or tool_result.get('mime_type') or mime_type,
        **({'page': page} if page else {}),
    }


def tool_result_content(tool_result: Any) -> str:
    if not tool_result:
        return ''
    if isinstance(tool_result, (dict, list)):
        return JSONCodec.dumps(tool_result, ensure_ascii=False)
    return str(tool_result)


def append_to_text_field(item: dict, key: str, value: str) -> None:
    # Default: keep the existing field intact until concatenation succeeds.
    # Non-string values and str subclasses also use this path to preserve errors.
    if not ENABLE_CHAT_RESPONSE_STREAM_INPLACE_APPEND or type(item[key]) is not str or type(value) is not str:
        item[key] += value
        return

    # Opt-in: dropping the dict's reference lets CPython extend an unshared str
    # in place. Only a host that is already out of memory can leave the field empty.
    text = item[key]
    item[key] = ''
    text += value
    item[key] = text


def merge_streamed_reasoning_details(target: list, details) -> None:
    items = details if isinstance(details, list) else [details]
    for item in items:
        if not isinstance(item, dict):
            continue

        index = item.get('index')
        existing = (
            next((detail for detail in target if detail.get('index') == index), None)
            if isinstance(index, int)
            else None
        )
        if existing is None:
            target.append(dict(item))
            continue

        for key, value in item.items():
            if key in ('text', 'summary') and isinstance(value, str) and isinstance(existing.get(key), str):
                append_to_text_field(existing, key, value)
            else:
                existing[key] = value


CITATION_TOOL_NAMES = ('fetch_url', 'view_file', 'view_knowledge_file', 'query_knowledge_files', 'query_chat_files')


def get_citation_source_from_tool_result(
    tool_name: str, tool_params: dict, tool_result: str, tool_id: str = ''
) -> list[dict]:
    """
    Parse a tool's result and convert it to source dicts for citation display.

    Follows the source format conventions from get_sources_from_items:
    - source: file/item info object with id, name, type
    - document: list of document contents
    - metadata: list of metadata objects with source, file_id, name fields

    Returns a list of sources (usually one, but query_knowledge_files/query_chat_files may return multiple).
    """
    try:
        try:
            tool_result = JSONCodec.loads(tool_result)
        except (JSONCodec.JSONDecodeError, TypeError):
            pass  # keep tool_result as-is (e.g. fetch_url returns plain text)
        if isinstance(tool_result, dict) and 'error' in tool_result:
            return []

        if tool_name in ('view_knowledge_file', 'view_file'):
            if not isinstance(tool_result, dict):
                return []

            file_data = tool_result
            filename = file_data.get('filename', 'Unknown File')
            file_id = file_data.get('id', '')
            knowledge_name = file_data.get('knowledge_name', '')

            return [
                {
                    'source': {
                        'id': file_id,
                        'name': filename,
                        'type': 'file',
                    },
                    'document': [file_data.get('content', '')],
                    'metadata': [
                        {
                            'file_id': file_id,
                            'name': filename,
                            'source': filename,
                            **({'knowledge_name': knowledge_name} if knowledge_name else {}),
                        }
                    ],
                }
            ]

        elif tool_name == 'fetch_url':
            url = tool_params.get('url', '')
            content = tool_result if isinstance(tool_result, str) else str(tool_result)
            snippet = content[:500] + ('...' if len(content) > 500 else '')

            return [
                {
                    'source': {'name': url or 'fetch_url', 'id': url or 'fetch_url'},
                    'document': [snippet],
                    'metadata': [
                        {
                            'source': url,
                            'name': url,
                            'url': url,
                        }
                    ],
                }
            ]

        elif tool_name in ('query_knowledge_files', 'query_chat_files'):
            if not isinstance(tool_result, list):
                return []

            chunks = tool_result

            # Group chunks by source for better citation display
            # Each unique source becomes a separate source entry
            sources_by_file = {}

            for chunk in chunks:
                source_name = chunk.get('source', 'Unknown')
                file_id = chunk.get('file_id', '')
                note_id = chunk.get('note_id', '')
                chunk_type = chunk.get('type', 'file')
                content = chunk.get('content', '')

                # Use file_id or note_id as the key
                key = file_id or note_id or source_name

                if key not in sources_by_file:
                    sources_by_file[key] = {
                        'source': {
                            'id': file_id or note_id,
                            'name': source_name,
                            'type': chunk_type,
                        },
                        'document': [],
                        'metadata': [],
                        'distances': [],
                    }

                sources_by_file[key]['document'].append(content)
                sources_by_file[key]['metadata'].append(
                    {
                        'file_id': file_id,
                        'name': source_name,
                        'source': source_name,
                        **({'note_id': note_id} if note_id else {}),
                    }
                )
                if 'distance' in chunk:
                    sources_by_file[key]['distances'].append(chunk['distance'])

            # Return all grouped sources as a list
            if sources_by_file:
                return list(sources_by_file.values())

            # Empty result fallback
            return []

        else:
            # Fallback for other tools
            return [
                {
                    'source': {
                        'name': tool_name,
                        'type': 'tool',
                        'id': tool_id or tool_name,
                    },
                    'document': [str(tool_result)],
                    'metadata': [{'source': tool_name, 'name': tool_name}],
                }
            ]
    except Exception as e:
        log.exception(f'Error parsing tool result for {tool_name}: {e}')
        return [
            {
                'source': {'name': tool_name, 'type': 'tool'},
                'document': [str(tool_result)],
                'metadata': [{'source': tool_name}],
            }
        ]


RESPONSE_COMPLETION_RESPONSE_FIELDS = ('error', 'id', 'output', 'usage')


def get_response_completion_event_data(event: dict) -> dict:
    """Build the data payload for response:completion events."""
    response = event.get('response')
    if not isinstance(response, dict):
        return event

    response_data = {key: response[key] for key in RESPONSE_COMPLETION_RESPONSE_FIELDS if key in response}

    return {
        **event,
        'response': response_data,
    }


def get_source_context(sources: list, source_ids: dict = None, include_content: bool = True) -> str:
    """
    Build <source> tag context string from citation sources.
    """
    context_string = ''
    if source_ids is None:
        source_ids = {}
    for source in sources:
        for doc, meta in zip(source.get('document', []), source.get('metadata', [])):
            source_id = meta.get('source') or source.get('source', {}).get('id') or 'N/A'
            if source_id not in source_ids:
                source_ids[source_id] = len(source_ids) + 1
            src_name = source.get('source', {}).get('name')
            src_type = source.get('source', {}).get('type')
            src_rid = source.get('source', {}).get('id') if src_type != 'text' else None
            body = doc if include_content else ''
            extra_attrs = ''
            for key, value in filter_source_metadata(meta).items():
                if key in ('id', 'name', 'resource-type', 'resource-id'):
                    continue
                extra_attrs += f' {key}="{html.escape(str(value))}"'
            context_string += (
                f'<source id="{source_ids[source_id]}"'
                + (f' name="{src_name}"' if src_name else '')
                + (f' resource-type="{src_type}"' if src_type else '')
                + (f' resource-id="{src_rid}"' if src_rid else '')
                + extra_attrs
                + f'>{body}</source>\n'
            )
    return context_string


async def apply_source_context_to_messages(
    request: Request,
    messages: list,
    sources: list,
    user_message: str,
    include_content: bool = True,
) -> list:
    """
    Build source context from citation sources and apply to messages.
    Uses RAG template to format context for model consumption.

    When include_content is False, emit <source> tags with id/name but no
    document body — useful when the content is already present elsewhere
    (e.g. in a tool result message) and only citation markers are needed.
    """
    if not sources or not user_message:
        return messages

    context = get_source_context(sources, include_content=include_content)

    context = context.strip()
    if not context:
        return messages

    if RAG_SYSTEM_CONTEXT:
        return add_or_update_system_message(
            await rag_template(await Config.get('rag.template'), context, user_message),
            messages,
            append=True,
        )
    else:
        return add_or_update_user_message(
            await rag_template(await Config.get('rag.template'), context, user_message),
            messages,
            append=False,
        )


async def tool_result_with_citations(name, params, result, tool, model, source_ids, user_message, event_emitter) -> str:
    content = tool_result_content(result)
    capabilities = model.get('info', {}).get('meta', {}).get('capabilities') or {}
    if not capabilities.get('citations', True) or name not in CITATION_TOOL_NAMES or not result:
        return content
    try:
        sources = get_citation_source_from_tool_result(name, params, result, (tool or {}).get('tool_id', ''))
    except Exception as e:
        log.exception(f'Error extracting citation source: {e}')
        return content
    if event_emitter:
        for source in sources:
            await event_emitter({'type': 'source', 'data': source})
    if sources:
        context = get_source_context(sources, source_ids, include_content=False).strip()
        if context:
            citation_context = await rag_template(await Config.get('rag.template'), context, user_message or '')
            # Save markers with this result so later requests replay the same prefix.
            content = f'{content}\n\n{citation_context}'
    return content


BASE64_IMAGE_DATA_URI_RE = re.compile(r'data:image/[a-zA-Z0-9.+-]+;base64,[A-Za-z0-9+/]+={0,2}', re.IGNORECASE)


def extract_base64_images(value: Any, files: list) -> Any:
    """Move base64 image data URIs out of a tool result so they do not reach the model as text."""
    if isinstance(value, str):
        if BASE64_IMAGE_DATA_URI_RE.fullmatch(value):
            files.append({'type': 'image', 'url': value})
            return '[image]'
        return value
    if isinstance(value, dict):
        return {key: extract_base64_images(item, files) for key, item in value.items()}
    if isinstance(value, list):
        return [extract_base64_images(item, files) for item in value]
    if isinstance(value, tuple):
        return tuple(extract_base64_images(item, files) for item in value)
    return value


async def store_tool_result_image(request, image_url, metadata, user):
    """Keep saved tool images out of chat JSON, falling back to inline data if storage fails."""
    metadata = metadata or {}
    if (
        not isinstance(image_url, str)
        or not image_url.startswith('data:image/')
        or not is_saved_chat_id(metadata.get('chat_id'))
    ):
        return image_url

    try:
        stored_url = await get_file_url_from_base64(
            request,
            image_url,
            {key: metadata.get(key) for key in ('chat_id', 'message_id', 'session_id')},
            user,
        )
        return stored_url or image_url
    except Exception:
        log.warning('Could not store tool image; retaining inline image')
        return image_url


async def process_tool_result(
    request,
    tool_function_name,
    tool_result,
    tool_type,
    metadata=None,
    user=None,
):
    tool_result_embeds = []
    if isinstance(tool_result, MCPAppResult):
        tool_result_embeds.append(tool_result.embed)
        tool_result = tool_result.context
    EXTERNAL_TOOL_TYPES = ('external', 'action', 'terminal')

    # Support (HTMLResponse, result_context) tuples: the optional second
    # element lets tool authors provide the LLM with actionable context
    # about the generated embed instead of the generic fallback message.
    result_context = None
    if isinstance(tool_result, tuple) and len(tool_result) == 2 and isinstance(tool_result[0], HTMLResponse):
        tool_result, result_context = tool_result

    if isinstance(tool_result, HTMLResponse):
        content_disposition = tool_result.headers.get('Content-Disposition', '')
        if 'inline' in content_disposition:
            content = tool_result.body.decode('utf-8', 'replace')
            tool_result_embeds.append(content)

            if 200 <= tool_result.status_code < 300:
                if result_context is not None and isinstance(result_context, (str, dict, list)):
                    tool_result = result_context
                else:
                    tool_result = {
                        'status': 'success',
                        'code': 'ui_component',
                        'message': f'{tool_function_name}: Embedded UI result is active and visible to the user.',
                    }
            elif 400 <= tool_result.status_code < 500:
                tool_result = {
                    'status': 'error',
                    'code': 'ui_component',
                    'message': f'{tool_function_name}: Client error {tool_result.status_code} from embedded UI result.',
                }
            elif 500 <= tool_result.status_code < 600:
                tool_result = {
                    'status': 'error',
                    'code': 'ui_component',
                    'message': f'{tool_function_name}: Server error {tool_result.status_code} from embedded UI result.',
                }
            else:
                tool_result = {
                    'status': 'error',
                    'code': 'ui_component',
                    'message': f'{tool_function_name}: Unexpected status code {tool_result.status_code} from embedded UI result.',
                }
        else:
            tool_result = tool_result.body.decode('utf-8', 'replace')

    elif tool_type in EXTERNAL_TOOL_TYPES and isinstance(tool_result, tuple):
        tool_result, tool_response_headers = tool_result

        try:
            if not isinstance(tool_response_headers, dict):
                tool_response_headers = dict(tool_response_headers)
        except Exception as e:
            tool_response_headers = {}
            log.debug(e)

        if tool_response_headers and isinstance(tool_response_headers, dict):
            content_disposition = tool_response_headers.get(
                'Content-Disposition',
                tool_response_headers.get('content-disposition', ''),
            )

            if 'inline' in content_disposition:
                content_type = tool_response_headers.get(
                    'Content-Type',
                    tool_response_headers.get('content-type', ''),
                )
                location = tool_response_headers.get(
                    'Location',
                    tool_response_headers.get('location', ''),
                )

                if 'text/html' in content_type:
                    # Support (html_content, result_context) nested tuple
                    result_context = None
                    html_content = tool_result
                    if isinstance(tool_result, (tuple, list)) and len(tool_result) == 2:
                        html_content, result_context = tool_result

                    # Display as iframe embed
                    tool_result_embeds.append(html_content)
                    if result_context is not None and isinstance(result_context, (str, dict, list)):
                        tool_result = result_context
                    else:
                        tool_result = {
                            'status': 'success',
                            'code': 'ui_component',
                            'message': f'{tool_function_name}: Embedded UI result is active and visible to the user.',
                        }
                elif location:
                    # Support (html_content, result_context) nested tuple for location embeds
                    result_context = None
                    if isinstance(tool_result, (tuple, list)) and len(tool_result) == 2:
                        _, result_context = tool_result

                    tool_result_embeds.append(location)
                    if result_context is not None and isinstance(result_context, (str, dict, list)):
                        tool_result = result_context
                    else:
                        tool_result = {
                            'status': 'success',
                            'code': 'ui_component',
                            'message': f'{tool_function_name}: Embedded UI result is active and visible to the user.',
                        }

    tool_result_files = []

    # Detect base64 image data URIs from tool results (e.g. binary image
    # responses from execute_tool_server).  Move the data URI to
    # tool_result_files and replace tool_result with a text summary.
    if isinstance(tool_result, str) and tool_result.startswith('data:image/'):
        tool_result_files.append({'type': 'image', 'url': tool_result})
        tool_result = f'{tool_function_name}: Image file read successfully.'

    if isinstance(tool_result, list):
        if tool_type == 'mcp':  # MCP
            tool_response = []
            for item in tool_result:
                if isinstance(item, dict):
                    if item.get('type') == 'text':
                        text = item.get('text', '')
                        if isinstance(text, str):
                            try:
                                text = JSONCodec.loads(text)
                            except JSONCodec.JSONDecodeError:
                                pass
                        tool_response.append(text)
                    elif item.get('type') in ['image', 'audio']:
                        file_url = await get_file_url_from_base64(
                            request,
                            f'data:{item.get("mimeType")};base64,{item.get("data", item.get("blob", ""))}',
                            {
                                'chat_id': metadata.get('chat_id', None),
                                'message_id': metadata.get('message_id', None),
                                'session_id': metadata.get('session_id', None),
                            },
                            user,
                        )

                        tool_result_files.append(
                            {
                                'type': item.get('type', 'data'),
                                'url': file_url,
                            }
                        )
                    elif item.get('type') == 'resource':
                        resource = item.get('resource', {})
                        text = resource.get('text', '')
                        if isinstance(text, str) and text:
                            try:
                                text = JSONCodec.loads(text)
                            except JSONCodec.JSONDecodeError:
                                pass
                            tool_response.append(text)
                        elif resource.get('blob'):
                            resource_mime_type = resource.get('mimeType') or 'application/octet-stream'
                            resource_blob = resource.get('blob', '')
                            if resource_mime_type.startswith('image/'):
                                tool_result_files.append(
                                    {
                                        'type': 'image',
                                        'url': f'data:{resource_mime_type};base64,{resource_blob}',
                                    }
                                )
                            else:
                                resource_uri = resource.get('uri', 'resource')
                                tool_response.append(
                                    f'[Resource: {resource_uri}] (binary data, mimeType: {resource_mime_type})'
                                )
                        elif resource.get('uri'):
                            tool_response.append(resource.get('uri'))
            tool_result = tool_response[0] if len(tool_response) == 1 else tool_response
        else:  # OpenAPI
            # Images are left to extract_base64_images below, which attaches them so the model can see them.
            for item in list(tool_result):
                if isinstance(item, str) and item.startswith('data:') and not BASE64_IMAGE_DATA_URI_RE.fullmatch(item):
                    tool_result_files.append(
                        {
                            'type': 'data',
                            'content': item,
                        }
                    )
                    tool_result.remove(item)

    tool_result = extract_base64_images(tool_result, tool_result_files)

    if isinstance(tool_result, list):
        tool_result = {'results': tool_result}

    if isinstance(tool_result, dict) or isinstance(tool_result, list):
        tool_result = json.dumps(tool_result, indent=2, ensure_ascii=False)

    # Safety: ensure tool_result is always a string (or None) to prevent
    # downstream TypeError when concatenating (e.g. if an upstream callable
    # returned a tuple that was not unpacked by the branches above).
    if tool_result is not None and not isinstance(tool_result, str):
        if isinstance(tool_result, tuple):
            # execute_tool_server returns (data, headers); unpack the data part
            tool_result = json.dumps(tool_result[0], indent=2, ensure_ascii=False) if len(tool_result) > 0 else ''
        else:
            tool_result = str(tool_result)

    return tool_result, tool_result_files, tool_result_embeds


def parse_terminal_tool_result(tool_result: Any) -> dict:
    try:
        result = JSONCodec.loads(tool_result)
    except (JSONCodec.JSONDecodeError, TypeError):
        return {}
    return result if isinstance(result, dict) else {}


async def terminal_event_handler(
    tool_function_name: str,
    tool_function_params: dict,
    tool_result,
    event_emitter,
):
    """Emit terminal:* events for Open Terminal tools.

    - display_file  → emits 'terminal:display_file' to open the file preview.
    - write_file / replace_file_content → emits 'terminal:write_file' to refresh.
    - run_command → emits 'terminal:run_command' with cwd to refresh if relevant.
    """
    if not event_emitter:
        return

    if tool_function_name == 'display_file':
        if tool_function_params.get('inline') is True:
            return
        result = parse_terminal_tool_result(tool_result)
        # Open Terminal resolves the argument against the session cwd, so prefer its resolved path
        path = result.get('path') or tool_function_params.get('path', '')
        if not path:
            return
        # Only open the viewer for existing files that are not displayed inline.
        if result.get('exists') is False or result.get('displayed') is True:
            return
        page = tool_function_params.get('page')

        await event_emitter(
            {
                'type': f'terminal:{tool_function_name}',
                'data': {
                    'path': path,
                    **({'page': page} if page else {}),
                },
            }
        )
    elif tool_function_name in ('write_file', 'replace_file_content'):
        result = parse_terminal_tool_result(tool_result)
        path = result.get('path') or tool_function_params.get('path', '')
        if not path:
            return
        await event_emitter(
            {
                'type': f'terminal:{tool_function_name}',
                'data': {'path': path},
            }
        )
    elif tool_function_name == 'run_command':
        await event_emitter(
            {
                'type': 'terminal:run_command',
                'data': {},
            }
        )


async def chat_completion_tools_handler(
    request: Request, body: dict, extra_params: dict, user: UserModel, models, tools
) -> tuple[dict, dict]:
    async def get_content_from_response(response) -> Optional[str]:
        content = None
        if hasattr(response, 'body_iterator'):
            async for chunk in response.body_iterator:
                data = JSONCodec.loads(chunk.decode('utf-8', 'replace'))
                content = data['choices'][0]['message']['content']

            # Cleanup any remaining background tasks if necessary
            if response.background is not None:
                await response.background()
        else:
            content = response['choices'][0]['message']['content']
        return content

    def get_tools_function_calling_payload(messages, task_model_id, content):
        user_message = get_last_user_message(messages)

        if user_message and messages and messages[-1]['role'] == 'user':
            # Remove the last user message to avoid duplication
            messages = messages[:-1]

        recent_messages = messages[-4:] if len(messages) > 4 else messages
        chat_history = '\n'.join(
            f'{message["role"].upper()}: """{get_content_from_message(message) or ""}"""' for message in recent_messages
        )

        prompt = f'History:\n{chat_history}\nQuery: {user_message}' if chat_history else f'Query: {user_message}'

        return {
            'model': task_model_id,
            'messages': [
                {'role': 'system', 'content': content},
                {'role': 'user', 'content': prompt},
            ],
            'stream': False,
            'metadata': {'task': str(TASKS.FUNCTION_CALLING)},
        }

    event_caller = extra_params['__event_call__']
    event_emitter = extra_params['__event_emitter__']
    metadata = extra_params['__metadata__']

    # One batched SELECT instead of four sequential round trips.
    task_config = await Config.get_many(
        'task.model.default',
        'task.model.external',
        'task.tools.prompt_template',
    )
    task_model_id = get_task_model_id(
        body['model'],
        task_config.get('task.model.default'),
        task_config.get('task.model.external'),
        models,
    )

    skip_files = False
    sources = []

    specs = [tool['spec'] for tool in tools.values()]
    tools_specs = JSONCodec.dumps(specs, ensure_ascii=False)

    tools_prompt_template = task_config.get('task.tools.prompt_template')
    if tools_prompt_template != '':
        template = tools_prompt_template
    else:
        template = DEFAULT_TOOLS_FUNCTION_CALLING_PROMPT_TEMPLATE

    tools_function_calling_prompt = tools_function_calling_generation_template(template, tools_specs)
    payload = get_tools_function_calling_payload(body['messages'], task_model_id, tools_function_calling_prompt)

    try:
        response = await generate_chat_completion(request, form_data=payload, user=user)
        log.debug('response=%r', response)
        content = await get_content_from_response(response)
        log.debug('content=%r', content)

        if not content:
            return body, {}

        try:
            content = content[content.find('{') : content.rfind('}') + 1]
            if not content:
                raise Exception('No JSON object found in the response')

            result = JSONCodec.loads(content)

            async def tool_call_handler(tool_call):
                nonlocal skip_files

                log.debug('tool_call=%r', tool_call)

                tool_function_name = tool_call.get('name', None)
                if tool_function_name not in tools:
                    log.warning(f'Tool "{tool_function_name}" not found')
                    return

                tool_function_params = tool_call.get('parameters', {})

                tool = None
                tool_type = ''

                try:
                    tool_function_params, tool_result, tool, tool_type = await execute_tool_call(
                        body,
                        {**metadata, 'tools': tools},
                        event_caller,
                        {'name': tool_function_name, 'arguments': tool_function_params, 'call_id': str(uuid4())},
                    )

                except Exception as e:
                    tool_result = {'error': str(e)}

                tool_result, tool_result_files, tool_result_embeds = await process_tool_result(
                    request,
                    tool_function_name,
                    tool_result,
                    tool_type,
                    metadata,
                    user,
                )

                if event_emitter:
                    await terminal_event_handler(
                        tool_function_name,
                        tool_function_params,
                        tool_result,
                        event_emitter,
                    )

                    if tool_result_files:
                        for file_item in tool_result_files:
                            if file_item.get('type') == 'image':
                                file_item['url'] = await store_tool_result_image(
                                    request, file_item.get('url'), metadata, user
                                )

                        await event_emitter(
                            {
                                'type': 'files',
                                'data': {
                                    'files': tool_result_files,
                                },
                            }
                        )

                    if tool_result_embeds:
                        await event_emitter(
                            {
                                'type': 'embeds',
                                'data': {
                                    'embeds': tool_result_embeds,
                                },
                            }
                        )

                if tool_result:
                    tool = tools[tool_function_name]
                    tool_id = tool.get('tool_id', '')

                    tool_name = f'{tool_id}/{tool_function_name}' if tool_id else f'{tool_function_name}'

                    # Citation is enabled for this tool
                    sources.append(
                        {
                            'source': {
                                'name': (f'{tool_name}'),
                            },
                            'document': [str(tool_result)],
                            'metadata': [
                                {
                                    'source': (f'{tool_name}'),
                                    'parameters': tool_function_params,
                                }
                            ],
                            'tool_result': True,
                        }
                    )

                    if tools[tool_function_name].get('metadata', {}).get('file_handler', False):
                        skip_files = True

            # check if "tool_calls" in result
            if result.get('tool_calls'):
                for tool_call in result.get('tool_calls'):
                    await tool_call_handler(tool_call)
            else:
                await tool_call_handler(result)

        except Exception as e:
            log.debug('Error: %s', e)
            content = None
    except Exception as e:
        log.debug('Error: %s', e)
        content = None

    log.debug('tool_contexts: %s', sources)

    if skip_files and 'files' in body.get('metadata', {}):
        del body['metadata']['files']

    return body, {'sources': sources}


async def chat_web_search_handler(request: Request, form_data: dict, extra_params: dict, user):
    event_emitter = extra_params['__event_emitter__']
    await event_emitter(
        {
            'type': 'status',
            'data': {
                'action': 'web_search',
                'description': 'Searching the web',
                'done': False,
            },
        }
    )

    messages = form_data['messages']
    user_message = get_last_user_message(messages)

    queries = []
    try:
        res = await generate_queries(
            request,
            {
                'model': form_data['model'],
                'messages': messages,
                'prompt': user_message,
                'type': 'web_search',
                'chat_id': extra_params.get('__chat_id__'),
            },
            user,
        )

        # generate_queries returns a JSONResponse on error (e.g. model not
        # found, chat completion failure).  Extract the error detail and
        # re-raise so the outer except block falls back to using the raw
        # user message as the search query.
        if isinstance(res, JSONResponse):
            try:
                error_body = JSONCodec.loads(res.body)
                detail = error_body.get('detail', 'Query generation failed')
            except Exception:
                detail = 'Query generation failed'
            raise Exception(detail)

        response = res['choices'][0]['message']['content']

        try:
            bracket_start = response.rfind('{')
            bracket_end = response.rfind('}') + 1

            if bracket_start == -1 or bracket_end == -1:
                raise Exception('No JSON object found in the response')

            response = response[bracket_start:bracket_end]
            queries = JSONCodec.loads(response)
            queries = queries.get('queries', [])
        except Exception as e:
            queries = [response]

        if ENABLE_QUERIES_CACHE:
            request.state.cached_queries = queries

    except Exception as e:
        log.exception(e)
        queries = [user_message or '']

    # Check if generated queries are empty
    if len(queries) == 1 and queries[0].strip() == '':
        queries = [user_message or '']

    # Check if queries are not found
    if len(queries) == 0:
        await event_emitter(
            {
                'type': 'status',
                'data': {
                    'action': 'web_search',
                    'description': 'No search query generated',
                    'done': True,
                },
            }
        )
        return form_data

    await event_emitter(
        {
            'type': 'status',
            'data': {
                'action': 'web_search_queries_generated',
                'queries': queries,
                'done': False,
            },
        }
    )

    try:
        results = await process_web_search(
            request,
            SearchForm(queries=queries),
            user=user,
        )

        if results:
            files = form_data.get('files', [])

            if results.get('collection_names'):
                for col_idx, collection_name in enumerate(results.get('collection_names')):
                    files.append(
                        {
                            'collection_name': collection_name,
                            'name': ', '.join(queries),
                            'type': 'web_search',
                            'urls': results['filenames'],
                            'queries': queries,
                        }
                    )
            elif results.get('docs'):
                # Invoked when bypass embedding and retrieval is set to True
                docs = results['docs']
                files.append(
                    {
                        'docs': docs,
                        'name': ', '.join(queries),
                        'type': 'web_search',
                        'urls': results['filenames'],
                        'queries': queries,
                    }
                )

            form_data['files'] = files

            await event_emitter(
                {
                    'type': 'status',
                    'data': {
                        'action': 'web_search',
                        'description': 'Searched {{count}} sites',
                        'urls': results['filenames'],
                        'items': results.get('items', []),
                        'done': True,
                    },
                }
            )
        else:
            await event_emitter(
                {
                    'type': 'status',
                    'data': {
                        'action': 'web_search',
                        'description': 'No search results found',
                        'done': True,
                        'error': True,
                    },
                }
            )

    except Exception as e:
        log.exception(e)
        detail = e.detail if isinstance(e, HTTPException) else None
        await event_emitter(
            {
                'type': 'status',
                'data': {
                    'action': 'web_search',
                    'description': (str(detail) if detail else 'An error occurred while searching the web'),
                    'queries': queries,
                    'done': True,
                    'error': True,
                },
            }
        )

    return form_data


def get_images_from_messages(message_list):
    images = []

    for message in reversed(message_list):
        message_images = []
        for file in message.get('files', []):
            if file.get('type') == 'image':
                message_images.append(file.get('url'))
            elif is_raster_image_content_type(file.get('content_type')):
                message_images.append(file.get('url'))

        if message_images:
            images.append(message_images)

    return images


async def get_image_urls(delta_images, request, metadata, user) -> list[str]:
    if not isinstance(delta_images, list):
        return []

    image_urls = []
    for img in delta_images:
        if not isinstance(img, dict) or img.get('type') != 'image_url':
            continue

        url = img.get('image_url', {}).get('url')
        if not url:
            continue

        if re.match(r'data:image/\w+;base64', url):
            url = await get_image_url_from_base64(request, url, metadata, user)

        image_urls.append(url)

    return image_urls


def add_file_context(messages: list) -> list:
    """
    Replay attachment references beside their original user messages for native tools.
    """

    def format_file_tag(file):
        attrs = {'type': file.get('type', 'file'), 'id': file.get('id') or file.get('url')}
        attrs.update({key: file[key] for key in ('url', 'content_type', 'name') if file.get(key)})
        return (
            '<file ' + ' '.join(f'{key}="{html.escape(str(value), quote=True)}"' for key, value in attrs.items()) + '/>'
        )

    for message in messages:
        if message.get('role') != 'user':
            continue
        # Notes, collections, and chats can be addressed by ID without a URL.
        attached_files = [
            file
            for file in message.get('files', [])
            if isinstance(file, dict)
            and (
                (isinstance(file.get('url'), str) and file['url'] and not file['url'].startswith('data:'))
                or (file.get('type') in ('chat', 'note', 'collection') and file.get('id'))
            )
        ]
        if not attached_files:
            continue

        file_tags = [format_file_tag(file) for file in attached_files]
        file_context = '<attached_files>\n' + '\n'.join(file_tags) + '\n</attached_files>\n\n'

        content = message.get('content', '')
        if isinstance(content, list):
            message['content'] = [{'type': 'text', 'text': file_context}] + content
        else:
            message['content'] = file_context + content

    return messages


async def chat_image_generation_handler(request: Request, form_data: dict, extra_params: dict, user):
    metadata = extra_params.get('__metadata__', {})
    chat_id = metadata.get('chat_id', None)
    __event_emitter__ = extra_params.get('__event_emitter__', None)

    if not chat_id or not isinstance(chat_id, str) or not __event_emitter__:
        return form_data

    is_channel_chat = chat_id.startswith('channel:')
    image_metadata = {
        'message_id': metadata.get('message_id', None),
        **({'channel_id': chat_id.removeprefix('channel:')} if is_channel_chat else {'chat_id': chat_id}),
    }
    terminal_metadata = {**metadata, 'terminal_id': form_data.get('terminal_id')}

    if not is_saved_chat_id(chat_id):
        message_list = form_data.get('messages', [])
    else:
        chat = await Chats.get_accessible_chat_by_id(chat_id, user, permission='write')

        messages_map = chat.chat.get('history', {}).get('messages', {})
        message_id = metadata.get('user_message_id') or chat.chat.get('history', {}).get('currentId')
        message_list = get_message_list(messages_map, message_id)

    user_message = get_last_user_message(message_list)

    prompt = user_message
    message_images = get_images_from_messages(message_list)

    # Limit to first 2 sets of images
    # We may want to change this in the future to allow more images
    input_images = []
    for idx, images in enumerate(message_images):
        if idx >= 2:
            break
        for image in images:
            input_images.append(image)

    # Called directly, bypassing the /images routes that enforce these switches.
    editing = len(input_images) > 0 and await Config.get('images.edit.enable')
    if not editing and not await Config.get('image_generation.enable'):
        return form_data

    if is_saved_chat_id(chat_id):
        await __event_emitter__(
            {
                'type': 'status',
                'data': {'description': 'Creating image', 'done': False},
            }
        )

    system_message_content = ''

    if editing:
        # Edit image(s)
        try:
            images = await image_edits(
                request=request,
                form_data=EditImageForm(**{'prompt': prompt, 'image': input_images}),
                metadata=image_metadata,
                user=user,
            )
            await upload_files_to_terminal(request, user, images, terminal_metadata, extra_params)

            await __event_emitter__(
                {
                    'type': 'status',
                    'data': {'description': 'Image created', 'done': True},
                }
            )

            await __event_emitter__(
                {
                    'type': 'files',
                    'data': {
                        'files': [
                            {
                                'type': 'image',
                                **image,
                            }
                            for image in images
                        ]
                    },
                }
            )

            system_message_content = (
                '<context>The edited images are shown in chat. Use terminal_path with terminal tools or '
                f'display_file when available. Images: {JSONCodec.dumps(images)}</context>'
            )
        except Exception as e:
            log.debug(e)

            error_message = ''
            if isinstance(e, HTTPException):
                if e.detail and isinstance(e.detail, dict):
                    error_message = e.detail.get('message', str(e.detail))
                else:
                    error_message = str(e.detail)

            await __event_emitter__(
                {
                    'type': 'status',
                    'data': {
                        'description': f'An error occurred while generating an image',
                        'done': True,
                    },
                }
            )

            system_message_content = f'<context>Image generation was attempted but failed. The system is currently unable to generate the image. Tell the user that the following error occurred: {error_message}</context>'

    elif not await Config.get('image_generation.enable'):
        await __event_emitter__(
            {
                'type': 'status',
                'data': {
                    'description': 'Image generation is disabled',
                    'done': True,
                },
            }
        )

        system_message_content = '<context>Image generation was requested but the feature is currently disabled by the administrator, so no image was created. Let the user know that image generation is currently unavailable.</context>'

    else:
        # Create image(s)
        if await Config.get('image_generation.prompt.enable'):
            try:
                res = await generate_image_prompt(
                    request,
                    {
                        'model': form_data['model'],
                        'messages': form_data['messages'],
                        'chat_id': metadata.get('chat_id'),
                    },
                    user,
                )

                # Handle JSONResponse from error paths
                if isinstance(res, JSONResponse):
                    try:
                        error_body = JSONCodec.loads(res.body)
                        detail = error_body.get('detail', 'Image prompt generation failed')
                    except Exception:
                        detail = 'Image prompt generation failed'
                    raise Exception(detail)

                response = res['choices'][0]['message']['content']

                try:
                    bracket_start = response.rfind('{')
                    bracket_end = response.rfind('}') + 1

                    if bracket_start == -1 or bracket_end == -1:
                        raise Exception('No JSON object found in the response')

                    response = response[bracket_start:bracket_end]
                    response = JSONCodec.loads(response)
                    prompt = response.get('prompt', [])
                except Exception as e:
                    prompt = user_message

            except Exception as e:
                log.exception(e)
                prompt = user_message

        try:
            images = await image_generations(
                request=request,
                form_data=CreateImageForm(**{'prompt': prompt}),
                metadata=image_metadata,
                user=user,
            )
            await upload_files_to_terminal(request, user, images, terminal_metadata, extra_params)

            await __event_emitter__(
                {
                    'type': 'status',
                    'data': {'description': 'Image created', 'done': True},
                }
            )

            await __event_emitter__(
                {
                    'type': 'files',
                    'data': {
                        'files': [
                            {
                                'type': 'image',
                                **image,
                            }
                            for image in images
                        ]
                    },
                }
            )

            system_message_content = (
                '<context>The generated images are shown in chat. Use terminal_path with terminal tools or '
                f'display_file when available. Images: {JSONCodec.dumps(images)}</context>'
            )
        except Exception as e:
            log.debug(e)

            error_message = ''
            if isinstance(e, HTTPException):
                if e.detail and isinstance(e.detail, dict):
                    error_message = e.detail.get('message', str(e.detail))
                else:
                    error_message = str(e.detail)

            await __event_emitter__(
                {
                    'type': 'status',
                    'data': {
                        'description': f'An error occurred while generating an image',
                        'done': True,
                    },
                }
            )

            system_message_content = f'<context>Image generation was attempted but failed because of an error. The system is currently unable to generate the image. Tell the user that the following error occurred: {error_message}</context>'

    if system_message_content:
        form_data['messages'] = add_or_update_system_message(system_message_content, form_data['messages'])

    return form_data


async def chat_completion_files_handler(
    request: Request, body: dict, extra_params: dict, user: UserModel
) -> tuple[dict, dict[str, list]]:
    __event_emitter__ = extra_params['__event_emitter__']
    sources = []

    files = [item for item in (body.get('metadata', {}).get('files', None) or []) if item.get('type') != 'filesystem']
    if files:
        # Check if all files are in full context mode
        all_full_context = all(item.get('context') == 'full' for item in files)

        queries = []
        if not all_full_context:
            try:
                queries_response = await generate_queries(
                    request,
                    {
                        'model': body['model'],
                        'messages': body['messages'],
                        'type': 'retrieval',
                        'chat_id': body.get('metadata', {}).get('chat_id'),
                    },
                    user,
                )
                queries_response = queries_response['choices'][0]['message']['content']

                try:
                    bracket_start = queries_response.rfind('{')
                    bracket_end = queries_response.rfind('}') + 1

                    if bracket_start == -1 or bracket_end == -1:
                        raise Exception('No JSON object found in the response')

                    queries_response = queries_response[bracket_start:bracket_end]
                    queries_response = JSONCodec.loads(queries_response)
                except Exception as e:
                    queries_response = {'queries': [queries_response]}

                queries = queries_response.get('queries', [])
            except Exception:
                pass

            await __event_emitter__(
                {
                    'type': 'status',
                    'data': {
                        'action': 'queries_generated',
                        'queries': queries,
                        'done': False,
                    },
                }
            )

        if len(queries) == 0:
            queries = [get_last_user_message(body['messages']) or '']

        try:
            # One batched SELECT instead of six sequential round trips.
            rag_config = await Config.get_many(
                'rag.top_k',
                'rag.top_k_reranker',
                'rag.relevance_threshold',
                'rag.hybrid_bm25_weight',
                'rag.enable_hybrid_search',
                'rag.full_context',
            )
            # Directly await async get_sources_from_items (no thread needed - fully async now)
            sources = await get_sources_from_items(
                request=request,
                items=files,
                queries=queries,
                embedding_function=lambda query, prefix: request.app.state.EMBEDDING_FUNCTION(
                    query, prefix=prefix, user=user
                ),
                k=rag_config.get('rag.top_k'),
                reranking_function=(
                    (lambda query, documents: request.app.state.RERANKING_FUNCTION(query, documents, user=user))
                    if request.app.state.RERANKING_FUNCTION
                    else None
                ),
                k_reranker=rag_config.get('rag.top_k_reranker'),
                r=rag_config.get('rag.relevance_threshold'),
                hybrid_bm25_weight=rag_config.get('rag.hybrid_bm25_weight'),
                hybrid_search=rag_config.get('rag.enable_hybrid_search'),
                full_context=all_full_context or rag_config.get('rag.full_context'),
                user=user,
            )
        except Exception as e:
            log.exception(e)

        log.debug('rag_contexts:sources: %s', sources)

        unique_ids = set()
        for source in sources or []:
            if not source or len(source.keys()) == 0:
                continue

            documents = source.get('document') or []
            metadatas = source.get('metadata') or []
            src_info = source.get('source') or {}

            for index, _ in enumerate(documents):
                metadata = metadatas[index] if index < len(metadatas) else None
                _id = (metadata or {}).get('source') or (src_info or {}).get('id') or 'N/A'
                unique_ids.add(_id)

        sources_count = len(unique_ids)
        await __event_emitter__(
            {
                'type': 'status',
                'data': {
                    'action': 'sources_retrieved',
                    'count': sources_count,
                    'done': True,
                },
            }
        )

    return body, {'sources': sources}


async def convert_url_images_to_base64(form_data, user=None):
    messages = form_data.get('messages', [])

    for message in messages:
        content = message.get('content')
        if not isinstance(content, list):
            continue

        new_content = []

        for item in content:
            if not isinstance(item, dict) or item.get('type') not in ('image_url', 'input_image'):
                new_content.append(item)
                continue

            image_url_data = item.get('image_url', {})
            if isinstance(image_url_data, dict):
                image_url = image_url_data.get('url') or ''
            elif isinstance(image_url_data, str):
                image_url = image_url_data
            else:
                image_url = ''
            if image_url.startswith('data:image/'):
                new_content.append(item)
                continue

            try:
                base64_data = await get_image_base64_from_url(image_url, user=user)
                if base64_data and isinstance(image_url_data, str):
                    new_content.append({**item, 'image_url': base64_data})
                elif base64_data:
                    image_url_payload = {'url': base64_data}
                    if isinstance(image_url_data, dict) and image_url_data.get('detail'):
                        image_url_payload['detail'] = image_url_data['detail']
                    new_content.append(
                        {
                            'type': item['type'],
                            'image_url': image_url_payload,
                        }
                    )
                else:
                    new_content.append(item)
            except Exception as e:
                log.debug('Error converting image URL to base64: %s', e)
                new_content.append(item)

        message['content'] = new_content

    return form_data


MESSAGE_REPLAY_KEYS = ('id', 'role', 'content', 'output', 'files', 'contextSummary', 'usage', 'model', 'meta')


async def load_messages_from_db(chat_id: str, message_id: str) -> Optional[list[dict]]:
    """
    Load the message chain from DB up to message_id,
    keeping only fields needed to rebuild the LLM payload.
    """
    messages_map = await Chats.get_messages_map_by_chat_id(chat_id)
    if not messages_map:
        return None

    db_messages = get_message_list(messages_map, message_id)
    if not db_messages:
        return None

    for msg in db_messages:
        skills_create = (msg.get('meta') or {}).get('skills_create')
        # Replay the saved expansion unless the user edited the message.
        if skills_create and skills_create.get('original_text') == get_content_from_message(msg):
            set_last_user_message_content(skills_create['expanded_text'], [msg])

    return [
        {k: v for k, v in msg.items() if k in MESSAGE_REPLAY_KEYS}
        for msg in db_messages
        if not (msg.get('role') == 'assistant' and not msg.get('content') and not msg.get('output'))
    ]


def get_reasoning_format(model: dict) -> str | None:
    """
    Determine how reasoning should be included in reconstructed messages.

    Returns:
        'thinking': Ollama expects reasoning in the native thinking field.
        'think_tags': wrap reasoning in <think> tags inside content.
        'reasoning_content': llama.cpp supports reasoning_content as a top-level field.
        None: skip reasoning (safe default for strict providers).
    """
    provider = model.get('provider', '')
    if model.get('owned_by') == 'ollama':
        return 'thinking'
    if provider == 'llama.cpp':
        return 'reasoning_content'
    return None


def strip_reasoning_details(output: list) -> list:
    return [
        {key: value for key, value in item.items() if key != 'reasoning_details'} if isinstance(item, dict) else item
        for item in output
    ]


def strip_tool_images(output: list, model: dict) -> list:
    """Replace tool result images with a text note when the model has Vision turned off."""
    if (model.get('info', {}).get('meta', {}).get('capabilities') or {}).get('vision', True):
        return output
    stripped_output = []
    for item in output:
        if isinstance(item, dict) and item.get('type') == 'function_call_output':
            parts = [
                {'type': 'input_text', 'text': '\n[image]'} if part.get('type') == 'input_image' else part
                for part in item.get('output', [])
            ]
            item = {**item, 'output': parts}
        stripped_output.append(item)
    return stripped_output


def process_messages_with_output(
    messages: list[dict],
    reasoning_format: str | None = None,
    include_file_context: bool = False,
) -> list[dict]:
    """
    Process messages with OR-aligned output items for LLM consumption.

    For assistant messages with 'output' field, produces properly formatted
    OpenAI-style messages (tool_calls + tool results). Strips 'output' before LLM.
    """
    processed = []

    for message in messages:
        meta = message.get('meta')
        voice = meta.get('voice') if isinstance(meta, dict) else None
        voice = voice if isinstance(voice, dict) else {}
        spoken = (
            message.get('role') == 'assistant'
            and voice
            and message.get('model') == voice.get('model')
            and not message.get('output')
        )
        # Keep voice speech available for follow-ups without attributing its status claims
        # to the reasoning model. A completed chat answer can predate a stale spoken reply.
        transcripts = []
        if message.get('role') == 'assistant':
            transcripts = voice.get('speech') or []
        if spoken:
            transcripts = [{'transcript': message.get('content', '')}]
        speech = [
            {
                'role': 'assistant',
                'content': '[Historical voice assistant transcript; not current task status]\n' + item['transcript'],
            }
            for item in transcripts
            if isinstance(item, dict) and isinstance(item.get('transcript'), str) and item['transcript']
        ]
        if spoken:
            processed.extend(speech)
            continue
        if message.get('role') == 'assistant' and message.get('output'):
            # Use output items for clean OpenAI-format messages
            output_messages = convert_output_to_messages(
                message['output'],
                raw=True,
                reasoning_format=reasoning_format,
                flatten_tool_images=True,
            )
            if output_messages:
                processed.extend(output_messages)
                processed.extend(speech)
                continue
            if not message.get('content'):
                continue

        clean_message = dict(message)
        for key in ('id', 'output', 'model', 'contextSummary', 'context_summary', 'usage', 'meta'):
            clean_message.pop(key, None)
        processed.append(clean_message)
        processed.extend(speech)

    if include_file_context:
        add_file_context(processed)
    for message in processed:
        message.pop('files', None)

    return processed


def sanitize_tool_pairs(messages: list[dict]) -> list[dict]:
    paired_ids_by_message = get_paired_tool_call_ids(messages)

    sanitized = []
    for message, paired_ids in zip(messages, paired_ids_by_message):
        if message.get('role') == 'assistant' and message.get('tool_calls'):
            kept = [tool_call for tool_call in message.get('tool_calls') or [] if tool_call.get('id') in paired_ids]
            if kept:
                sanitized.append({**message, 'tool_calls': kept})
            else:
                clean = dict(message)
                clean.pop('tool_calls', None)
                clean.pop('reasoning_items', None)
                if clean.get('content'):
                    sanitized.append(clean)
        elif message.get('role') != 'tool' or message.get('tool_call_id') in paired_ids:
            sanitized.append(message)

    return sanitized


async def process_chat_payload(request, form_data, user, metadata, model):
    # Ensure chat_id is always a string — external API clients may omit it.
    if not isinstance(metadata.get('chat_id'), str):
        metadata['chat_id'] = ''

    # Pipeline Inlet -> Filter Inlet -> Chat Memory -> Chat Web Search -> Chat Image Generation
    # -> Chat Code Interpreter (Form Data Update) -> (Default) Chat Tools Function Calling
    # -> Chat Files

    # Arena model resolution — pick the sub-model now so all downstream
    # processing (knowledge, capabilities, tools, params) uses its settings
    # instead of the empty arena wrapper.
    if model.get('owned_by') == 'arena':
        arena_model_ids = model.get('info', {}).get('meta', {}).get('model_ids')
        arena_filter_mode = model.get('info', {}).get('meta', {}).get('filter_mode')
        if arena_model_ids and arena_filter_mode == 'exclude':
            arena_model_ids = [
                available_model['id']
                for available_model in request.app.state.MODELS.values()
                if available_model.get('owned_by') != 'arena' and available_model['id'] not in arena_model_ids
            ]

        if isinstance(arena_model_ids, list) and arena_model_ids:
            selected_model_id = random.choice(arena_model_ids)
        else:
            arena_model_ids = [
                available_model['id']
                for available_model in request.app.state.MODELS.values()
                if available_model.get('owned_by') != 'arena'
            ]
            selected_model_id = random.choice(arena_model_ids)

        selected_model = request.app.state.MODELS.get(selected_model_id)
        if selected_model:
            model = selected_model
            form_data['model'] = selected_model_id
            metadata['selected_model_id'] = selected_model_id

    payload_tools = form_data.get('tools', None)  # snapshot before filters
    form_data = apply_params_to_form_data(form_data, model)
    log.debug('form_data: %s', form_data)

    # Guided regeneration: extract before it reaches the LLM provider
    regeneration_prompt = form_data.pop('regeneration_prompt', None)

    # Load messages from DB when available — DB preserves structured 'output' items
    # which the frontend strips, causing tool calls to be merged into content.
    chat_id = metadata.get('chat_id')
    user_message_id = metadata.get('user_message_id')
    pending_assistant_message = None
    chat = await Chats.get_chat_by_id(chat_id) if is_saved_chat_id(chat_id) else None
    is_note_chat = bool(chat and (chat.meta or {}).get('internal') is True and (chat.meta or {}).get('type') == 'note')
    use_builtin_tools = is_note_chat 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)
    )
    metadata['include_file_context'] = bool(
        chat and chat.user_id == user.id and payload_tools is None and use_builtin_tools
    )

    if is_saved_chat_id(chat_id) and user_message_id:
        db_messages = await load_messages_from_db(chat_id, user_message_id)
        if db_messages:
            # Continue: frontend sends assistant_message_id when continuing
            # an existing response. Load its content so the LLM sees prior output.
            assistant_message_id = metadata.get('assistant_message_id')
            if assistant_message_id:
                assistant_message = await Chats.get_message_by_id_and_message_id(chat_id, assistant_message_id)
                if assistant_message and (assistant_message.get('content') or assistant_message.get('output')):
                    output = assistant_message.get('output') or []
                    if (
                        metadata.get('tool_approval_resume') != 'model'
                        and not assistant_message.get('done', True)
                        and output
                        and output[-1].get('type') == 'message'
                        and output[-1].get('status') == 'in_progress'
                        and any(item.get('type') == 'function_call_output' for item in output)
                        and not pending_tool_calls(output)
                    ):
                        return form_data, metadata, [], True
                    assistant_message = {k: v for k, v in assistant_message.items() if k in MESSAGE_REPLAY_KEYS}
                    db_messages.append(assistant_message)
                    output = assistant_message.get('output')
                    output = output if isinstance(output, list) else []
                    if metadata.get('tool_approval_resume') == 'tools' or pending_tool_calls(output):
                        pending_assistant_message = assistant_message

            # Guided regeneration: show the model the reply it is asked to revise.
            regeneration_message_id = metadata.get('regeneration_message_id')
            if regeneration_message_id:
                regenerated_message = await Chats.get_message_by_id_and_message_id(chat_id, regeneration_message_id)
                if regenerated_message and (regenerated_message.get('content') or regenerated_message.get('output')):
                    db_messages.append({k: v for k, v in regenerated_message.items() if k in MESSAGE_REPLAY_KEYS})

            system_message = get_system_message(form_data.get('messages', []))
            form_data['messages'] = [system_message, *db_messages] if system_message else db_messages

            # Inject image files into content as image_url parts (mirrors frontend logic)
            for message in form_data['messages']:
                image_files = [
                    f
                    for f in message.get('files', [])
                    if f.get('type') == 'image' or is_raster_image_content_type(f.get('content_type'))
                ]
                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': f['url']},
                                }
                                for f in image_files
                                if f.get('url')
                            ],
                        ]

    if regeneration_prompt:
        form_data['messages'].append({'role': 'user', 'content': regeneration_prompt})

    filter_functions = (
        await get_filter_functions(request, model, metadata.get('filter_ids', [])) if ENABLE_FUNCTIONS else []
    )
    compaction_messages = None
    context_summary = None
    if (is_saved_chat_id(chat_id) or is_temporary_chat_id(chat_id)) and user_message_id:
        system_message = get_system_message(form_data.get('messages', []))
        chat_system_prompt = get_content_from_message(system_message) if system_message else ''

        try:
            # Filters and resumed tool calls must still receive compacted history.
            if (
                filter_functions
                or 'pipeline' in model
                or model.get('pipe')
                or get_sorted_filters(form_data['model'], request.app.state.MODELS)
                or pending_assistant_message is not None
                or metadata.get('params', {}).get('function_calling') == 'legacy'
            ):
                form_data['messages'], context_summary, _ = await compact_messages_for_request(
                    request,
                    None,
                    user,
                    form_data.get('messages', []),
                    metadata,
                    form_data['model'],
                    request.app.state.MODELS,
                    chat_system_prompt,
                )
                if context_summary:
                    form_data['messages'] = add_or_update_system_message(
                        f'[CONVERSATION SUMMARY]\n{context_summary}', form_data['messages'], append=True
                    )
            else:
                compaction_messages = form_data.get('messages', [])
                form_data['messages'], context_summary = await apply_summary_checkpoint(compaction_messages)
        except Exception:
            compaction_messages = None
            log.exception('Context compaction failed; continuing with full chat history')

    # Compact the full chain, but prepare history separately from the response being resumed.
    if pending_assistant_message is not None:
        form_data['messages'] = [
            message for message in form_data['messages'] if message.get('id') != pending_assistant_message.get('id')
        ]

    # Process messages with OR-aligned output items for clean LLM messages
    for message in form_data.get('messages', []):
        output = message.get('output')
        # reasoning_details can be model/provider-bound, so only replay them
        # for output produced by the same model.
        if message.get('role') == 'assistant' and message.get('model') != model['id'] and isinstance(output, list):
            message['output'] = strip_reasoning_details(output)
        if message.get('role') == 'assistant' and isinstance(output, list):
            message['output'] = strip_tool_images(message['output'], model)

    form_data['messages'] = process_messages_with_output(
        form_data.get('messages', []),
        reasoning_format=get_reasoning_format(model),
        include_file_context=metadata['include_file_context'],
    )
    form_data['messages'] = sanitize_tool_pairs(form_data['messages'])
    history_messages = [message for message in form_data['messages'] if message.get('role') != 'system']

    system_message = get_system_message(form_data.get('messages', []))
    if system_message:  # Chat Controls/User Settings
        try:
            form_data = await apply_system_prompt_to_body(
                system_message.get('content'), form_data, metadata, user, replace=True
            )  # Required to handle system prompt variables
        except Exception:
            pass

    form_data = await convert_url_images_to_base64(form_data, user=user)

    event_emitter = await get_event_emitter(metadata)
    event_caller = await get_event_call(metadata)

    extra_params = {
        '__event_emitter__': event_emitter,
        '__event_call__': event_caller,
        '__user__': user.model_dump() if isinstance(user, UserModel) else {},
        '__metadata__': metadata,
        '__oauth_token__': await get_system_oauth_token(request, user),
        '__request__': request,
        '__model__': model,
        '__chat_id__': metadata.get('chat_id'),
        '__message_id__': metadata.get('message_id'),
    }
    # Initialize events to store additional event to be sent to the client
    # Initialize contexts and citation
    from open_webui.utils.models import add_tether_model

    models = await add_tether_model(request, user, form_data.get('model'))

    task_model_id = get_task_model_id(
        form_data['model'],
        await Config.get('task.model.default'),
        await Config.get('task.model.external'),
        models,
    )

    events = []
    sources = []

    # Folder "Project" handling
    # Check if the request has chat_id and is inside of a folder
    # Uses lightweight column query — only fetches folder_id, not the full chat JSON blob
    chat_id = metadata.get('chat_id', None)
    folder_id = None
    if user and is_saved_chat_id(chat_id):
        folder_id = await Chats.get_chat_folder_id(chat_id, user.id)

    # Fallback: use folder_id from metadata (temporary chats have no DB record)
    if not folder_id:
        folder_id = metadata.get('folder_id', None)

    if folder_id and user:
        folder = await Folders.get_folder_by_id(folder_id)
        if folder and user.role != 'admin' and not await has_folder_access(user.id, folder, 'read', db=None):
            folder = None

        if folder and folder.data:
            if 'system_prompt' in folder.data:
                form_data = await apply_system_prompt_to_body(folder.data['system_prompt'], form_data, metadata, user)
            if 'files' in folder.data:
                if metadata.get('params', {}).get('function_calling') == 'legacy':
                    form_data['files'] = [
                        {'type': 'folder', 'id': folder.id},
                        *form_data.get('files', []),
                    ]
                else:
                    # Native FC: skip RAG injection, builtin tools
                    # will read folder knowledge from metadata.
                    metadata['folder_knowledge'] = await get_owner_accessible_folder_files(folder)

    # Model "Knowledge" handling
    user_message = get_last_user_message(form_data['messages'])
    model_knowledge = model.get('info', {}).get('meta', {}).get('knowledge', False)

    if model_knowledge and metadata.get('params', {}).get('function_calling') == 'legacy':
        await event_emitter(
            {
                'type': 'status',
                'data': {
                    'action': 'knowledge_search',
                    'query': user_message,
                    'done': False,
                },
            }
        )

        knowledge_files = []
        for item in model_knowledge:
            if item.get('collection_name'):
                knowledge_files.append(
                    {
                        'id': item.get('collection_name'),
                        'name': item.get('name'),
                        'legacy': True,
                    }
                )
            elif item.get('collection_names'):
                knowledge_files.append(
                    {
                        'name': item.get('name'),
                        'type': 'collection',
                        'collection_names': item.get('collection_names'),
                        'legacy': True,
                    }
                )
            else:
                knowledge_files.append(item)

        files = form_data.get('files', [])
        files.extend(knowledge_files)
        form_data['files'] = files

    variables = form_data.pop('variables', None)

    # Process the form_data through the pipeline
    try:
        form_data = await process_pipeline_inlet_filter(request, form_data, user, models)
    except Exception as e:
        raise e

    filter_context = get_filter_context(request) if ENABLE_FUNCTIONS else None
    if ENABLE_FUNCTIONS:
        try:
            form_data, flags = await process_filter_functions(
                request=request,
                filter_context=filter_context,
                filter_functions=filter_functions,
                filter_type='inlet',
                form_data=form_data,
                extra_params=extra_params,
            )
        except Exception as e:
            raise Exception(f'{e}')

    initial_output = []
    features = form_data.pop('features', None) or {}
    extra_params['__features__'] = features
    if features:
        if 'voice' in features and features['voice']:
            if await Config.get('task.voice.prompt.enable'):
                template = await Config.get('task.voice.prompt_template')
                if not template:
                    template = DEFAULT_VOICE_MODE_PROMPT_TEMPLATE

                form_data['messages'] = add_or_update_system_message(
                    template,
                    form_data['messages'],
                )

        if (
            'memory' in features
            and features['memory']
            and await Config.get('memories.enable')
            and await Config.get('memories.system_context.enable')
        ):
            # features is client-supplied; re-check the permission the native FC path enforces.
            if getattr(user, 'role', None) == 'admin' or await has_permission(
                getattr(user, 'id', ''),
                'features.memories',
                await Config.get('user.permissions'),
            ):
                if use_builtin_tools:
                    if not metadata.get('assistant_message_id'):
                        initial_output.extend(await preload_memory(request, form_data['messages'], user, model))
                else:
                    form_data = await add_memory_context(request, form_data, user, model)

        if 'web_search' in features and features['web_search'] and await Config.get('web.search.enable'):
            # features is client-supplied; re-check the permission the native FC path enforces.
            if getattr(user, 'role', None) == 'admin' or await has_permission(
                getattr(user, 'id', ''),
                'features.web_search',
                await Config.get('user.permissions'),
            ):
                # Skip forced RAG web search when native FC is enabled - model can use web_search tool
                if metadata.get('params', {}).get('function_calling') == 'legacy':
                    form_data = await chat_web_search_handler(request, form_data, extra_params, user)

        if 'image_generation' in features and features['image_generation']:
            # features is client-supplied; re-check the permission the direct /images routes enforce.
            if getattr(user, 'role', None) == 'admin' or await has_permission(
                getattr(user, 'id', ''),
                'features.image_generation',
                await Config.get('user.permissions'),
            ):
                # Skip forced image generation when native FC is enabled - model can use generate_image tool
                if metadata.get('params', {}).get('function_calling') == 'legacy':
                    form_data = await chat_image_generation_handler(request, form_data, extra_params, user)

        if 'code_interpreter' in features and features['code_interpreter']:
            engine = await Config.get('code_interpreter.engine', 'pyodide')

            # Skip XML-tag prompt injection when native FC is enabled —
            # execute_code will be injected as a builtin tool instead
            if metadata.get('params', {}).get('function_calling') == 'legacy':
                ci_prompt_template = await Config.get('code_interpreter.prompt_template')
                prompt = ci_prompt_template if ci_prompt_template != '' else DEFAULT_CODE_INTERPRETER_PROMPT

                # Append filesystem awareness only for pyodide engine
                if engine != 'jupyter':
                    prompt += CODE_INTERPRETER_PYODIDE_PROMPT

                form_data['messages'] = add_or_update_user_message(
                    prompt,
                    form_data['messages'],
                )
            else:
                # Put dynamic Pyodide filesystem context in the system prompt to preserve prefix caching.
                if engine != 'jupyter':
                    form_data['messages'] = add_or_update_system_message(
                        CODE_INTERPRETER_PYODIDE_PROMPT,
                        form_data['messages'],
                        append=True,
                    )

    form_data['messages'] = set_tool_approval_prompt(form_data['messages'])
    tool_ids = form_data.pop('tool_ids', None)
    terminal_id = form_data.pop('terminal_id', None)
    if not ENABLE_TOOL_SERVERS:
        if terminal_id:
            log.debug('Excluded external plugins disabled by plugin configuration')
        terminal_id = None
    files = form_data.pop('files', None)
    form_data.pop('folder_id', None)
    metadata['terminal_id'] = terminal_id
    can_author_skills = (
        use_builtin_tools
        and (model.get('info', {}).get('meta', {}).get('builtinTools') or {}).get('skills', True)
        and (
            user.role == 'admin'
            or await has_permission(user.id, 'workspace.skills', await Config.get('user.permissions'))
        )
    )
    skill_authoring_allowed = can_author_skills and has_prior_user_message(form_data.get('messages', []))
    skill_create_denial_reason = 'empty_chat' if can_author_skills else 'disabled'
    if (
        apply_skills_create_prompt(
            form_data.get('messages', []),
            allowed=skill_authoring_allowed,
            denial_reason=skill_create_denial_reason,
        )
        and is_saved_chat_id(chat_id)
        and user_message_id
    ):
        stored_message = await Chats.get_message_by_id_and_message_id(chat_id, user_message_id) or {}
        await Chats.upsert_message_to_chat_by_id_and_message_id(
            chat_id,
            user_message_id,
            {
                'meta': {
                    **(stored_message.get('meta') or {}),
                    'skills_create': {
                        'original_text': get_content_from_message(stored_message),
                        'expanded_text': get_last_user_message(form_data['messages']),
                    },
                }
            },
            touch=False,
        )

    # If the original caller provided tools, use them as-is (skip resolution).
    # Otherwise, save any tools that filter inlets added for merging later.
    inlet_filter_tools = None if payload_tools is not None else form_data.get('tools', None)

    # Explicit activation preloads view_skill; lazy selection leaves loading to the model.
    chat_context = metadata.get('chat_context') or {}
    metadata['chat_context'] = chat_context
    skill_versions = chat_context.setdefault('skill_versions', {})
    mentioned_skill_ids = set(chat_context.get('mentioned_skill_ids') or []) | extract_skill_ids_from_messages(
        [message for message in form_data.get('messages', []) if message.get('role') == 'user']
    )
    activated_skill_ids = set(form_data.pop('skill_ids', None) or []) | mentioned_skill_ids
    skill_ids = sorted(activated_skill_ids | set(model.get('info', {}).get('meta', {}).get('skillIds', [])))
    if payload_tools is None and await Config.get('skills.lazy_loading.enable', False):
        activated_skill_ids = mentioned_skill_ids
    available_skills = []
    terminal_skills = []

    model_builtin_tools = model.get('info', {}).get('meta', {}).get('builtinTools') or {}
    skill_tools_enabled = payload_tools is None and use_builtin_tools and model_builtin_tools.get('skills', True)

    if skill_ids or (use_builtin_tools and model_builtin_tools.get('skills', True)):
        from open_webui.models.skills import Skills as SkillsModel
        from open_webui.utils.skill_files import format_skill_content, skill_overview
        from open_webui.utils.terminals import (
            format_terminal_skill_context,
            format_terminal_skill_manifest_entry,
            get_terminal_json,
            get_terminal_skill,
        )

        terminal_skill_prefix = 'terminal:'
        db_skill_ids = [sid for sid in skill_ids if not sid.startswith(terminal_skill_prefix)]
        terminal_skill_ids = [sid for sid in skill_ids if sid.startswith(terminal_skill_prefix)]

        if use_builtin_tools and model_builtin_tools.get('skills', True):
            accessible_skills = {s.id: s for s in await SkillsModel.get_skills(user_id=user.id)}
            db_skill_ids = sorted(accessible_skills)
        else:
            accessible_skills = {s.id: s for s in await SkillsModel.get_skills(user_id=user.id, ids=db_skill_ids)}

        for sid in db_skill_ids:
            skill = accessible_skills.get(sid)
            if skill and skill.is_active:
                available_skills.append(skill)

        skill_manifest = ''
        for skill in available_skills:
            if skill_tools_enabled:
                skill_manifest += (
                    f'<skill>\n<id>{skill.id}</id>\n<name>{skill.name}</name>\n'
                    f'<description>{skill.description or ""}</description>\n</skill>\n'
                )
                continue
            if skill.id in activated_skill_ids or not use_builtin_tools or not model_builtin_tools.get('skills', True):
                from open_webui.models.skills import get_skill_snapshot

                version_id = skill_versions.get(skill.id) or skill.version_id
                try:
                    snapshot = await get_skill_snapshot(skill, version_id)
                except HTTPException as error:
                    if error.status_code != 404:
                        raise
                    continue
                skill_versions[skill.id] = version_id
                overview = await skill_overview(
                    snapshot['content'],
                    [f['path'] for f in snapshot['data']['files'] if f['path'] != 'SKILL.md'],
                    tools_enabled=skill_tools_enabled,
                )
                root_content = format_skill_content(overview, skill.id, skill_tools_enabled)
                resources = '\n'.join(overview['files'])
                resource_hint = (
                    f'\nSupporting files:\n{resources}'
                    if skill_tools_enabled and resources
                    else '\nSupporting files require a model with skill tools enabled.'
                    if resources
                    else ''
                )
                if overview.get('notice'):
                    resource_hint += '\n' + overview['notice']
                form_data['messages'] = add_or_update_system_message(
                    f'<skill id="{skill.id}" name="{skill.name}">\n{root_content}{resource_hint}\n</skill>',
                    form_data['messages'],
                    append=True,
                )
            else:
                skill_manifest += (
                    f'<skill>\n<id>{skill.id}</id>\n<name>{skill.name}</name>\n'
                    f'<description>{skill.description or ""}</description>\n</skill>\n'
                )

        listed = (
            await get_terminal_json(request, user, metadata, '/skills', extra_params)
            if terminal_id or terminal_skill_ids
            else None
        )

        listed_terminal_skills = listed if isinstance(listed, list) else []

        if terminal_id and use_builtin_tools and model_builtin_tools.get('skills', True):
            terminal_skills = listed_terminal_skills
        elif terminal_skill_ids:
            terminal_skill_map = {skill['id']: skill for skill in listed_terminal_skills}
            terminal_skills = [skill for sid in terminal_skill_ids if (skill := terminal_skill_map.get(sid))]

        for skill in sorted(terminal_skills, key=lambda skill: skill['id']):
            sid = skill['id']
            if skill_tools_enabled:
                skill_manifest += format_terminal_skill_manifest_entry(skill)
                continue
            if sid in activated_skill_ids or not use_builtin_tools or not model_builtin_tools.get('skills', True):
                skill_name = unquote(sid.removeprefix(terminal_skill_prefix))
                loaded = await get_terminal_skill(
                    request,
                    user.model_dump(),
                    metadata,
                    skill_name,
                    extra_params,
                    overview=True,
                    tools_enabled=use_builtin_tools and model_builtin_tools.get('skills', True),
                )
                if loaded:
                    form_data['messages'] = add_or_update_system_message(
                        format_terminal_skill_context(
                            loaded, sid, use_builtin_tools and model_builtin_tools.get('skills', True)
                        ),
                        form_data['messages'],
                        append=True,
                    )
            else:
                skill_manifest += format_terminal_skill_manifest_entry(skill)

        if skill_tools_enabled and skill_manifest:
            if not metadata.get('assistant_message_id'):
                initial_output.extend(
                    hidden_tool_result(
                        'list_available_skills', f'<available_skills>\n{skill_manifest}</available_skills>'
                    )
                )
        elif skill_manifest:
            form_data['messages'] = add_or_update_system_message(
                f'<available_skills>\n{skill_manifest}</available_skills>',
                form_data['messages'],
                append=True,
            )

    # Strip only resolved skill mentions; ordinary text such as Perl's <$fh> stays intact.
    resolved_skill_ids = {s.id for s in available_skills} | {s['id'] for s in terminal_skills}
    chat_context['mentioned_skill_ids'] = sorted(mentioned_skill_ids & resolved_skill_ids)
    strip_skill_mentions(form_data.get('messages', []), resolved_skill_ids)

    prompt = get_last_user_message(form_data['messages'])

    # Supply fallback text for skill-only messages; some providers reject empty content.
    if not prompt or not prompt.strip():
        fallback = ', '.join([s.name for s in available_skills] + [s['name'] for s in terminal_skills])
        # Attachment-only messages keep their empty text, same as on models without skills.
        if fallback and not (metadata.get('user_message') or {}).get('files'):
            set_last_user_message_content(fallback, form_data['messages'])
            prompt = fallback
    # TODO: re-enable URL extraction from prompt
    # urls = []
    # if prompt and len(prompt or "") < 500 and (not files or len(files) == 0):
    #     urls = extract_urls(prompt)

    if files:
        # files = [*files, *[{"type": "url", "url": url, "name": url} for url in urls]]
        # Remove duplicate files based on their content
        files = list({json.dumps(f, sort_keys=True): f for f in files}.values())

    metadata.update(
        {
            'model_id': form_data.get('model'),
            'tool_ids': tool_ids,
            'skill_ids': skill_ids,
            'terminal_id': terminal_id,
            'files': files,
            'features': features,
        }
    )
    form_data['metadata'] = metadata

    # When the caller provides an explicit `tools` key in the request body,
    # skip all server-side tool resolution and pass the caller's tools through
    # unchanged.  Sending `tools: []` explicitly opts out of builtin injection.
    if payload_tools is None:
        tool_ids = metadata.get('tool_ids', None)
        log.debug('tool_ids=%r', tool_ids)

        tools_dict = {}

        mcp_clients = {}
        mcp_tools_dict = {}

        if tool_ids:
            resolved_tool_ids = []
            for tool_id in tool_ids:
                if tool_id.startswith('server:mcp:'):
                    try:
                        server_id = tool_id[len('server:mcp:') :]

                        result = await connect_mcp_server(
                            request,
                            server_id,
                            user,
                            extra_params,
                        )
                        if result is None:
                            continue

                        client, tool_specs = result
                        mcp_clients[server_id] = client

                        if client.instructions:
                            form_data['messages'] = add_or_update_system_message(
                                f'MCP server {JSONCodec.dumps(server_id)} instructions:\n{client.instructions}',
                                form_data['messages'],
                                append=True,
                            )

                        for tool_spec in tool_specs:

                            async def make_tool_function(client, function_name):
                                async def tool_function(**kwargs):
                                    return await client.call_tool(
                                        function_name,
                                        function_args=kwargs,
                                    )

                                return tool_function

                            tool_function = await make_tool_function(client, tool_spec['name'])

                            tool_name = f'{server_id}_{tool_spec["name"]}'
                            # Providers reject tool names over 64 characters
                            if len(tool_name) > 64:
                                tool_name = f'{tool_name[:55]}_{hashlib.sha256(tool_name.encode()).hexdigest()[:8]}'

                            mcp_tools_dict[tool_name] = {
                                'spec': {
                                    **tool_spec,
                                    'name': tool_name,
                                },
                                'callable': tool_function,
                                'type': 'mcp',
                                'client': client,
                            }
                    except Exception:
                        log.exception("Failed to connect to MCP server '%s'", server_id)
                        if event_emitter:
                            await event_emitter(
                                {
                                    'type': 'status',
                                    'data': {
                                        'action': 'tool_connection',
                                        'description': f"Failed to connect to MCP server '{server_id}'",
                                        'error': True,
                                        'done': True,
                                    },
                                }
                            )
                        continue
                else:
                    resolved_tool_ids.append(tool_id)

            if resolved_tool_ids:
                tools_dict = await get_tools(
                    request,
                    resolved_tool_ids,
                    user,
                    {
                        **extra_params,
                        '__model__': models[task_model_id],
                        '__messages__': form_data['messages'],
                        '__files__': metadata.get('files') or [],
                    },
                )

            if mcp_tools_dict:
                tools_dict = {**tools_dict, **mcp_tools_dict}

        # Resolve terminal tools if terminal_id is set (outside tool_ids check
        # so system terminals work even when no other tools are selected).
        # get_terminal_tools does the access check, including a shared chat's owner pick.
        terminal_capability = (model.get('info', {}).get('meta', {}).get('capabilities') or {}).get('terminal', True)
        if terminal_id and terminal_capability:
            try:
                terminal_result = await get_terminal_tools(
                    request,
                    terminal_id,
                    user,
                    extra_params,
                )
                if isinstance(terminal_result, tuple):
                    terminal_tools, terminal_system_prompt = terminal_result
                else:
                    terminal_tools = terminal_result
                    terminal_system_prompt = None
                if terminal_tools:
                    tools_dict = {**tools_dict, **terminal_tools}
                if terminal_system_prompt:
                    form_data['messages'] = add_or_update_system_message(
                        terminal_system_prompt,
                        form_data['messages'],
                        append=True,
                    )
            except Exception as e:
                log.exception(e)
                raise HTTPException(status_code=503, detail=f'Terminal unavailable: {e}') from e

        if terminal_id and terminal_capability:
            from open_webui.utils.terminals import add_terminal_agents_md, get_terminal_agents_md

            agents_md = await get_terminal_agents_md(request, user, metadata, extra_params)
            if agents_md:
                form_data['messages'] = add_terminal_agents_md(form_data['messages'], agents_md)

        if mcp_clients:
            metadata['mcp_clients'] = mcp_clients

        # Inject builtin tools for native function calling based on enabled features and model capability.
        # Only inject when the request originates from the UI (identified by session_id).
        # API callers don't expect hidden tools; they can explicitly request tools via tool_ids.
        if use_builtin_tools:
            if model_builtin_tools.get('knowledge', True):
                from html import escape

                knowledge_tags = []
                for item in get_attached_knowledge(model, metadata):
                    # Chat attachments are already replayed from each user message's files.
                    if item.get('source') == 'chat':
                        continue
                    if not item.get('id') or not item.get('type'):
                        continue
                    attrs = f'type="{escape(str(item["type"]), quote=True)}" id="{escape(str(item["id"]), quote=True)}"'
                    if item.get('name'):
                        attrs += f' name="{escape(str(item["name"]), quote=True)}"'
                    if item.get('source'):
                        attrs += f' source="{escape(str(item["source"]), quote=True)}"'
                    knowledge_tags.append(f'<knowledge {attrs}/>')

                if knowledge_tags and not metadata.get('assistant_message_id'):
                    initial_output.extend(
                        hidden_tool_result(
                            'list_attached_knowledge',
                            '<attached_knowledge>\n' + '\n'.join(knowledge_tags) + '\n</attached_knowledge>',
                        )
                    )

            builtin_tools = await get_builtin_tools(
                request,
                {
                    **extra_params,
                    '__event_emitter__': event_emitter,
                    '__skill_ids__': sorted(resolved_skill_ids),
                },
                features,
                model,
                is_note_chat=is_note_chat,
            )
            for name, tool_dict in builtin_tools.items():
                if name not in tools_dict:
                    tools_dict[name] = tool_dict
                elif name == 'display_file' and tools_dict[name].get('type') == 'terminal':
                    tools_dict[name] = {**tools_dict[name], 'file_tool': tool_dict}

        await filter_user_terminal_tools(request, user, tools_dict, metadata, extra_params)

        if tools_dict:
            # Always store resolved tools in metadata so downstream consumers
            # (e.g. pipe functions) can access all tools including MCP and builtins.
            metadata['tools'] = tools_dict

            if metadata.get('params', {}).get('function_calling') != 'legacy':
                deferred = await apply_tool_search(
                    form_data, metadata, tools_dict, inlet_filter_tools, extra_params=extra_params
                )

                settings = await Config.get_many(
                    'chat.tool_call_labels.enable',
                    'chat.tool_permissions.enable',
                    'chat.tool_permissions.auto.enable',
                    'chat.tool_permissions.auto.engine',
                )
                approval = (
                    metadata.get('params', {}).get('tool_approval_mode') == 'auto'
                    and settings.get('chat.tool_permissions.enable')
                    and settings.get('chat.tool_permissions.auto.enable')
                    and settings.get('chat.tool_permissions.auto.engine', '') == ''
                )
                add_tool_call_metadata(
                    tools_dict,
                    labels=settings.get('chat.tool_call_labels.enable', False),
                    approval=approval,
                    language=(metadata.get('variables') or {}).get('{{USER_LANGUAGE}}') or 'English',
                )
                form_data['messages'] = set_tool_approval_prompt(
                    form_data['messages'], (await get_tool_permissions_auto_config())['prompt'] if approval else None
                )

                # If the function calling is native, then call the tools function calling handler
                form_data['tools'] = [
                    {'type': 'function', 'function': tool.get('spec', {})}
                    for name, tool in tools_dict.items()
                    if name not in deferred
                ]
                if inlet_filter_tools:
                    form_data['tools'].extend(inlet_filter_tools)
            else:
                # If the function calling is not native, then call the tools function calling handler
                try:
                    form_data, flags = await chat_completion_tools_handler(
                        request, form_data, extra_params, user, models, tools_dict
                    )
                    sources.extend(flags.get('sources', []))
                except Exception as e:
                    log.exception(e)

    file_context_enabled = is_file_context_enabled(model, metadata)

    if file_context_enabled:
        try:
            form_data, flags = await chat_completion_files_handler(request, form_data, extra_params, user)
            sources.extend(flags.get('sources', []))
        except Exception as e:
            log.exception(e)

    # Compaction summarizes the user's text without injected retrieval context.
    metadata['base_user_prompt'] = get_last_user_message(form_data['messages'])
    metadata['sources'] = sources[:] if sources else []

    # If context is not empty, insert it into the messages
    if sources and prompt:
        form_data['messages'] = await apply_source_context_to_messages(request, form_data['messages'], sources, prompt)

    # If there are citations, add them to the data_items
    sources = [
        source
        for source in sources
        if source.get('source', {}).get('name', '') or source.get('source', {}).get('id', '')
    ]

    if len(sources) > 0:
        events.append({'sources': sources})

    if model_knowledge:
        await event_emitter(
            {
                'type': 'status',
                'data': {
                    'action': 'knowledge_search',
                    'query': user_message,
                    'done': True,
                    'hidden': True,
                },
            }
        )

    if pending_assistant_message is not None:
        if await resume_tool_calls(request, form_data, user, model, metadata, pending_assistant_message):
            return form_data, metadata, events, True
        assistant_message = dict(pending_assistant_message)
        if assistant_message.get('model') != model['id'] and isinstance(assistant_message.get('output'), list):
            assistant_message['output'] = strip_reasoning_details(assistant_message['output'])
        if isinstance(assistant_message.get('output'), list):
            assistant_message['output'] = strip_tool_images(assistant_message['output'], model)
        form_data['messages'].extend(
            process_messages_with_output([assistant_message], reasoning_format=get_reasoning_format(model))
        )
        await convert_url_images_to_base64(form_data, user=user)

    if ENABLE_FUNCTIONS:
        try:
            form_data, _ = await process_filter_functions(
                request=request,
                filter_context=filter_context,
                filter_functions=filter_functions,
                filter_type='request',
                form_data=form_data,
                extra_params=extra_params,
            )
        except Exception as e:
            raise Exception(f'{e}')

    metadata.pop('initial_output', None)
    if skill_tools_enabled and not metadata.get('assistant_message_id'):
        initial_output.extend(
            await preload_skills(
                form_data['messages'],
                activated_skill_ids & resolved_skill_ids,
                metadata.get('tools', {}).get('view_skill'),
            )
        )
    if initial_output:
        form_data['messages'].extend(
            convert_output_to_messages(initial_output, raw=True, reasoning_format=get_reasoning_format(model))
        )
        metadata['initial_output'] = initial_output

    form_data = normalize_messages_for_model(form_data)

    if compaction_messages is not None:
        summary_form_data = {**form_data}
        if context_summary:
            summary_form_data['messages'] = add_or_update_system_message(
                f'[CONVERSATION SUMMARY]\n{context_summary}',
                [dict(message) for message in form_data['messages']],
                append=True,
            )
        try:
            _, context_summary, compacted = await compact_messages_for_request(
                request,
                summary_form_data,
                user,
                compaction_messages,
                metadata,
                form_data['model'],
                request.app.state.MODELS,
                chat_system_prompt,
            )
            if compacted:
                # Compacted messages lead the history this request was built from.
                compacted_count = len(
                    sanitize_tool_pairs(
                        process_messages_with_output(
                            compacted,
                            reasoning_format=get_reasoning_format(model),
                            include_file_context=metadata['include_file_context'],
                        )
                    )
                )
                compacted_ids = {id(message) for message in history_messages[:compacted_count]}
                form_data['messages'] = [
                    message for message in form_data['messages'] if id(message) not in compacted_ids
                ]
        except Exception:
            log.exception('Context compaction failed; continuing with full chat history')

        if context_summary:
            summary_content = f'[CONVERSATION SUMMARY]\n{context_summary}'
            form_data['messages'] = add_or_update_system_message(summary_content, form_data['messages'], append=True)

    metadata['context_summary'] = context_summary
    return form_data, metadata, events, False


async def preload_memory(request, messages: list[dict], user, model: dict) -> list[dict]:
    """Keep automatic recall in ordinary tool history so later turns replay it unchanged."""
    result = await recall_memories(request, messages, user, model)
    return hidden_tool_result('recall_memories', result) if result else []


def hidden_tool_result(name: str, result: str) -> list[dict]:
    """Use ordinary saved tool output for internal context without displaying it in chat."""
    call_id = output_id('fc')
    return [
        {
            'type': 'function_call',
            'id': call_id,
            'call_id': call_id,
            'name': name,
            'hidden': True,
            'arguments': '{}',
            'status': 'completed',
        },
        {
            'type': 'function_call_output',
            'id': output_id('fco'),
            'call_id': call_id,
            'output': [{'type': 'input_text', 'text': result}],
            'status': 'completed',
        },
    ]


async def preload_skills(messages: list[dict], skill_ids: set[str], tool: dict | None) -> list[dict]:
    """Record explicit activation as ordinary tool history for saved and temporary chats."""
    if not tool or not skill_ids:
        return []
    completed = {
        message.get('tool_call_id')
        for message in messages
        if message.get('role') == 'tool' and not _is_tool_result_error(message.get('content', ''))
    }
    loaded = set()
    for message in messages:
        for call in message.get('tool_calls') or []:
            function = call.get('function') or {}
            if function.get('name') == 'view_skill' and call.get('id') in completed:
                try:
                    loaded.add(JSONCodec.loads(function.get('arguments', '{}')).get('id'))
                except (ValueError, TypeError, AttributeError):
                    pass
    output = []
    for skill_id in sorted(skill_ids - loaded):
        result = await tool['callable'](id=skill_id)
        call_id = output_id('fc')
        status = 'failed' if _is_tool_result_error(result) else 'completed'
        output.extend(
            [
                {
                    'type': 'function_call',
                    'id': call_id,
                    'call_id': call_id,
                    'name': 'view_skill',
                    'hidden': True,
                    'arguments': JSONCodec.dumps({'id': skill_id}),
                    'status': status,
                },
                {
                    'type': 'function_call_output',
                    'id': output_id('fco'),
                    'call_id': call_id,
                    'output': [{'type': 'input_text', 'text': result}],
                    'status': status,
                },
            ]
        )
    return output


async def get_event_emitter_and_caller(metadata):
    event_emitter = None
    event_caller = None

    # event_emitter only needs user_id + chat_id + message_id.
    # It broadcasts to user:{user_id} room AND persists to DB,
    # so it works for backend-initiated calls (automations, API).
    if metadata.get('chat_id') and metadata.get('message_id'):
        event_emitter = await get_event_emitter(metadata)

    # event_caller needs session_id — it calls back to a specific
    # websocket session (used by direct tools, pyodide code interpreter).
    if metadata.get('session_id') and metadata.get('chat_id') and metadata.get('message_id'):
        event_caller = await get_event_call(metadata)

    return event_emitter, event_caller


async def build_chat_response_context(request, form_data, user, model, metadata, tasks, events):
    event_emitter, event_caller = await get_event_emitter_and_caller(metadata)
    assistant_message = None
    if metadata.get('assistant_message_id'):
        # Preserve the original output before request conversion strips it from temporary chats.
        if is_saved_chat_id(metadata.get('chat_id')):
            assistant_message = await Chats.get_message_by_id_and_message_id(
                metadata['chat_id'], metadata['assistant_message_id']
            )
        elif form_data.get('messages') and form_data['messages'][-1].get('role') == 'assistant':
            assistant_message = form_data['messages'][-1]
    return {
        'request': request,
        'form_data': form_data,
        'user': user,
        'model': model,
        'metadata': metadata,
        'tasks': tasks,
        'events': events,
        'event_emitter': event_emitter,
        'event_caller': event_caller,
        'assistant_message': copy.deepcopy(assistant_message),
    }


async def execute_tool_call(form_data, metadata, event_caller, tool_call, history=None):
    """Execute a resolved call; resumed approvals use the same validation and loop policy."""
    name = tool_call.get('name', '')
    tool = metadata.get('tools', {}).get(name)
    error = (
        prepare_tool_call(
            tool_call,
            tool.get('spec', {}).get('parameters'),
            display_labels=tool.get('display_labels', False),
            approval_review=tool.get('approval_review', False),
        )
        if tool
        else f'Error: Tool "{name}" not found.'
    )
    if history is not None:
        error = error or tool_repetition_error(tool_call, history)
        history.append(tool_call)
    tool_call['status'] = 'failed' if error else 'in_progress'
    if error:
        return {}, error, None, None
    params = JSONCodec.loads(tool_call['arguments'])
    if name == 'display_file' and (
        str(params.get('path', '')).startswith('/api/v1/files/')
        or re.fullmatch(r'[0-9a-fA-F]{8}(?:-[0-9a-fA-F]{4}){3}-[0-9a-fA-F]{12}', str(params.get('path', '')))
    ):
        if tool.get('file_tool'):
            tool = tool['file_tool']
        elif tool.get('type') == 'terminal':
            tool_call['status'] = 'failed'
            return params, 'Error: Uploaded file display is not available.', None, None
    tool_type = tool.get('type', '')
    if tool_type != 'mcp':
        allowed_params = tool.get('spec', {}).get('parameters', {}).get('properties', {})
        params = {key: value for key, value in params.items() if key in allowed_params}
    requires_approval = await requires_tool_approval(form_data, metadata, tool_call, tool, params)
    if (
        requires_approval
        and is_saved_chat_id(metadata.get('chat_id'))
        and (metadata.get('message_id') or metadata.get('assistant_message_id'))
        and metadata.get('params', {}).get('function_calling') != 'legacy'
    ):
        tool_call['status'] = 'pending'
        raise ToolApprovalRequired
    try:
        if requires_approval and (
            not event_caller
            or await event_caller(
                {
                    'type': 'confirmation',
                    'data': {'tool_call': {'id': tool_call.get('call_id'), 'name': name, 'arguments': params}},
                }
            )
            is not True
        ):
            tool_call['status'] = 'failed'
            return params, 'Error: tool call was not approved.', tool, tool_type

        tool_call['started_at'] = time.time()
        started = time.perf_counter()
        try:
            function = await get_updated_tool_function(
                function=tool['callable'],
                extra_params={
                    '__messages__': form_data.get('messages', []),
                    '__files__': metadata.get('files') or [],
                },
            )
            result = await function(**params)
        finally:
            tool_call['ended_at'] = time.time()
            tool_call['duration'] = time.perf_counter() - started
    except Exception as e:
        result = {'error': str(e)}
    tool_call['status'] = 'failed' if _is_tool_result_error(result) else 'completed'
    return params, result, tool, tool_type


async def resume_tool_calls(request, form_data, user, model, metadata, message) -> bool:
    """Execute approved calls in the supplied response without rebuilding prepared history."""
    chat_id = metadata.get('chat_id')
    message_id = metadata.get('message_id') or metadata.get('assistant_message_id')
    output = message.get('output')
    if not isinstance(output, list):
        return False

    event_emitter, event_caller = await get_event_emitter_and_caller(metadata)
    citation_source_ids = {}
    get_source_context(
        message.get('sources') or metadata.get('sources', []), citation_source_ids, include_content=False
    )

    # Request filters must be able to reject the complete request before tool side effects.
    # Give tools their filtered view without letting it overwrite the prepared history.
    assistant_message = dict(message)
    if assistant_message.get('model') != model['id']:
        assistant_message['output'] = strip_reasoning_details(output)
    form_data = {
        **form_data,
        'messages': copy.deepcopy(
            [
                *form_data['messages'],
                *process_messages_with_output([assistant_message], reasoning_format=get_reasoning_format(model)),
            ]
        ),
    }
    await convert_url_images_to_base64(form_data, user=user)
    if ENABLE_FUNCTIONS:
        filter_functions = await get_filter_functions(request, model, metadata.get('filter_ids', []))
        if filter_functions:
            form_data, _ = await process_filter_functions(
                request=request,
                filter_context=get_filter_context(request),
                filter_functions=filter_functions,
                filter_type='request',
                form_data=form_data,
                extra_params={
                    '__event_emitter__': event_emitter,
                    '__event_call__': event_caller,
                    '__user__': user.model_dump() if isinstance(user, UserModel) else {},
                    '__metadata__': metadata,
                    '__oauth_token__': await get_system_oauth_token(request, user),
                    '__request__': request,
                    '__model__': model,
                    '__chat_id__': metadata.get('chat_id'),
                    '__message_id__': metadata.get('message_id'),
                },
            )
    normalize_messages_for_model(form_data)

    while True:
        async with Chats.edit_message_output(chat_id, message_id) as saved:
            if not saved:
                raise HTTPException(status_code=404, detail='Message not found.')
            output = saved.get('output') or []
            pending = pending_tool_calls(output)
            if any(entry.get('status') == 'in_progress' for entry in pending):
                return True
            item = next(
                (
                    entry
                    for entry in pending
                    if entry.get('name') != 'ask_user'
                    and (
                        entry.get('approved') is True
                        or metadata.get('params', {}).get('tool_approval_mode') in {'full', 'auto'}
                    )
                ),
                None,
            )
            if item is None:
                if pending:
                    pending[0]['status'] = 'pending'
                # A stale continuation must not start another model response.
                return True
            item['call_id'] = item.get('call_id') or item['id']
            item['status'] = 'in_progress'
            saved['done'] = False
            item = copy.deepcopy(item)

        try:
            history = [call for call in output if call.get('type') == 'function_call']
            history = history[: next(i for i, call in enumerate(history) if call.get('call_id') == item['call_id'])]
            params, result, tool, tool_type = await execute_tool_call(form_data, metadata, event_caller, item, history)
            files, embeds = [], []
            if result is None and tool is None:
                result = (
                    'Error: Tool call arguments could not be parsed. The model generated malformed or incomplete JSON.'
                )
            elif tool:
                name = item.get('name', '')
                terminal_file_result = build_terminal_file_tool_result(name, params, result, tool, metadata)
                if terminal_file_result:
                    result = terminal_file_result
                result, files, embeds = await process_tool_result(request, name, result, tool_type, metadata, user)
                await terminal_event_handler(name, params, result, event_emitter)
            content = await tool_result_with_citations(
                item.get('name', ''),
                params,
                result,
                tool,
                model,
                citation_source_ids,
                get_last_user_message(form_data['messages']),
                event_emitter,
            )
            output_parts = [{'type': 'input_text', 'text': content}]
            item['status'] = 'failed' if _is_tool_result_error(content) else 'completed'
            display_files = []
            for file_item in files:
                if file_item.get('type') == 'image' and file_item.get('url', '').startswith('data:'):
                    image_url = await store_tool_result_image(request, file_item['url'], metadata, user)
                    output_parts.append({'type': 'input_image', 'image_url': image_url})
                else:
                    display_files.append(file_item)
                    if file_item.get('type') == 'image' and file_item.get('url'):
                        output_parts.append({'type': 'input_image', 'image_url': file_item['url']})

            result_item = {
                'type': 'function_call_output',
                'id': output_id('fco'),
                'call_id': item['call_id'],
                'output': output_parts,
                'status': item['status'],
                **({'files': display_files} if display_files else {}),
                **({'embeds': embeds} if embeds else {}),
            }
            output = await complete_tool_call(chat_id, message_id, item, result_item)
        except ToolApprovalRequired:
            async with Chats.edit_message_output(chat_id, message_id) as saved:
                if not saved:
                    raise HTTPException(404, 'Message not found.')
                output = saved.get('output') or []
                target = next(
                    entry
                    for entry in output
                    if entry.get('call_id') == item['call_id'] and entry.get('type') == 'function_call'
                )
                target.update(item)
                target['status'] = 'pending'
            if event_emitter:
                await event_emitter({'type': 'chat:completion', 'data': {'done': False, 'output': output}})
            return True
        except BaseException:
            item['status'] = 'incomplete'
            await asyncio.shield(
                complete_tool_call(
                    chat_id,
                    message_id,
                    item,
                    {
                        'type': 'function_call_output',
                        'id': output_id('fco'),
                        'call_id': item['call_id'],
                        'status': 'incomplete',
                        'output': [
                            {
                                'type': 'input_text',
                                'text': (
                                    'Error: tool execution was interrupted; its outcome is unknown.'
                                    if item.get('started_at')
                                    else 'Error: tool review was interrupted; the tool was not executed.'
                                ),
                            }
                        ],
                    },
                )
            )
            raise
        if event_emitter:
            await event_emitter({'type': 'chat:completion', 'data': {'done': False, 'output': output}})
        pending = pending_tool_calls(output)
        if not pending:
            message['output'] = output
            return False
        if metadata.get('params', {}).get('tool_approval_mode') not in {'full', 'auto'} and not any(
            entry.get('approved') is True for entry in pending
        ):
            return True


async def pause_for_tool_approval(chat_id: str, message_id: str, output: list[dict], form_data: dict, metadata: dict):
    result_call_ids = {
        item.get('call_id') for item in output if item.get('type') == 'function_call_output' and item.get('call_id')
    }
    has_pending_approval = False
    for item in output:
        if item.get('type') == 'function_call' and not item.get('call_id') and item.get('id'):
            item['call_id'] = item['id']

        if (
            item.get('type') == 'function_call'
            and item.get('call_id')
            and item.get('call_id') not in result_call_ids
            and item.get('status') != 'rejected'
        ):
            if not has_pending_approval:
                item['status'] = 'pending'
                has_pending_approval = True
            elif item.get('status') in {'in_progress', 'completed'}:
                item['status'] = 'queued'

    await Chats.upsert_message_to_chat_by_id_and_message_id(
        chat_id,
        message_id,
        {
            'done': False,
            'output': output,
            'meta': {
                **(metadata.get('tool_approval') or {}),
                'session_id': metadata.get('session_id'),
                'tool_ids': metadata.get('tool_ids') or [],
                'skill_ids': metadata.get('skill_ids') or [],
                'terminal_id': metadata.get('terminal_id'),
                'filter_ids': metadata.get('filter_ids') or [],
                'features': metadata.get('features') or {},
                'variables': metadata.get('variables') or {},
                'files': metadata.get('files') or [],
                'params': metadata.get('params') or {},
            },
        },
        touch=False,
    )
    event_emitter = await get_event_emitter(metadata, update_db=False)
    if event_emitter:
        await event_emitter({'type': 'chat:active', 'data': {'active': True}})


def get_response_data(response):
    if isinstance(response, list) and len(response) == 1:
        # If the response is a single-item list, unwrap it #17213
        response = response[0]

    if isinstance(response, JSONResponse):
        if isinstance(response.body, bytes):
            try:
                response_data = JSONCodec.loads(response.body.decode('utf-8', 'replace'))
            except JSONCodec.JSONDecodeError:
                response_data = {'error': {'detail': 'Invalid JSON response'}}
        else:
            response_data = response
    elif isinstance(response, dict):
        response_data = response
    else:
        response_data = None

    return response, response_data


def merge_events_into_response(response_data, events):
    if events and isinstance(events, list):
        extra_response = {}
        for event in events:
            if isinstance(event, dict):
                extra_response.update(event)
            else:
                extra_response[event] = True

        return {
            **extra_response,
            **response_data,
        }
    return response_data


def build_response_object(response, response_data):
    if isinstance(response, dict):
        return response_data
    if isinstance(response, JSONResponse):
        return JSONResponse(
            content=response_data,
            headers=response.headers,
            status_code=response.status_code,
        )
    return response


def update_assistant_message_from_stream(assistant_message, raw) -> bool:
    """Accumulate the message and report whether the chunk contains [DONE]."""
    line = raw.decode('utf-8', 'replace') if isinstance(raw, bytes) else raw
    if not isinstance(line, str):
        return False

    def append_output_text(item, text):
        parts = item.setdefault('content', [])
        if parts and parts[-1].get('type') == 'output_text':
            append_to_text_field(parts[-1], 'text', text)
        else:
            parts.append({'type': 'output_text', 'text': text})

    done = False
    for raw_part in line.splitlines():
        part = raw_part.removeprefix('data:').strip()
        if part == '[DONE]':
            done = True
            continue
        if not part:
            continue

        try:
            data = JSONCodec.loads(part)
        except Exception:
            continue

        if not isinstance(data, dict):
            continue

        if data.get('type', '').startswith('response.'):
            output, meta = handle_responses_streaming_event(data, assistant_message.get('output', []))
            if output:
                assistant_message['output'] = output
            if meta and meta.get('usage'):
                assistant_message['usage'] = update_usage_snapshot(assistant_message.get('usage'), meta['usage'])
            continue

        raw_usage = data.get('usage', {}) or {}
        raw_usage.update(data.get('timings', {}))
        if raw_usage:
            assistant_message['usage'] = update_usage_snapshot(assistant_message.get('usage'), raw_usage)

        for choice in data.get('choices', []):
            if choice.get('finish_reason'):
                assistant_message['finish_reason'] = choice['finish_reason']
            delta = choice.get('delta', {}) or {}
            content = delta.get('content')
            reasoning_content = delta.get('reasoning_content') or delta.get('reasoning') or delta.get('thinking')

            if reasoning_content:
                output = assistant_message.setdefault('output', [])
                if not output or output[-1].get('type') != 'reasoning':
                    output.append(
                        {
                            'type': 'reasoning',
                            'id': output_id('r'),
                            'status': 'in_progress',
                            'start_tag': '<think>',
                            'end_tag': '</think>',
                            'attributes': {'type': 'reasoning_content'},
                            'content': [],
                            'summary': None,
                            'started_at': time.time(),
                        }
                    )

                append_output_text(output[-1], reasoning_content)

            if content:
                output = assistant_message.get('output')
                if output:
                    if output[-1].get('type') == 'reasoning':
                        output[-1]['status'] = 'completed'
                        output[-1]['ended_at'] = time.time()
                        output[-1]['duration'] = int(output[-1]['ended_at'] - output[-1]['started_at'])

                    if not output or output[-1].get('type') != 'message':
                        output.append(
                            {
                                'type': 'message',
                                'id': output_id('msg'),
                                'status': 'in_progress',
                                'role': 'assistant',
                                'content': [],
                            }
                        )

                    append_output_text(output[-1], content)

                if 'content' in assistant_message:
                    append_to_text_field(assistant_message, 'content', content)
                else:
                    assistant_message['content'] = '' + content

    return done


async def background_tasks_handler(ctx):
    request = ctx['request']
    form_data = ctx['form_data']
    user = ctx['user']
    metadata = ctx['metadata']
    tasks = ctx['tasks']
    event_emitter = ctx['event_emitter']

    if not await get_verified_user_by_id(user.id):
        return

    message = None
    messages = []

    if is_saved_chat_id(metadata.get('chat_id')):
        messages_map = await Chats.get_messages_map_by_chat_id(metadata['chat_id'])
        if not messages_map:
            # Chat was deleted while the response was streaming — skip background tasks
            return
        message = messages_map.get(metadata['message_id'])

        message_list = get_message_list(messages_map, metadata['message_id'])

        # Remove details tags and files from the messages.
        # as get_message_list creates a new list, it does not affect
        # the original messages outside of this handler

        messages = []
        for message in message_list:
            content = message.get('content', '')
            if isinstance(content, list):
                content = '\n'.join(
                    item['text']
                    for item in content
                    if isinstance(item, dict) and item.get('type') == 'text' and isinstance(item.get('text'), str)
                )

            if isinstance(content, str):
                content = re.sub(
                    r'<details\b[^<>]*>(?:(?!<details\b).)*?<\/details>|!\[[^\[\]]*\]\([^()]*\)',
                    '',
                    content,
                    flags=re.S | re.I,
                ).strip()

            messages.append(
                {
                    **message,
                    'role': message.get('role', 'assistant'),  # Safe fallback for missing role
                    'content': content,
                }
            )
    else:
        # Local temp chat, get the model and message from the form_data
        message = get_last_user_message_item(form_data.get('messages', []))
        messages = form_data.get('messages', [])
        if message:
            message = {**message, 'model': form_data.get('model')}

    if message and 'model' in message:
        if tasks and messages:
            if TASKS.FOLLOW_UP_GENERATION in tasks and tasks[TASKS.FOLLOW_UP_GENERATION]:
                follow_up_messages = list(messages)
                assistant_message = ctx.get('assistant_message')
                if not is_saved_chat_id(metadata.get('chat_id')) and assistant_message:
                    if follow_up_messages[-1].get('role') == 'assistant' and (
                        follow_up_messages[-1].get('id') == metadata['message_id']
                        or get_content_from_message(follow_up_messages[-1])
                        == get_content_from_message(assistant_message)
                    ):
                        follow_up_messages.pop()
                    follow_up_messages.append({**assistant_message, 'role': 'assistant', 'id': metadata['message_id']})

                res = await generate_follow_ups(
                    request,
                    {
                        'model': message['model'],
                        'messages': follow_up_messages,
                        'message_id': metadata['message_id'],
                        'chat_id': metadata['chat_id'],
                    },
                    user,
                )

                if res and isinstance(res, dict):
                    if len(res.get('choices', [])) == 1:
                        response_message = res.get('choices', [])[0].get('message', {})

                        follow_ups_string = response_message.get('content') or response_message.get(
                            'reasoning_content', ''
                        )
                    else:
                        follow_ups_string = ''

                    try:
                        follow_ups = parse_follow_ups(follow_ups_string)
                        await event_emitter(
                            {
                                'type': 'chat:message:follow_ups',
                                'data': {
                                    'follow_ups': follow_ups,
                                },
                            }
                        )

                        if is_saved_chat_id(metadata.get('chat_id')):
                            await Chats.upsert_message_to_chat_by_id_and_message_id(
                                metadata['chat_id'],
                                metadata['message_id'],
                                {
                                    'followUps': follow_ups,
                                },
                                touch=False,
                            )

                    except Exception as e:
                        pass

            if is_saved_chat_id(metadata.get('chat_id')):  # Only update titles and tags for saved chats
                if TASKS.TITLE_GENERATION in tasks:
                    user_message = get_last_user_message(messages)
                    if user_message and len(user_message) > 100:
                        user_message = user_message[:100] + '...'

                    title = None
                    expected_title = None
                    if str(message['model']).startswith('agent:'):
                        first_user = next((m for m in messages if m.get('role') == 'user'), {})
                        title = (
                            replace_skill_mentions_with_labels(first_user.get('content', '')).strip()[:50] or 'New Chat'
                        )
                        updated = await Chats.update_chat_title_by_id(
                            metadata['chat_id'], title, expected_title='New Chat'
                        )
                        if not updated:
                            return  # The chat was deleted or already renamed.
                        await event_emitter({'type': 'chat:title', 'data': title})
                        expected_title = title
                    if tasks[TASKS.TITLE_GENERATION]:
                        res = await generate_title(
                            request,
                            {
                                'model': message['model'],
                                'messages': messages,
                                'chat_id': metadata['chat_id'],
                            },
                            user,
                        )

                        if res and isinstance(res, dict):
                            if len(res.get('choices', [])) == 1:
                                response_message = res.get('choices', [])[0].get('message', {})

                                title_string = (
                                    response_message.get('content')
                                    or response_message.get(
                                        'reasoning_content',
                                    )
                                    or message.get('content', user_message)
                                )
                            else:
                                title_string = ''

                            title_string = title_string[title_string.find('{') : title_string.rfind('}') + 1]

                            try:
                                title = JSONCodec.loads(title_string).get('title', expected_title or user_message)
                            except Exception as e:
                                title = ''

                            if not title:
                                title = expected_title or replace_skill_mentions_with_labels(
                                    messages[0].get('content', user_message)
                                )

                            if expected_title is not None and (not isinstance(title, str) or not title.strip()):
                                title = expected_title
                            updated = await Chats.update_chat_title_by_id(
                                metadata['chat_id'], title, expected_title=expected_title
                            )

                            if updated:
                                await event_emitter(
                                    {
                                        'type': 'chat:title',
                                        'data': title,
                                    }
                                )

                    if title == None and len(messages) == 2 and (not messages_map or len(messages_map) <= 2):
                        title = replace_skill_mentions_with_labels(messages[0].get('content', user_message))

                        await Chats.update_chat_title_by_id(metadata['chat_id'], title)

                        await event_emitter(
                            {
                                'type': 'chat:title',
                                'data': title,
                            }
                        )

                if TASKS.TAGS_GENERATION in tasks and tasks[TASKS.TAGS_GENERATION]:
                    res = await generate_chat_tags(
                        request,
                        {
                            'model': message['model'],
                            'messages': messages,
                            'chat_id': metadata['chat_id'],
                        },
                        user,
                    )

                    if res and isinstance(res, dict):
                        if len(res.get('choices', [])) == 1:
                            response_message = res.get('choices', [])[0].get('message', {})

                            tags_string = response_message.get('content') or response_message.get(
                                'reasoning_content', ''
                            )
                        else:
                            tags_string = ''

                        tags_string = tags_string[tags_string.find('{') : tags_string.rfind('}') + 1]

                        try:
                            tags = JSONCodec.loads(tags_string).get('tags', [])
                            await Chats.update_chat_tags_by_id(metadata['chat_id'], tags, user)

                            await event_emitter(
                                {
                                    'type': 'chat:tags',
                                    'data': tags,
                                }
                            )
                        except Exception as e:
                            pass

        if messages and not is_temporary_chat_id(metadata.get('chat_id')):
            await review_memory_after_turn(
                request=request,
                user=user,
                model=ctx.get('model'),
                metadata=metadata,
                form_data=form_data,
                assistant_message=ctx.get('assistant_message') or {},
                messages=messages,
            )


async def outlet_filter_handler(ctx):
    """Run outlet filters inline after chat completion.

    Replaces the separate POST /api/chat/completed round-trip.
    Persists outlet-modified content to DB and emits a chat:outlet event
    so the frontend can sync its in-memory state. Returns immediately when
    the model has no filters.

    For temp/API chats, messages are built from form_data plus ctx['assistant_message'].
    """
    request = ctx['request']
    user = ctx['user']
    model = ctx['model']
    metadata = ctx['metadata']
    event_emitter = ctx.get('event_emitter')
    event_caller = ctx.get('event_caller')

    chat_id = metadata.get('chat_id', '')
    message_id = metadata.get('message_id')

    if not chat_id and not ctx.get('assistant_message'):
        return

    if not message_id:
        message_id = output_id('msg')

    is_unsaved_chat = not is_saved_chat_id(chat_id)
    try:
        filter_functions = (
            await get_filter_functions(request, model, metadata.get('filter_ids', [])) if ENABLE_FUNCTIONS else []
        )
        model_id = model.get('id') if isinstance(model, dict) else model
        models = request.app.state.MODELS
        has_pipeline_outlet_filters = bool(
            (isinstance(model, dict) and 'pipeline' in model) or get_sorted_filters(model_id, models)
        )
        if not filter_functions and not has_pipeline_outlet_filters:
            return

        messages_map = None

        if is_unsaved_chat:
            form_messages = ctx.get('form_data', {}).get('messages', [])
            assistant_message = ctx.get('assistant_message', {})

            message_list = [
                {
                    'role': m.get('role'),
                    'content': m.get('content') or get_output_text(m.get('output')),
                }
                for m in form_messages
            ]

            if assistant_message:
                message_list.append(
                    {
                        'id': message_id,
                        'role': 'assistant',
                        **assistant_message,
                    }
                )

            if not message_list:
                return
        else:
            messages_map = await Chats.get_messages_map_by_chat_id(chat_id)
            if not messages_map:
                return

            message_list = get_message_list(messages_map, message_id)
            if not message_list:
                return

        finish_reason = (ctx.get('assistant_message') or {}).get('finish_reason')
        outlet_data = {
            'model': model_id,
            'messages': [
                {
                    'id': m.get('id'),
                    'role': m.get('role'),
                    'content': m.get('content') or get_output_text(m.get('output')),
                    'info': m.get('info'),
                    'timestamp': m.get('timestamp'),
                    # Deepcopy so in-place filter mutations do not alias messages_map's baseline
                    **({'output': copy.deepcopy(m['output'])} if m.get('output') else {}),
                    **({'usage': m['usage']} if m.get('usage') else {}),
                    **({'sources': m['sources']} if m.get('sources') else {}),
                    **({'finish_reason': finish_reason} if finish_reason and m.get('id') == message_id else {}),
                }
                for m in message_list
            ],
            'filter_ids': metadata.get('filter_ids', []),
            'chat_id': chat_id,
            'session_id': metadata.get('session_id'),
            'id': message_id,
        }

        # Pipeline outlet filters
        try:
            outlet_data = await process_pipeline_outlet_filter(request, outlet_data, user, models)
        except Exception as e:
            log.debug('Pipeline outlet filter error: %s', e)

        # Function outlet filters
        extra_params = {
            '__event_emitter__': event_emitter,
            '__event_call__': event_caller,
            '__user__': user.model_dump() if isinstance(user, UserModel) else {},
            '__metadata__': metadata,
            '__request__': request,
            '__model__': model,
        }

        if filter_functions:
            outlet_result, _ = await process_filter_functions(
                request=request,
                filter_context=None,
                filter_functions=filter_functions,
                filter_type='outlet',
                form_data=outlet_data,
                extra_params=extra_params,
            )
        else:
            outlet_result = outlet_data

        if outlet_result and outlet_result.get('messages'):
            if not is_unsaved_chat and messages_map:
                for message in outlet_result['messages']:
                    outlet_message_id = message.get('id')
                    if outlet_message_id and outlet_message_id in messages_map:
                        original_message = messages_map[outlet_message_id]
                        original_content = original_message.get('content') or get_output_text(
                            original_message.get('output')
                        )
                        message_content = message.get('content') or get_output_text(message.get('output'))
                        content_changed = original_content != message_content
                        output_changed = message.get('output') and message.get('output') != original_message.get(
                            'output'
                        )
                        if content_changed or output_changed:
                            message_update = {
                                'originalContent': original_content,
                                **({'output': message['output']} if output_changed else {}),
                            }
                            if content_changed:
                                message_update['content'] = message_content or ''
                            await Chats.upsert_message_to_chat_by_id_and_message_id(
                                chat_id,
                                outlet_message_id,
                                message_update,
                            )

            if event_emitter:
                await event_emitter(
                    {
                        'type': 'chat:outlet',
                        'data': {'messages': outlet_result['messages']},
                    }
                )
    except Exception as e:
        log.debug('Error running outlet filters: %s', e)


async def non_streaming_chat_response_handler(response, ctx):
    request = ctx['request']

    user = ctx['user']
    metadata = ctx['metadata']
    events = ctx['events']

    event_emitter = ctx['event_emitter']

    response, response_data = get_response_data(response)
    if response_data is None:
        return response

    usage = response_data.get('usage') or {}
    timings = response_data.get('timings') or {}
    if isinstance(usage, dict) and isinstance(timings, dict):
        log_model_usage(
            request,
            user,
            response_data.get('selected_model_id') or ctx['form_data']['model'],
            {**usage, **timings},
        )

    chat_id = metadata.get('chat_id') or ''
    save_to_chat = is_saved_chat_id(chat_id)
    continuing = bool(metadata.get('assistant_message_id'))

    if event_emitter:
        try:
            if 'error' in response_data:
                error = response_data.get('error')

                if isinstance(error, dict):
                    error = error.get('detail', error)
                else:
                    error = str(error)

                log.error('Provider returned error (non-streaming): %s', error)

                if save_to_chat:
                    await Chats.upsert_message_to_chat_by_id_and_message_id(
                        metadata['chat_id'],
                        metadata['message_id'],
                        {
                            'error': {'content': error},
                            'done': True,
                            'meta': {'response_ended_at': time.time()},
                        },
                    )
                if isinstance(error, str) or isinstance(error, dict):
                    await event_emitter(
                        {
                            'type': 'chat:message:error',
                            'data': {'error': {'content': error}, 'done': True},
                        }
                    )

            if 'selected_model_id' in response_data and save_to_chat:
                await Chats.upsert_message_to_chat_by_id_and_message_id(
                    metadata['chat_id'],
                    metadata['message_id'],
                    {
                        'selectedModelId': response_data['selected_model_id'],
                    },
                    touch=False,
                )

            choices = response_data.get('choices', [])
            response_output = response_data.get('output')
            content = (choices[0].get('message', {}).get('content') or '') if choices else ''
            # Native tools only run on the streaming path; still finish the turn.
            tool_calls = choices[0].get('message', {}).get('tool_calls') if choices else None

            if (continuing and 'error' not in response_data) or (
                not continuing and choices and (content or response_output or tool_calls)
            ):
                if content or response_output or tool_calls or continuing:
                    if not continuing:
                        await event_emitter(
                            {
                                'type': 'chat:completion',
                                'data': response_data,
                            }
                        )

                    title = await Chats.get_chat_title_by_id(metadata['chat_id']) if save_to_chat else ''

                    # Use output from backend if provided (OR-compliant backends),
                    # otherwise generate from response content
                    if not response_output:
                        choice_message = choices[0].get('message', {}) if choices else {}
                        reasoning_content = choice_message.get('reasoning_content') or choice_message.get('reasoning')
                        reasoning_details = get_reasoning_details(choice_message)
                        response_output = []
                        if reasoning_content or reasoning_details:
                            reasoning_item = {
                                'type': 'reasoning',
                                'id': output_id('r'),
                                'status': 'completed',
                                'start_tag': '<think>',
                                'end_tag': '</think>',
                                'attributes': {'type': 'reasoning_content'},
                                'content': (
                                    [{'type': 'output_text', 'text': reasoning_content}] if reasoning_content else []
                                ),
                                'summary': None,
                            }
                            if reasoning_details:
                                reasoning_item['reasoning_details'] = (
                                    reasoning_details if isinstance(reasoning_details, list) else [reasoning_details]
                                )
                            response_output.append(reasoning_item)
                        response_output.append(
                            {
                                'type': 'message',
                                'id': output_id('msg'),
                                'status': 'completed',
                                'role': 'assistant',
                                'content': [{'type': 'output_text', 'text': content}],
                            }
                        )

                    if continuing:
                        message = ctx.get('assistant_message') or {}
                        previous = message.get('output') or []
                        if not previous and message.get('content'):
                            previous = [
                                {
                                    'type': 'message',
                                    'id': message.get('id') or output_id('msg'),
                                    'role': 'assistant',
                                    'content': [{'type': 'output_text', 'text': message['content']}],
                                }
                            ]
                        response_output = list(response_output)
                        if (
                            previous
                            and response_output
                            and previous[-1].get('type') == response_output[0].get('type') == 'message'
                            and previous[-1].get('role', 'assistant') == 'assistant'
                            and response_output[0].get('role', 'assistant') == 'assistant'
                        ):
                            response_output[0] = {
                                **previous[-1],
                                **response_output[0],
                                'id': previous[-1].get('id'),
                                'content': [*previous[-1].get('content', []), *response_output[0].get('content', [])],
                            }
                            previous = previous[:-1]
                        response_output = previous + response_output
                        content = get_output_text(response_output)

                    response_output = [*(metadata.get('initial_output') or []), *response_output]
                    await event_emitter(
                        {
                            'type': 'chat:completion',
                            'data': {
                                **(response_data if continuing else {}),
                                'done': True,
                                'output': response_output,
                                'title': title,
                            },
                        }
                    )

                    # Save message in the database
                    usage = normalize_usage(response_data.get('usage', {}) or {})
                    if continuing:
                        usage = merge_usage(
                            (ctx.get('assistant_message') or {}).get('usage'), usage or {'incomplete': True}
                        )
                    finish_reason = choices[0].get('finish_reason') if choices else None

                    if save_to_chat:
                        await Chats.upsert_message_to_chat_by_id_and_message_id(
                            metadata['chat_id'],
                            metadata['message_id'],
                            {
                                'done': True,
                                'meta': {'response_ended_at': time.time()},
                                'role': 'assistant',
                                'output': response_output,
                                **({'usage': usage} if usage else {}),
                            },
                        )

                    await publish_chat_finished_event(request, user, metadata, title, content, response_output)

                    ctx['assistant_message'] = {
                        'content': content,
                        'output': response_output,
                        **({'usage': usage} if usage else {}),
                        **({'finish_reason': finish_reason} if finish_reason else {}),
                    }
                    await outlet_filter_handler(ctx)
                    await background_tasks_handler(ctx)

            response = build_response_object(response, merge_events_into_response(response_data, events))
        except Exception as e:
            log.debug('Error occurred while processing request: %s', e)
            await publish_chat_failed_event(request, user, metadata, str(e))
            pass

        return response

    choices = response_data.get('choices', [])
    output = response_data.get('output')
    content = choices[0].get('message', {}).get('content') if choices else ''
    if ENABLE_API_OUTLET_FILTERS and (content or output):
        usage = normalize_usage(response_data.get('usage', {}) or {})
        finish_reason = choices[0].get('finish_reason') if choices else None
        ctx['assistant_message'] = {
            **({'content': content} if content else {}),
            **({'output': output} if output else {}),
            **({'usage': usage} if usage else {}),
            **({'finish_reason': finish_reason} if finish_reason else {}),
        }
        await outlet_filter_handler(ctx)

    if isinstance(response, dict):
        response = merge_events_into_response(response_data, events)

    return response


async def streaming_chat_response_handler(response, ctx):
    request = ctx['request']

    form_data = ctx['form_data']

    user = ctx['user']
    model = ctx['model']

    metadata = ctx['metadata']
    events = ctx['events']

    event_emitter = ctx['event_emitter']
    event_caller = ctx['event_caller']
    chat_id = metadata.get('chat_id') or ''
    save_to_chat = is_saved_chat_id(chat_id)
    continuing = bool(metadata.get('assistant_message_id'))

    extra_params = {
        '__event_emitter__': event_emitter,
        '__event_call__': event_caller,
        '__user__': user.model_dump() if isinstance(user, UserModel) else {},
        '__metadata__': metadata,
        '__oauth_token__': await get_system_oauth_token(request, user),
        '__request__': request,
        '__model__': model,
        '__chat_id__': metadata.get('chat_id'),
        '__message_id__': metadata.get('message_id'),
    }

    filter_functions = (
        await get_filter_functions(request, model, metadata.get('filter_ids', [])) if ENABLE_FUNCTIONS else []
    )

    # Standard streaming response handler
    # event_caller is optional — only needed for direct (client-side) tools
    # and pyodide code interpreter. Server-side tools work without it.
    if event_emitter:
        task_id = str(uuid4())  # Create a unique task ID.

        # Handle as a background task
        async def response_handler(response, events):
            filter_context = FilterContext()
            tag_scan_positions = {}
            tag_boundary_positions = {}
            response_stream_task_id = metadata.get('task_id') or metadata.get('message_id')

            def tag_output_handler(content_type, tags, output):
                """
                Detect special tags (reasoning, solution, code_interpreter) in streaming
                content and create corresponding OR-aligned output items directly.
                Operates on output items instead of content_blocks.

                Uses the text from the output items themselves for tag detection,
                eliminating state divergence between accumulated content and items.

                Mutates output in place; returns the rewritten output (or None
                if no tag was consumed) and whether a block ended.
                """
                end_flag = False

                def extract_attributes(tag_content):
                    """Extract attributes from a tag if they exist."""
                    attributes = {}
                    if not tag_content:
                        return attributes
                    matches = re.findall(r'(\w+)\s*=\s*"([^"]+)"', tag_content)
                    for key, value in matches:
                        attributes[key] = value
                    return attributes

                def get_last_text(out):
                    """Get text from last message item, or empty string."""
                    if out and out[-1].get('type') == 'message':
                        parts = out[-1].get('content', [])
                        if parts and parts[-1].get('type') == 'output_text':
                            return parts[-1].get('text', '')
                    return ''

                def set_last_text(out, text):
                    """Set text on last message item's output_text."""
                    if out and out[-1].get('type') == 'message':
                        parts = out[-1].get('content', [])
                        if parts and parts[-1].get('type') == 'output_text':
                            parts[-1]['text'] = text

                def get_scanned_length(item, text):
                    item_id = item.get('id')
                    if not item_id:
                        return 0

                    scanned_length = tag_scan_positions.get((item_id, content_type), 0)
                    return scanned_length if scanned_length <= len(text) else 0

                def save_scanned_length(item, text):
                    item_id = item.get('id')
                    if item_id:
                        tag_scan_positions[(item_id, content_type)] = len(text)

                def clear_scanned_length(item):
                    item_id = item.get('id')
                    if item_id:
                        tag_scan_positions.pop((item_id, content_type), None)
                        tag_boundary_positions.pop((item_id, content_type), None)

                def get_tag_boundaries(item, text, scanned_length):
                    """Index of the last '<', and of the last '>' or newline, before scanned_length."""
                    key = (item.get('id'), content_type)
                    scanned, last_open, last_boundary = tag_boundary_positions.get(key, (0, -1, -1))
                    if scanned > scanned_length:  # the item was rewritten, so the cached positions are stale
                        scanned, last_open, last_boundary = 0, -1, -1

                    if scanned < scanned_length:
                        # only text added since the last call can move either position
                        open_tag = text.rfind('<', scanned, scanned_length)
                        if open_tag != -1:
                            last_open = open_tag
                        boundary = max(
                            text.rfind('>', scanned, scanned_length),
                            text.rfind('\n', scanned, scanned_length),
                        )
                        if boundary != -1:
                            last_boundary = boundary
                        tag_boundary_positions[key] = (scanned_length, last_open, last_boundary)

                    return last_open, last_boundary

                # Map content_type to output item type
                output_type_map = {
                    'reasoning': 'reasoning',
                    'solution': 'message',  # solution tags just produce text
                    'code_interpreter': 'open_webui:code_interpreter',
                }
                output_item_type = output_type_map.get(content_type, content_type)

                last_type = output[-1].get('type', '') if output else ''

                if last_type == 'message' and output[-1].get('_tag_type') != content_type:
                    # Use the output item's own text for tag detection
                    item = output[-1]
                    item_text = get_last_text(output)
                    scanned_length = get_scanned_length(item, item_text)
                    max_start_tag_length = max((len(start_tag) for start_tag, _ in tags), default=1)
                    search_start = max(0, scanned_length - max_start_tag_length + 1)

                    if scanned_length and any(
                        start_tag.startswith('<') and start_tag.endswith('>') for start_tag, _ in tags
                    ):
                        open_tag_start, last_tag_boundary = get_tag_boundaries(item, item_text, scanned_length)
                        if open_tag_start > last_tag_boundary:
                            search_start = min(search_start, open_tag_start)

                    for start_tag, end_tag in tags:
                        match = re.compile(_start_tag_pattern(start_tag)).search(item_text, search_start)
                        if match:
                            clear_scanned_length(item)
                            try:
                                attr_content = match.group(1) if match.group(1) else ''
                            except Exception:
                                attr_content = ''

                            attributes = extract_attributes(attr_content)

                            before_tag = item_text[: match.start()]
                            after_tag = item_text[match.end() :]

                            # Keep only text before the tag in the message
                            set_last_text(output, before_tag)

                            if not before_tag.strip():
                                # Remove empty message item
                                if output and output[-1].get('type') == 'message':
                                    output.pop()

                            # Append the new output item
                            if output_item_type == 'reasoning':
                                output.append(
                                    {
                                        'type': 'reasoning',
                                        'id': output_id('r'),
                                        'status': 'in_progress',
                                        'start_tag': start_tag,
                                        'end_tag': end_tag,
                                        'attributes': attributes,
                                        'content': [],
                                        'summary': None,
                                        'started_at': time.time(),
                                    }
                                )
                            elif output_item_type == 'open_webui:code_interpreter':
                                output.append(
                                    {
                                        'type': 'open_webui:code_interpreter',
                                        'id': output_id('ci'),
                                        'status': 'in_progress',
                                        'start_tag': start_tag,
                                        'end_tag': end_tag,
                                        'attributes': attributes,
                                        'lang': attributes.get('lang', 'python'),
                                        'code': '',
                                        'output': None,
                                        'started_at': time.time(),
                                    }
                                )
                            else:
                                # solution or other text-producing tag
                                output.append(
                                    {
                                        'type': 'message',
                                        'id': output_id('msg'),
                                        'status': 'in_progress',
                                        'role': 'assistant',
                                        'content': [{'type': 'output_text', 'text': ''}],
                                        '_tag_type': content_type,
                                        'start_tag': start_tag,
                                        'end_tag': end_tag,
                                        'attributes': attributes,
                                        'started_at': time.time(),
                                    }
                                )

                            if after_tag:
                                # Set the after_tag content on the new item
                                if output_item_type == 'reasoning':
                                    output[-1]['content'] = [{'type': 'output_text', 'text': after_tag}]
                                elif output_item_type == 'open_webui:code_interpreter':
                                    output[-1]['code'] = after_tag
                                else:
                                    set_last_text(output, after_tag)

                                _, recursive_end = tag_output_handler(content_type, tags, output)
                                if recursive_end:
                                    end_flag = True

                            return output, end_flag
                    else:
                        save_scanned_length(item, item_text)

                elif (
                    (last_type == 'reasoning' and content_type == 'reasoning')
                    or (last_type == 'open_webui:code_interpreter' and content_type == 'code_interpreter')
                    or (last_type == 'message' and output[-1].get('_tag_type') == content_type)
                ):
                    item = output[-1]
                    start_tag = item.get('start_tag', '')
                    end_tag = item.get('end_tag', '')

                    # Get the block content from the item itself
                    if last_type == 'reasoning':
                        parts = item.get('content', [])
                        block_content = ''
                        if parts and parts[-1].get('type') == 'output_text':
                            block_content = parts[-1].get('text', '')
                    elif last_type == 'open_webui:code_interpreter':
                        block_content = item.get('code', '')
                    else:
                        block_content = get_last_text(output)

                    scanned_length = get_scanned_length(item, block_content)
                    end_tag_search_start = max(0, scanned_length - max(len(end_tag), 1) + 1)

                    if block_content.find(end_tag, end_tag_search_start) != -1:
                        clear_scanned_length(item)
                        end_flag = True

                        # Strip start and end tags from content
                        start_tag_pattern = _start_tag_pattern(start_tag)
                        block_content = re.sub(start_tag_pattern, '', block_content)

                        end_tag_pattern = rf'{re.escape(end_tag)}'
                        end_tag_regex = re.compile(end_tag_pattern, re.DOTALL)
                        split_content = end_tag_regex.split(block_content, maxsplit=1)

                        block_content = split_content[0].strip() if split_content else ''
                        leftover_content = split_content[1].lstrip() if len(split_content) > 1 else ''

                        if block_content:
                            # Update the item with final content
                            if last_type == 'reasoning':
                                item['content'] = [{'type': 'output_text', 'text': block_content}]
                                item['ended_at'] = time.time()
                                item['duration'] = int(item['ended_at'] - item['started_at'])
                                item['status'] = 'completed'
                            elif last_type == 'open_webui:code_interpreter':
                                item['code'] = block_content
                                item['ended_at'] = time.time()
                                item['duration'] = int(item['ended_at'] - item['started_at'])
                            else:
                                set_last_text(output, block_content)
                                item['ended_at'] = time.time()

                            # Reset by appending a new message item for leftover
                            output.append(
                                {
                                    'type': 'message',
                                    'id': output_id('msg'),
                                    'status': 'in_progress',
                                    'role': 'assistant',
                                    'content': [
                                        {
                                            'type': 'output_text',
                                            'text': leftover_content,
                                        }
                                    ],
                                }
                            )
                        else:
                            # Remove the block if content is empty
                            output.pop()
                            output.append(
                                {
                                    'type': 'message',
                                    'id': output_id('msg'),
                                    'status': 'in_progress',
                                    'role': 'assistant',
                                    'content': [
                                        {
                                            'type': 'output_text',
                                            'text': leftover_content,
                                        }
                                    ],
                                }
                            )
                        return output, end_flag
                    else:
                        save_scanned_length(item, block_content)

                return None, False

            message = (
                ctx.get('assistant_message')
                if continuing
                else (
                    await Chats.get_message_by_id_and_message_id(metadata['chat_id'], metadata['message_id'])
                    if save_to_chat
                    else None
                )
            )

            tool_calls = []

            last_assistant_message = None
            try:
                if form_data['messages'][-1]['role'] == 'assistant':
                    last_assistant_message = get_last_assistant_message(form_data['messages'])
            except Exception as e:
                pass

            initial_content = (
                message.get('content', '') if message else last_assistant_message if last_assistant_message else ''
            )
            content_parts = [initial_content] if initial_content else []

            # Initialize output: use existing from message if continuing, else create new
            existing_output = message.get('output') if message else None
            prior_output = metadata.get('initial_output') or []
            if existing_output and metadata.get('assistant_message_id'):
                prior_output = list(existing_output)
                if (
                    prior_output
                    and prior_output[-1].get('type') == 'message'
                    and prior_output[-1].get('status') == 'in_progress'
                ):
                    msg_parts = prior_output[-1].get('content', [])
                    if not msg_parts or (len(msg_parts) == 1 and not msg_parts[0].get('text', '')):
                        prior_output.pop()
                output = []
                content_parts = []
            elif existing_output:
                output = existing_output
            else:
                # Only create an initial message item if there is content to initialize with
                if initial_content:
                    output = [
                        {
                            'type': 'message',
                            'id': output_id('msg'),
                            'status': 'in_progress',
                            'role': 'assistant',
                            'content': [{'type': 'output_text', 'text': initial_content}],
                        }
                    ]
                    if not continuing:
                        # the client needs this item before the first delta arrives
                        await event_emitter({'type': 'chat:completion', 'data': {'output': output}})
                else:
                    output = []

            if continuing and not prior_output:
                # Keep the prefix separate from provider output indices and tool follow-up messages.
                prior_output, output = output, []
                content_parts = []

            usage = normalize_usage((message or {}).get('usage') or {}) if existing_output else None
            finish_reason = None
            last_response_id = None

            def is_message_continuation() -> bool:
                return bool(
                    continuing
                    and prior_output
                    and output
                    and prior_output[-1].get('type') == output[0].get('type') == 'message'
                    and prior_output[-1].get('role', 'assistant') == 'assistant'
                    and output[0].get('role', 'assistant') == 'assistant'
                )

            def full_output():
                for item in output:
                    set_tool_call_metadata(item, metadata.get('tools', {}))
                if is_message_continuation():
                    return [
                        *prior_output[:-1],
                        {
                            **prior_output[-1],
                            **output[0],
                            'id': prior_output[-1].get('id'),
                            'content': [*prior_output[-1].get('content', []), *output[0].get('content', [])],
                        },
                        *output[1:],
                    ]
                return prior_output + output if prior_output else output

            stream_size = 0
            snapshot_size = 0

            async def save_current_response_stream():
                nonlocal stream_size, snapshot_size
                stream_size = snapshot_size = await save_response_stream(
                    request.app.state.redis,
                    response_stream_task_id,
                    chat_id,
                    metadata.get('message_id'),
                    full_output(),
                )

            async def save_response_delta(delta):
                nonlocal stream_size
                if not stream_size or not request.app.state.redis:
                    return
                # Code blocks write to `code`, not the event's text field.
                if not isinstance(delta.get('delta'), str) or any(
                    item.get('type') == 'open_webui:code_interpreter' for item in output
                ):
                    stream_size = 0
                    return
                line = dumps_bytes(delta) + b'\n'
                if stream_size + len(line) <= snapshot_size * 2 and await append_response_stream(
                    request.app.state.redis, response_stream_task_id, line, stream_size
                ):
                    stream_size += len(line)
                else:
                    # Snapshot after the whole provider chunk, which may update several items.
                    stream_size = 0

            def get_message_error_content(error):
                if isinstance(error, HTTPException):
                    error = error.detail
                elif isinstance(error, dict):
                    error = error.get('detail', error)
                else:
                    error = str(error)

                return error if isinstance(error, (str, dict)) else str(error)

            async def emit_message_error(error_content):
                if save_to_chat:
                    await Chats.upsert_message_to_chat_by_id_and_message_id(
                        metadata['chat_id'],
                        metadata['message_id'],
                        {'error': {'content': error_content}},
                    )
                await event_emitter(
                    {
                        'type': 'chat:message:error',
                        'data': {'error': {'content': error_content}},
                    }
                )

            reasoning_tags_param = metadata.get('params', {}).get('reasoning_tags')
            DETECT_REASONING_TAGS = reasoning_tags_param is not False

            # Legacy tool-calling only: native FC gets execute_code as a builtin tool.
            # Same five authz gates as utils/tools.py get_builtin_tools.
            features = metadata.get('features', {}) or {}
            model_capabilities = model.get('info', {}).get('meta', {}).get('capabilities') or {}
            builtin_tools_meta = model.get('info', {}).get('meta', {}).get('builtinTools', {})
            DETECT_CODE_INTERPRETER = (
                metadata.get('params', {}).get('function_calling') == 'legacy'
                and bool(features.get('code_interpreter'))
                and builtin_tools_meta.get('code_interpreter', True)
                and await Config.get('code_interpreter.enable')
                and model_capabilities.get('code_interpreter', True)
                and (
                    getattr(user, 'role', None) == 'admin'
                    or await has_permission(
                        getattr(user, 'id', ''),
                        'features.code_interpreter',
                        await Config.get('user.permissions'),
                    )
                )
            )

            reasoning_tags = []
            if DETECT_REASONING_TAGS:
                if isinstance(reasoning_tags_param, list) and len(reasoning_tags_param) == 2:
                    reasoning_tags = [(reasoning_tags_param[0], reasoning_tags_param[1])]
                else:
                    reasoning_tags = DEFAULT_REASONING_TAGS

            try:
                # deltas of a continued reply assume the client already holds the prior output
                if prior_output:
                    await event_emitter({'type': 'chat:completion', 'data': {'output': full_output()}})

                for event in events:
                    await event_emitter(
                        {
                            'type': 'chat:completion',
                            'data': event,
                        }
                    )

                    # Save message in the database
                    if save_to_chat:
                        await Chats.upsert_message_to_chat_by_id_and_message_id(
                            metadata['chat_id'],
                            metadata['message_id'],
                            {
                                **event,
                            },
                        )

                async def stream_body_handler(response, current_form_data):
                    nonlocal usage
                    nonlocal current_usage
                    nonlocal finish_reason
                    nonlocal output
                    nonlocal prior_output
                    nonlocal last_response_id
                    nonlocal stream_size

                    response_tool_calls = []
                    finish_reason = None
                    prior_usage = usage
                    current_usage = {}
                    form_data['model'] = current_form_data['model']

                    delta_count = 0
                    delta_chunk_size = max(
                        CHAT_RESPONSE_STREAM_DELTA_CHUNK_SIZE,
                        int(metadata.get('params', {}).get('stream_delta_chunk_size') or 1),
                    )
                    last_delta_data = None
                    last_delta_type = None
                    last_delta_key = None

                    def get_response_data_with_full_output_index(response_data: dict):
                        if prior_output and isinstance(response_data.get('output_index'), int):
                            if is_message_continuation():
                                output_index = response_data['output_index']
                                content_offset = len(prior_output[-1].get('content', [])) if output_index == 0 else 0
                                return {
                                    **response_data,
                                    'output_index': output_index + len(prior_output) - 1,
                                    'content_index': response_data.get('content_index', 0) + content_offset,
                                }
                            return {
                                **response_data,
                                'output_index': response_data['output_index'] + len(prior_output),
                            }
                        if prior_output and isinstance(response_data.get('response'), dict):
                            # response.output is this round only; the response.completed reducer drops earlier rounds
                            return {
                                **response_data,
                                'response': {
                                    key: value for key, value in response_data['response'].items() if key != 'output'
                                },
                            }
                        return response_data

                    async def flush_pending_delta_data(threshold: int = 0):
                        nonlocal delta_count
                        nonlocal last_delta_data
                        nonlocal last_delta_type
                        nonlocal last_delta_key

                        if delta_count >= threshold and last_delta_data:
                            await event_emitter({'type': 'response:completion', 'data': last_delta_data})
                            delta_count = 0
                            last_delta_data = None
                            last_delta_type = None
                            last_delta_key = None

                    async def queue_pending_delta_data(delta_data: dict, delta_type: str):
                        nonlocal delta_count
                        nonlocal last_delta_data
                        nonlocal last_delta_type
                        nonlocal last_delta_key

                        delta_data = get_response_data_with_full_output_index(delta_data)
                        await save_response_delta(delta_data)
                        delta_key = get_response_delta_key(delta_data)
                        if (
                            last_delta_data
                            and last_delta_key == delta_key
                            and isinstance(last_delta_data.get('delta'), str)
                            and isinstance(delta_data.get('delta'), str)
                        ):
                            append_to_text_field(last_delta_data, 'delta', delta_data['delta'])
                            delta_count += 1
                        else:
                            if last_delta_data and (last_delta_type != delta_type or last_delta_key != delta_key):
                                await flush_pending_delta_data()

                            delta_count += 1
                            last_delta_data = delta_data
                            last_delta_type = delta_type
                            last_delta_key = delta_key

                        if delta_count >= delta_chunk_size:
                            await flush_pending_delta_data(delta_chunk_size)

                    async def emit_response_completion_event(response_data: dict):
                        nonlocal stream_size
                        for item in [
                            response_data.get('item') or {},
                            *((response_data.get('response') or {}).get('output') or []),
                        ]:
                            set_tool_call_metadata(item, metadata.get('tools', {}))
                        if response_data.get('type', '').endswith('.delta'):
                            await queue_pending_delta_data(
                                response_data,
                                response_data.get('type', 'response.delta'),
                            )
                            return

                        response_data = get_response_data_with_full_output_index(response_data)
                        await flush_pending_delta_data()
                        await event_emitter(
                            {
                                'type': 'chat:completion' if continuing else 'response:completion',
                                'data': {'output': full_output()}
                                if continuing
                                else get_response_completion_event_data(response_data),
                            }
                        )
                        stream_size = 0

                    filter_extra_params = {'__body__': current_form_data, **extra_params} if filter_functions else None

                    async for line in response.body_iterator:
                        line = line.decode('utf-8', 'replace') if isinstance(line, bytes) else line
                        data = line

                        # Skip empty lines
                        if not data or data.isspace():
                            continue

                        # "data:" is the prefix for each event
                        if not data.startswith('data:'):
                            # Some upstreams return plain JSON error lines in a streaming response
                            # (without SSE `data:` prefix). Try to normalize these into standard
                            # error events so frontend and DB paths still receive them.
                            try:
                                raw_obj = JSONCodec.loads(data)
                                raw_error = raw_obj.get('error') if isinstance(raw_obj, dict) else None
                                if raw_error:
                                    if save_to_chat:
                                        try:
                                            await Chats.upsert_message_to_chat_by_id_and_message_id(
                                                metadata['chat_id'],
                                                metadata['message_id'],
                                                {
                                                    'error': {'content': raw_error},
                                                },
                                            )
                                        except Exception:
                                            pass
                                    await event_emitter({'type': 'chat:completion', 'data': {'error': raw_error}})
                            except Exception:
                                pass
                            continue

                        # Remove the "data:" prefix
                        data = data[5:].strip()

                        try:
                            data = JSONCodec.loads(data)

                            if filter_functions:
                                data, _ = await process_filter_functions(
                                    request=request,
                                    filter_context=filter_context,
                                    filter_functions=filter_functions,
                                    filter_type='stream',
                                    form_data=data,
                                    extra_params=filter_extra_params,
                                )

                            if data:
                                if 'event' in data:
                                    await event_emitter(data.get('event', {}))

                                if 'selected_model_id' in data:
                                    form_data['model'] = data['selected_model_id']
                                    if save_to_chat:
                                        await Chats.upsert_message_to_chat_by_id_and_message_id(
                                            metadata['chat_id'],
                                            metadata['message_id'],
                                            {
                                                'selectedModelId': form_data['model'],
                                            },
                                            touch=False,
                                        )
                                    await event_emitter(
                                        {
                                            'type': 'chat:completion',
                                            'data': data,
                                        }
                                    )
                                # Check for Responses API events
                                elif data.get('type', '').startswith('response.') or data.get('type', '') == 'error':
                                    response_data_type = data.get('type', '')
                                    if response_data_type in {'response.incomplete', 'response.failed', 'error'}:
                                        finish_reason = 'incomplete'
                                    elif response_data_type == 'response.completed':
                                        finish_reason = data.get('response', {}).get('status', 'completed')
                                    response_data_is_delta = response_data_type.endswith('.delta')
                                    output, response_metadata = handle_responses_streaming_event(data, output)

                                    if not response_data_is_delta:
                                        await flush_pending_delta_data()

                                    # Emit citation sources from finalized output items
                                    # (mirrors Chat Completions annotation handling at delta level)
                                    if response_data_type == 'response.output_item.done':
                                        item = data.get('item', {})
                                        if item.get('type') == 'message':
                                            for part in item.get('content', []):
                                                for annotation in part.get('annotations', []):
                                                    if annotation.get('type') == 'url_citation':
                                                        # Handle both flat (Responses API) and nested (Chat Completions) formats
                                                        url_citation = annotation.get('url_citation', annotation)

                                                        url = url_citation.get('url', '')
                                                        title = url_citation.get('title', url)

                                                        if url:
                                                            await event_emitter(
                                                                {
                                                                    'type': 'source',
                                                                    'data': {
                                                                        'source': {
                                                                            'name': title,
                                                                            'url': url,
                                                                        },
                                                                        'document': [title],
                                                                        'metadata': [
                                                                            {
                                                                                'source': url,
                                                                                'name': title,
                                                                            }
                                                                        ],
                                                                    },
                                                                }
                                                            )

                                    # Merge metadata without 'done'; the outer middleware signals completion after tool calls.
                                    if response_metadata:
                                        if ENABLE_RESPONSES_API_STATEFUL:
                                            response_id = response_metadata.pop('response_id', None)
                                            if response_id:
                                                last_response_id = response_id

                                        # Normalize and capture usage for DB persistence
                                        if response_metadata.get('usage'):
                                            current_usage = update_usage_snapshot(
                                                current_usage, response_metadata['usage']
                                            )
                                            usage = merge_usage(prior_usage, current_usage)
                                            response_metadata['usage'] = usage

                                        if response_metadata.get('error'):
                                            log.error(
                                                'Provider returned error (streaming): %s', response_metadata['error']
                                            )
                                            if save_to_chat:
                                                try:
                                                    await Chats.upsert_message_to_chat_by_id_and_message_id(
                                                        metadata['chat_id'],
                                                        metadata['message_id'],
                                                        {
                                                            'error': {'content': response_metadata['error']},
                                                        },
                                                    )
                                                except Exception:
                                                    pass
                                            await event_emitter(
                                                {
                                                    'type': 'chat:completion',
                                                    'data': {'error': response_metadata['error']},
                                                }
                                            )

                                    await emit_response_completion_event(data)

                                    if response_metadata and response_metadata.get('usage'):
                                        await event_emitter(
                                            {
                                                'type': 'chat:completion',
                                                'data': {'usage': usage},
                                            }
                                        )
                                    continue
                                else:
                                    choices = data.get('choices', [])

                                    # Normalize usage data to standard format
                                    raw_usage = data.get('usage', {}) or {}
                                    raw_usage.update(data.get('timings', {}))  # llama.cpp
                                    if raw_usage:
                                        current_usage = update_usage_snapshot(current_usage, raw_usage)
                                        usage = merge_usage(prior_usage, current_usage)
                                        await event_emitter(
                                            {
                                                'type': 'chat:completion',
                                                'data': {
                                                    'usage': usage,
                                                },
                                            }
                                        )

                                    if not choices:
                                        error = data.get('error', {})
                                        if error:
                                            log.error('Provider returned error (streaming): %s', error)
                                            if save_to_chat:
                                                try:
                                                    await Chats.upsert_message_to_chat_by_id_and_message_id(
                                                        metadata['chat_id'],
                                                        metadata['message_id'],
                                                        {
                                                            'error': {'content': error},
                                                        },
                                                    )
                                                except Exception:
                                                    pass
                                            await event_emitter(
                                                {
                                                    'type': 'chat:completion',
                                                    'data': {
                                                        'error': error,
                                                    },
                                                }
                                            )
                                        continue

                                    if choices[0].get('finish_reason'):
                                        finish_reason = choices[0]['finish_reason']

                                    delta = choices[0].get('delta', {})
                                    delta_type = 'content'

                                    # Handle delta annotations
                                    annotations = delta.get('annotations')
                                    if annotations:
                                        for annotation in annotations:
                                            if (
                                                annotation.get('type') == 'url_citation'
                                                and 'url_citation' in annotation
                                            ):
                                                url_citation = annotation['url_citation']

                                                url = url_citation.get('url', '')
                                                title = url_citation.get('title', url)

                                                await event_emitter(
                                                    {
                                                        'type': 'source',
                                                        'data': {
                                                            'source': {
                                                                'name': title,
                                                                'url': url,
                                                            },
                                                            'document': [title],
                                                            'metadata': [
                                                                {
                                                                    'source': url,
                                                                    'name': title,
                                                                }
                                                            ],
                                                        },
                                                    }
                                                )

                                    delta_tool_calls = delta.get('tool_calls', None)
                                    if delta_tool_calls:
                                        for delta_tool_call in delta_tool_calls:
                                            tool_call_index = delta_tool_call.get('index')

                                            if tool_call_index is not None:
                                                # Check if the tool call already exists
                                                current_response_tool_call = None
                                                for response_tool_call in response_tool_calls:
                                                    if response_tool_call.get('index') == tool_call_index:
                                                        current_response_tool_call = response_tool_call
                                                        break

                                                if current_response_tool_call is None:
                                                    # Add the new tool call
                                                    delta_tool_call.setdefault('function', {})
                                                    if delta_tool_call['function'].get('name') is None:
                                                        delta_tool_call['function']['name'] = ''
                                                    delta_tool_call['id'] = delta_tool_call.get('id') or output_id('fc')
                                                    # Some providers restart tool call ids on every round
                                                    if any(
                                                        item.get('type') == 'function_call'
                                                        and item.get('call_id') == delta_tool_call['id']
                                                        for item in full_output()
                                                    ):
                                                        delta_tool_call['id'] = output_id('fc')
                                                    delta_arguments = delta_tool_call['function'].get('arguments')
                                                    if not isinstance(delta_arguments, str):
                                                        delta_tool_call['function']['arguments'] = (
                                                            ''
                                                            if delta_arguments is None
                                                            else JSONCodec.dumps(delta_arguments)
                                                        )
                                                    response_tool_calls.append(delta_tool_call)
                                                else:
                                                    # Update the existing tool call
                                                    delta_name = delta_tool_call.get('function', {}).get('name')
                                                    delta_arguments = delta_tool_call.get('function', {}).get(
                                                        'arguments'
                                                    )

                                                    if delta_name:
                                                        current_response_tool_call['function']['name'] = delta_name

                                                    if delta_arguments is not None:
                                                        if not isinstance(delta_arguments, str):
                                                            delta_arguments = JSONCodec.dumps(delta_arguments)
                                                        current_response_tool_call.setdefault('function', {})
                                                        if not isinstance(
                                                            current_response_tool_call['function'].get('arguments'),
                                                            str,
                                                        ):
                                                            current_response_tool_call['function']['arguments'] = ''
                                                        append_to_text_field(
                                                            current_response_tool_call['function'],
                                                            'arguments',
                                                            delta_arguments,
                                                        )

                                        # Emit pending tool calls in real-time as Responses events.
                                        if response_tool_calls:
                                            output_by_call_id = {
                                                item.get('call_id'): (idx, item)
                                                for idx, item in enumerate(output)
                                                if item.get('type') == 'function_call'
                                            }

                                            tool_name_changed = False
                                            for tc in response_tool_calls:
                                                call_id = tc.get('id') or output_id('fc')
                                                tc['id'] = call_id
                                                func = tc.get('function', {})
                                                if call_id in output_by_call_id:
                                                    output_index, item = output_by_call_id[call_id]
                                                    if func.get('name', item.get('name', '')) != item.get('name'):
                                                        stream_size = 0
                                                        tool_name_changed = True
                                                    item['name'] = func.get('name', item.get('name', ''))
                                                    item['arguments'] = func.get('arguments', item.get('arguments', ''))
                                                    item['status'] = 'in_progress'
                                                else:
                                                    output_index = len(output)
                                                    item = {
                                                        'type': 'function_call',
                                                        'id': call_id,
                                                        'call_id': call_id,
                                                        'name': func.get('name', ''),
                                                        'arguments': '',
                                                        'status': 'in_progress',
                                                    }
                                                    output.append(item)
                                                    output_by_call_id[call_id] = (output_index, item)
                                                    await emit_response_completion_event(
                                                        {
                                                            'type': 'response.output_item.added',
                                                            'output_index': output_index,
                                                            'item': item.copy(),
                                                        }
                                                    )
                                                    item['arguments'] = func.get('arguments', '')

                                            for delta_tool_call in delta_tool_calls:
                                                tool_call_index = delta_tool_call.get('index')
                                                current_response_tool_call = next(
                                                    (
                                                        tc
                                                        for tc in response_tool_calls
                                                        if tc.get('index') == tool_call_index
                                                    ),
                                                    None,
                                                )
                                                if not current_response_tool_call:
                                                    continue
                                                call_id = current_response_tool_call.get('id')
                                                output_index, _ = output_by_call_id.get(call_id, (len(output) - 1, {}))
                                                delta_arguments = delta_tool_call.get('function', {}).get('arguments')
                                                if delta_arguments is not None:
                                                    if not isinstance(delta_arguments, str):
                                                        delta_arguments = JSONCodec.dumps(delta_arguments)
                                                    await emit_response_completion_event(
                                                        {
                                                            'type': 'response.function_call_arguments.delta',
                                                            'item_id': call_id,
                                                            'output_index': output_index,
                                                            'delta': delta_arguments,
                                                        }
                                                    )

                                            if tool_name_changed:
                                                # A late tool name can select a native renderer. Flush deltas
                                                # before the full snapshot so arguments are not appended twice.
                                                await flush_pending_delta_data()
                                                await event_emitter(
                                                    {
                                                        'type': 'chat:completion',
                                                        'data': {'output': full_output(), 'flush': True},
                                                    }
                                                )

                                            data = None
                                            delta_type = 'tool_call'

                                    delta_images = delta.get('images')
                                    image_urls = (
                                        await get_image_urls(delta_images, request, metadata, user)
                                        if delta_images
                                        else []
                                    )
                                    if image_urls:
                                        image_file_list = [{'type': 'image', 'url': url} for url in image_urls]
                                        message_files = image_file_list
                                        if save_to_chat:
                                            message_files = await Chats.add_message_files_by_id_and_message_id(
                                                metadata['chat_id'],
                                                metadata['message_id'],
                                                image_file_list,
                                            )
                                            if message_files is None:
                                                message_files = image_file_list

                                        # 'files' would make the emitter save these again.
                                        await event_emitter(
                                            {
                                                'type': 'chat:message:files',
                                                'data': {'files': message_files},
                                            }
                                        )

                                    # content and reasoning deltas are raw JSON: a stream filter can make them any type
                                    value = delta.get('content')
                                    if value and not isinstance(value, str):
                                        value = f'{value}'

                                    reasoning_content = (
                                        delta.get('reasoning_content')
                                        or delta.get('reasoning')
                                        or delta.get('thinking')
                                    )
                                    if reasoning_content and not isinstance(reasoning_content, str):
                                        reasoning_content = f'{reasoning_content}'
                                    reasoning_details = get_reasoning_details(delta)
                                    reasoning_detail_items = (
                                        [item for item in reasoning_details if isinstance(item, dict)]
                                        if isinstance(reasoning_details, list)
                                        else [reasoning_details]
                                        if isinstance(reasoning_details, dict)
                                        else []
                                    )
                                    existing_reasoning_item = next(
                                        (item for item in reversed(output) if item.get('type') == 'reasoning'),
                                        None,
                                    )
                                    message_index = next(
                                        (i for i, item in enumerate(output) if item.get('type') == 'message'),
                                        None,
                                    )
                                    if reasoning_content or (
                                        reasoning_detail_items
                                        and (
                                            existing_reasoning_item
                                            or any(
                                                item.get('text') or item.get('summary') or item.get('data')
                                                for item in reasoning_detail_items
                                            )
                                        )
                                    ):
                                        reasoning_item = (
                                            existing_reasoning_item
                                            if (reasoning_detail_items and not reasoning_content)
                                            or message_index is not None
                                            else None
                                        )

                                        if reasoning_item is None:
                                            if not output or output[-1].get('type') != 'reasoning':
                                                reasoning_item = {
                                                    'type': 'reasoning',
                                                    'id': output_id('r'),
                                                    'status': 'in_progress',
                                                    'start_tag': '<think>',
                                                    'end_tag': '</think>',
                                                    'attributes': {'type': 'reasoning_content'},
                                                    'content': [],
                                                    'summary': None,
                                                    'started_at': time.time(),
                                                }
                                                if message_index is not None:
                                                    reasoning_item['ended_at'] = time.time()
                                                    reasoning_item['duration'] = 0
                                                    reasoning_item['status'] = 'completed'
                                                    output.insert(message_index, reasoning_item)
                                                else:
                                                    output.append(reasoning_item)
                                                await emit_response_completion_event(
                                                    {
                                                        'type': 'response.output_item.added',
                                                        'output_index': output.index(reasoning_item),
                                                        'item': reasoning_item.copy(),
                                                    }
                                                )
                                            else:
                                                reasoning_item = output[-1]

                                        if reasoning_content:
                                            # Append to reasoning content
                                            parts = reasoning_item.get('content', [])
                                            if parts and parts[-1].get('type') == 'output_text':
                                                append_to_text_field(parts[-1], 'text', reasoning_content)
                                            else:
                                                reasoning_item['content'] = [
                                                    {
                                                        'type': 'output_text',
                                                        'text': reasoning_content,
                                                    }
                                                ]

                                            reasoning_index = output.index(reasoning_item)
                                            data = {
                                                'type': 'response.reasoning_text.delta',
                                                'item_id': reasoning_item.get('id'),
                                                'output_index': reasoning_index,
                                                'content_index': max(
                                                    len(reasoning_item.get('content', [])) - 1,
                                                    0,
                                                ),
                                                'delta': reasoning_content,
                                            }
                                            await emit_response_completion_event(data)
                                            data = None

                                        if reasoning_detail_items:
                                            merge_streamed_reasoning_details(
                                                reasoning_item.setdefault('reasoning_details', []),
                                                reasoning_detail_items,
                                            )
                                            data = None

                                    if value:
                                        if (
                                            output
                                            and output[-1].get('type') == 'reasoning'
                                            and output[-1].get('attributes', {}).get('type') == 'reasoning_content'
                                        ):
                                            reasoning_item = output[-1]
                                            reasoning_item['ended_at'] = time.time()
                                            reasoning_item['duration'] = int(
                                                reasoning_item['ended_at'] - reasoning_item['started_at']
                                            )
                                            reasoning_item['status'] = 'completed'
                                            await emit_response_completion_event(
                                                {
                                                    'type': 'response.output_item.done',
                                                    'output_index': len(output) - 1,
                                                    'item': reasoning_item.copy(),
                                                }
                                            )

                                            output.append(
                                                {
                                                    'type': 'message',
                                                    'id': output_id('msg'),
                                                    'status': 'in_progress',
                                                    'role': 'assistant',
                                                    'content': [
                                                        {
                                                            'type': 'output_text',
                                                            'text': '',
                                                        }
                                                    ],
                                                }
                                            )

                                        if ENABLE_CHAT_RESPONSE_BASE64_IMAGE_URL_CONVERSION:
                                            value = await convert_markdown_base64_images(
                                                request,
                                                value,
                                                {
                                                    'chat_id': metadata.get('chat_id', None),
                                                    'message_id': metadata.get('message_id', None),
                                                },
                                                user,
                                            )

                                        # closure-cell str += recopies per chunk; append + join once at read is O(n)
                                        content_parts.append(value)

                                        # Continue the open tag block so its start tag is not re-detected on each chunk.
                                        last_item = output[-1] if output else None
                                        last_item_type = last_item.get('type', '') if last_item else ''
                                        inside_tag_block = (
                                            last_item is not None
                                            and last_item.get('status') == 'in_progress'
                                            and last_item.get('attributes', {}).get('type') != 'reasoning_content'
                                            and (
                                                last_item_type == 'reasoning'
                                                or last_item_type == 'open_webui:code_interpreter'
                                                or (
                                                    last_item_type == 'message'
                                                    and last_item.get('_tag_type') is not None
                                                )
                                            )
                                        )

                                        if inside_tag_block:
                                            # Append to the existing tag-based item
                                            if last_item_type == 'open_webui:code_interpreter':
                                                last_item['code'] = last_item.get('code', '') + value
                                            elif last_item_type == 'reasoning':
                                                parts = last_item.get('content', [])
                                                if parts and parts[-1].get('type') == 'output_text':
                                                    append_to_text_field(parts[-1], 'text', value)
                                                else:
                                                    last_item['content'] = [
                                                        {
                                                            'type': 'output_text',
                                                            'text': value,
                                                        }
                                                    ]
                                            else:
                                                # solution or other _tag_type message
                                                msg_parts = last_item.get('content', [])
                                                if msg_parts and msg_parts[-1].get('type') == 'output_text':
                                                    append_to_text_field(msg_parts[-1], 'text', value)
                                                else:
                                                    last_item['content'] = [
                                                        {
                                                            'type': 'output_text',
                                                            'text': value,
                                                        }
                                                    ]
                                        else:
                                            if not output or output[-1].get('type') != 'message':
                                                output.append(
                                                    {
                                                        'type': 'message',
                                                        'id': output_id('msg'),
                                                        'status': 'in_progress',
                                                        'role': 'assistant',
                                                        'content': [
                                                            {
                                                                'type': 'output_text',
                                                                'text': '',
                                                            }
                                                        ],
                                                    }
                                                )

                                            # Append value to last message item's text
                                            msg_parts = output[-1].get('content', [])
                                            if msg_parts and msg_parts[-1].get('type') == 'output_text':
                                                append_to_text_field(msg_parts[-1], 'text', value)
                                            else:
                                                output[-1]['content'] = [
                                                    {
                                                        'type': 'output_text',
                                                        'text': value,
                                                    }
                                                ]

                                        tag_output = None

                                        if DETECT_REASONING_TAGS:
                                            if not any(
                                                item.get('attributes', {}).get('type') == 'reasoning_content'
                                                for item in output
                                            ):
                                                tag_output, _ = tag_output_handler(
                                                    'reasoning',
                                                    reasoning_tags,
                                                    output,
                                                )

                                            solution_output, _ = tag_output_handler(
                                                'solution',
                                                DEFAULT_SOLUTION_TAGS,
                                                output,
                                            )
                                            if solution_output is not None:
                                                tag_output = solution_output

                                        if DETECT_CODE_INTERPRETER:
                                            code_output, end = tag_output_handler(
                                                'code_interpreter',
                                                DEFAULT_CODE_INTERPRETER_TAGS,
                                                output,
                                            )
                                            if code_output is not None:
                                                tag_output = code_output

                                            if end:
                                                break

                                        target_index = len(output) - 1
                                        target_item = output[target_index] if target_index >= 0 else {}
                                        target_content = target_item.get('content', [])
                                        content_index = max(len(target_content) - 1, 0)
                                        delta_event_type = (
                                            'response.reasoning_text.delta'
                                            if target_item.get('type') == 'reasoning'
                                            else 'response.output_text.delta'
                                        )
                                        data = {
                                            'type': delta_event_type,
                                            'item_id': target_item.get('id'),
                                            'output_index': target_index,
                                            'content_index': content_index,
                                            'delta': value,
                                        }
                                        delta_type = delta_event_type

                                        # the raw chunk still carries the tag text: resend the cleaned output instead
                                        if tag_output is not None:
                                            await flush_pending_delta_data()
                                            await event_emitter(
                                                {'type': 'chat:completion', 'data': {'output': full_output()}}
                                            )
                                            stream_size = 0
                                            data = None

                                if delta and data:
                                    await queue_pending_delta_data(data, delta_type)
                                elif data:
                                    await event_emitter(
                                        {
                                            'type': 'chat:completion',
                                            'data': data,
                                        }
                                    )
                        except (asyncio.CancelledError, KeyboardInterrupt):
                            raise
                        except Exception as e:
                            done = 'data: [DONE]' in line
                            if done:
                                pass
                            else:
                                log.debug('Error: %s', e)
                                stream_size = 0
                                continue
                        finally:
                            if not stream_size or not request.app.state.redis:
                                await save_current_response_stream()
                    await flush_pending_delta_data()

                    if output:
                        # Clean up the last message item
                        if output[-1].get('type') == 'message' and not continuing:
                            parts = output[-1].get('content', [])
                            if parts and parts[-1].get('type') == 'output_text':
                                parts[-1]['text'] = parts[-1]['text'].strip()

                                if not parts[-1]['text']:
                                    output.pop()

                                    if not output:
                                        output.append(
                                            {
                                                'type': 'message',
                                                'id': output_id('msg'),
                                                'status': 'in_progress',
                                                'role': 'assistant',
                                                'content': [{'type': 'output_text', 'text': ''}],
                                            }
                                        )

                        if output[-1].get('type') == 'reasoning':
                            reasoning_item = output[-1]
                            if reasoning_item.get('ended_at') is None:
                                reasoning_item['ended_at'] = time.time()
                                if reasoning_item.get('started_at') is not None:
                                    reasoning_item['duration'] = int(
                                        reasoning_item['ended_at'] - reasoning_item['started_at']
                                    )
                                reasoning_item['status'] = 'completed'

                    incomplete = finish_reason in {'length', 'content_filter', 'incomplete', 'failed'}
                    if (
                        not response_tool_calls
                        and not incomplete
                        and current_form_data.get('tool_choice') != 'none'
                        and current_form_data.get('response_format') in (None, {'type': 'text'})
                    ):
                        text_items = [item for item in output if item.get('type') == 'message']
                        text = ''.join(
                            part.get('text', '')
                            for item in text_items
                            for part in item.get('content', [])
                            if part.get('type') == 'output_text'
                        )
                        allowed = {
                            spec.get('function', spec).get('name') for spec in current_form_data.get('tools') or []
                        }
                        choice = current_form_data.get('tool_choice')
                        if isinstance(choice, dict):
                            allowed &= {choice.get('function', choice).get('name')}
                        recovered = recover_tool_calls(text, allowed)
                        if recovered and not any(item.get('type') == 'function_call' for item in output):
                            response_tool_calls = recovered
                            output = [item for item in output if item not in text_items]
                            content_parts[:] = [get_output_text(full_output())]
                            for call in recovered:
                                output.append(
                                    {
                                        'type': 'function_call',
                                        'id': call['id'],
                                        'call_id': call['id'],
                                        **call['function'],
                                        'status': 'in_progress',
                                        'recovered_from_text': True,
                                    }
                                )
                            await event_emitter(
                                {
                                    'type': 'chat:completion',
                                    'data': {'output': full_output(), 'flush': True},
                                }
                            )

                    if response_tool_calls:
                        for tc in response_tool_calls:
                            call_id = tc.get('id', '')
                            arguments = tc.get('function', {}).get('arguments', '{}')
                            for output_index, item in enumerate(output):
                                if item.get('type') == 'function_call' and item.get('call_id') == call_id:
                                    item['arguments'] = arguments
                                    item['status'] = 'completed'
                                    await emit_response_completion_event(
                                        {
                                            'type': 'response.function_call_arguments.done',
                                            'item_id': item.get('id'),
                                            'output_index': output_index,
                                            'arguments': arguments,
                                        }
                                    )
                                    await emit_response_completion_event(
                                        {
                                            'type': 'response.output_item.done',
                                            'output_index': output_index,
                                            'item': item.copy(),
                                        }
                                    )
                                    break
                        calls = split_tool_calls(response_tool_calls) if not incomplete else response_tool_calls
                        if incomplete:
                            for call in calls:
                                call['execution_error'] = INCOMPLETE_CALL_ERROR
                        tool_calls.append(calls)

                    # Responses API path: extract function_call items from output
                    if not response_tool_calls and output:
                        # Collect call_ids that already have results,
                        # including those from prior_output so we don't
                        # re-process tool calls from a previous turn.
                        handled_call_ids = {
                            item.get('call_id')
                            for item in (prior_output + output)
                            if item.get('type') == 'function_call_output'
                        }
                        responses_api_tool_calls = []
                        for item in output:
                            call_id = item.get('call_id') or item.get('id') or output_id('fc')
                            if item.get('type') == 'function_call' and call_id not in handled_call_ids:
                                arguments = item.get('arguments', '{}')
                                responses_api_tool_calls.append(
                                    {
                                        'id': call_id,
                                        'index': len(responses_api_tool_calls),
                                        **(
                                            {'execution_error': INCOMPLETE_CALL_ERROR}
                                            if incomplete or item.get('status') != 'completed'
                                            else {}
                                        ),
                                        'function': {
                                            'name': item.get('name', ''),
                                            'arguments': (
                                                arguments if isinstance(arguments, str) else JSONCodec.dumps(arguments)
                                            ),
                                        },
                                    }
                                )
                        if responses_api_tool_calls:
                            tool_calls.append(split_tool_calls(responses_api_tool_calls))

                    await save_current_response_stream()
                    if not current_usage:
                        usage = {**(usage or {}), 'incomplete': True}
                    return normalize_usage(current_usage)

                output_start = len(prior_output)
                try:
                    current_usage = await stream_body_handler(response, form_data)
                finally:
                    log_model_usage(request, user, form_data['model'], current_usage)
                    if response.background:
                        await response.background()

                tool_history = [item for item in prior_output if item.get('type') == 'function_call']
                tool_call_iterations = 0
                max_tool_call_iterations = getattr(
                    request.state, 'max_tool_call_iterations', CHAT_RESPONSE_MAX_TOOL_CALL_ITERATIONS
                )
                final_answer_requested = form_data.get('tool_choice') == 'none'
                stop_reason = None
                citation_source_ids = {}
                get_source_context(
                    (message or {}).get('sources') or metadata.get('sources', []),
                    citation_source_ids,
                    include_content=False,
                )
                user_message = get_last_user_message(form_data['messages'])
                messages = list(form_data['messages'])

                async def emit_output():
                    await save_current_response_stream()
                    # Channels publish whole messages; Continue can merge into the preceding item.
                    snapshot = continuing or (metadata.get('chat_id') or '').startswith('channel:')
                    frontend_output = []
                    for item in full_output() if snapshot else full_output()[output_start:]:
                        if item.get('type') == 'function_call_output':
                            # input_image parts are for the LLM only, via convert_output_to_messages
                            item = {
                                **item,
                                'output': [
                                    part for part in item.get('output', []) if part.get('type') != 'input_image'
                                ],
                            }
                        frontend_output.append(item)

                    if snapshot:
                        await event_emitter(
                            {
                                'type': 'chat:completion',
                                'data': {'output': frontend_output, 'flush': True},
                            }
                        )
                        return

                    for output_index, item in enumerate(frontend_output, start=output_start):
                        await event_emitter(
                            {
                                'type': 'response:completion',
                                'data': {
                                    'type': 'response.output_item.done',
                                    'output_index': output_index,
                                    'item': item,
                                },
                            }
                        )

                while tool_calls:
                    round_error = stop_reason
                    if max_tool_call_iterations is not None and tool_call_iterations >= max_tool_call_iterations:
                        round_error = 'Tool-call iteration limit reached.'
                    if final_answer_requested:
                        round_error = round_error or 'Tool calls are disabled for this response.'
                    if not round_error:
                        tool_call_iterations += 1

                    # Resolve dispatch before approvals and special handling such as ask_user.
                    response_tool_calls = []
                    output_by_call_id = {
                        item.get('call_id'): item for item in output if item.get('type') == 'function_call'
                    }
                    for tool_call in split_tool_call_batches(tool_calls.pop(0), metadata.get('tools', {})):
                        error = tool_call.get('execution_error') or resolve_tool_call(
                            tool_call, metadata.get('tools', {})
                        )
                        call_id = tool_call.get('id') or output_id('fc')
                        tool_call['id'] = call_id
                        function = tool_call.get('function', {})
                        item = output_by_call_id.get(call_id)
                        if item is None:
                            item = {'type': 'function_call', 'id': call_id, 'call_id': call_id}
                            output.append(item)
                            output_by_call_id[call_id] = item
                        item.update(
                            name=function.get('name', ''),
                            arguments=function.get('arguments', '{}'),
                            status='failed' if error else item.get('status', 'in_progress'),
                        )
                        tool = metadata.get('tools', {}).get(item['name'])
                        error = error or (
                            prepare_tool_call(
                                item,
                                tool.get('spec', {}).get('parameters'),
                                display_labels=tool.get('display_labels', False),
                                approval_review=tool.get('approval_review', False),
                            )
                            if tool
                            else f'Error: Tool "{item["name"]}" not found.'
                        )
                        blocked = tool_repetition_error(item, tool_history)
                        is_incomplete = tool_call.get('execution_error') == INCOMPLETE_CALL_ERROR
                        stop_reason = (
                            stop_reason or round_error or blocked or (INCOMPLETE_CALL_ERROR if is_incomplete else None)
                        )
                        error = f'Error: {round_error}' if round_error else error or blocked
                        if error:
                            tool_history.append(item)
                        function['arguments'] = item['arguments']
                        item['status'] = 'failed' if error else item['status']
                        if 'provider_function' in tool_call:
                            item['provider_function'] = tool_call['provider_function']
                            # The first resolved call reuses the hidden dispatcher's output item.
                            item['hidden'] = tool.get('hidden') is True if tool else False
                        if 'provider_batch' in tool_call:
                            item['provider_batch'] = tool_call['provider_batch']
                        if error:
                            output.append(
                                {
                                    'type': 'function_call_output',
                                    'id': output_id('fco'),
                                    'call_id': call_id,
                                    'output': [{'type': 'input_text', 'text': error}],
                                    'status': 'failed',
                                }
                            )
                        else:
                            response_tool_calls.append(tool_call)

                    ask_user_staged, ask_user_error = stage_ask_user_tool_calls(response_tool_calls, output, output_id)
                    if ask_user_error:
                        response_tool_calls = [
                            tool_call
                            for tool_call in response_tool_calls
                            if tool_call.get('function', {}).get('name') != 'ask_user'
                        ]
                    elif ask_user_staged and save_to_chat:
                        if is_saved_chat_id(metadata.get('chat_id')) and metadata.get('message_id'):
                            await pause_for_tool_approval(
                                metadata['chat_id'],
                                metadata['message_id'],
                                full_output(),
                                form_data,
                                metadata,
                            )
                        await event_emitter({'type': 'chat:completion', 'data': {'output': full_output()}})
                        return
                    elif ask_user_staged:
                        # Live questions await the socket callback, not the saved-chat resolver.
                        for item in output:
                            if item.get('name') == 'ask_user' and item.get('status') == 'pending':
                                item['status'] = 'in_progress'

                    tool_approval_mode = metadata.get('params', {}).get('tool_approval_mode', 'full')
                    if (
                        response_tool_calls
                        and tool_approval_mode == 'ask'
                        and is_saved_chat_id(metadata.get('chat_id'))
                        and metadata.get('message_id')
                    ):
                        await pause_for_tool_approval(
                            metadata['chat_id'],
                            metadata['message_id'],
                            full_output(),
                            form_data,
                            metadata,
                        )
                        await event_emitter(
                            {
                                'type': 'chat:completion',
                                'data': {
                                    'output': full_output(),
                                },
                            }
                        )
                        return

                    await emit_output()

                    results = []

                    # Provider calls have been resolved and staged; execute their saved items directly.
                    response_tool_calls = [output_by_call_id[call['id']] for call in response_tool_calls]
                    tool_results = {}
                    if tool_approval_mode != 'auto':
                        delegate_calls = [call for call in response_tool_calls if call.get('name') == 'delegate_task']
                        for tool_call in response_tool_calls:
                            if tool_call.get('name') != 'delegate_task':
                                tool_results[tool_call['call_id']] = await execute_tool_call(
                                    form_data, metadata, event_caller, tool_call, tool_history
                                )
                        tool_results.update(
                            zip(
                                [tool_call['call_id'] for tool_call in delegate_calls],
                                await asyncio.gather(
                                    *(
                                        execute_tool_call(form_data, metadata, event_caller, tool_call, tool_history)
                                        for tool_call in delegate_calls
                                    )
                                ),
                            )
                        )

                    for tool_call in response_tool_calls:
                        tool_call_id = tool_call['call_id']
                        tool_function_name = tool_call.get('name', '')
                        if tool_approval_mode == 'auto':
                            try:
                                tool_results[tool_call_id] = await execute_tool_call(
                                    form_data, metadata, event_caller, tool_call, tool_history
                                )
                            except ToolApprovalRequired:
                                break
                        tool_function_params, tool_result, tool, tool_type = tool_results[tool_call['call_id']]
                        if tool_result is None:
                            results.append(
                                {
                                    'tool_call_id': tool_call_id,
                                    'content': (
                                        'Error: Tool call arguments could not be parsed. The model generated '
                                        f'malformed or incomplete JSON for `{tool_function_name}`. Please try again.'
                                    ),
                                }
                            )
                            continue

                        terminal_file_result = build_terminal_file_tool_result(
                            tool_function_name,
                            tool_function_params,
                            tool_result,
                            tool,
                            metadata,
                        )
                        if terminal_file_result:
                            tool_result = terminal_file_result

                        tool_result, tool_result_files, tool_result_embeds = await process_tool_result(
                            request,
                            tool_function_name,
                            tool_result,
                            tool_type,
                            metadata,
                            user,
                        )

                        await terminal_event_handler(
                            tool_function_name,
                            tool_function_params,
                            tool_result,
                            event_emitter,
                        )

                        results.append(
                            {
                                'tool_call_id': tool_call_id,
                                'content': await tool_result_with_citations(
                                    tool_function_name,
                                    tool_function_params,
                                    tool_result,
                                    tool,
                                    model,
                                    citation_source_ids,
                                    user_message,
                                    event_emitter,
                                ),
                                **({'files': tool_result_files} if tool_result_files else {}),
                                **({'embeds': tool_result_embeds} if tool_result_embeds else {}),
                            }
                        )

                    result_status_by_call_id = {}
                    for result in results:
                        output_parts = [{'type': 'input_text', 'text': result.get('content', '')}]
                        local_output_status = (
                            'failed' if _is_tool_result_error(result.get('content', '')) else 'completed'
                        )
                        result_status_by_call_id[result.get('tool_call_id', '')] = local_output_status

                        # Data-URI images: LLM only. File-URL images: LLM and frontend. Other files: frontend.
                        display_files = []
                        for file_item in result.get('files', []):
                            if file_item.get('type') == 'image' and file_item.get('url', '').startswith('data:'):
                                # LLM-only: add as input_image part, not frontend display output.
                                image_url = await store_tool_result_image(request, file_item['url'], metadata, user)
                                output_parts.append({'type': 'input_image', 'image_url': image_url})
                            else:
                                display_files.append(file_item)
                                if file_item.get('type') == 'image' and file_item.get('url'):
                                    output_parts.append({'type': 'input_image', 'image_url': file_item['url']})

                        output.append(
                            {
                                'type': 'function_call_output',
                                'id': output_id('fco'),
                                'call_id': result.get('tool_call_id', ''),
                                'output': output_parts,
                                'status': local_output_status,
                                **({'files': display_files} if display_files else {}),
                                **({'embeds': result.get('embeds')} if result.get('embeds') else {}),
                            }
                        )

                    for item in response_tool_calls:
                        if item['call_id'] in result_status_by_call_id:
                            item['status'] = result_status_by_call_id[item['call_id']]

                    if any(call.get('status') == 'pending' for call in response_tool_calls):
                        # Include all completed results before pausing the remaining calls.
                        await pause_for_tool_approval(
                            metadata['chat_id'], metadata['message_id'], full_output(), form_data, metadata
                        )
                        await event_emitter({'type': 'chat:completion', 'data': {'output': full_output()}})
                        return

                    await emit_output()
                    if final_answer_requested:
                        await emit_message_error(round_error or 'The provider requested tools after tool use stopped.')
                        break

                    try:
                        new_form_data = {
                            **form_data,
                            'stream': True,
                            'metadata': metadata,
                        }

                        if ENABLE_RESPONSES_API_STATEFUL and last_response_id:
                            system_message = get_system_message(form_data['messages'])
                            new_form_data['messages'] = (
                                [system_message] if system_message else []
                            ) + convert_output_to_messages(
                                strip_tool_images(output, model), raw=True, reasoning_format=get_reasoning_format(model)
                            )
                            new_form_data['previous_response_id'] = last_response_id
                        else:
                            tool_messages = convert_output_to_messages(
                                strip_tool_images(output, model),
                                raw=True,
                                reasoning_format=get_reasoning_format(model),
                                flatten_tool_images=True,
                            )

                            # Chat Completions providers don't support multimodal
                            # tool messages.  Extract images into a user message.
                            image_urls = []
                            for message in tool_messages:
                                if message.get('role') == 'tool' and isinstance(message.get('content'), list):
                                    text_parts = []
                                    for part in message['content']:
                                        if part.get('type') == 'input_text':
                                            text_parts.append(part.get('text', ''))
                                        elif part.get('type') == 'input_image':
                                            image_urls.append(part.get('image_url', ''))
                                    message['content'] = ''.join(text_parts)

                            if current_usage and tool_messages:
                                tool_messages[0]['usage'] = current_usage

                            user_prompt = get_last_user_message_item(form_data['messages'])
                            messages = [*messages, *tool_messages]

                            if image_urls:
                                messages.append(
                                    {
                                        'role': 'user',
                                        'content': [
                                            {
                                                'type': 'text',
                                                'text': 'Here are the images from the tool results above. Please analyze them.',
                                            },
                                            *[{'type': 'image_url', 'image_url': {'url': url}} for url in image_urls],
                                        ],
                                    }
                                )

                            # Use the existing checkpoint format internally and strip it from the provider payload.
                            def get_request_messages(messages: list[dict]) -> list[dict]:
                                request_messages = process_messages_with_output(messages)
                                context_summary = next(
                                    (
                                        message['contextSummary']
                                        for message in messages
                                        if message.get('contextSummary')
                                    ),
                                    None,
                                )
                                if context_summary:
                                    request_messages.insert(
                                        1 if request_messages and request_messages[0].get('role') == 'system' else 0,
                                        {'role': 'system', 'content': f'[CONVERSATION SUMMARY]\n{context_summary}'},
                                    )
                                return request_messages

                            new_form_data['messages'] = get_request_messages(messages)
                            try:
                                user_messages = [
                                    (message, message.get('content'))
                                    for message in messages
                                    if user_prompt
                                    and message.get('role') == 'user'
                                    and message.get('content') == user_prompt.get('content')
                                ]
                                summary_form_data = None
                                if not filter_functions:
                                    summary_form_data = normalize_messages_for_model(
                                        await convert_url_images_to_base64({**new_form_data}, user=user)
                                    )
                                try:
                                    # Summarize the user's text, then restore source context for the model.
                                    for message, _ in user_messages:
                                        message['content'] = copy.deepcopy(message.get('content'))
                                        set_last_user_message_content(
                                            metadata.get('base_user_prompt') or user_message, [message]
                                        )
                                    messages, context_summary, compacted = await compact_messages_for_request(
                                        request,
                                        summary_form_data,
                                        user,
                                        messages,
                                        metadata,
                                        form_data['model'],
                                        request.app.state.MODELS,
                                        bypass_system_prompt=True,
                                    )
                                finally:
                                    for message, content in user_messages:
                                        message['content'] = content
                                if compacted:
                                    if user_prompt is not None:
                                        messages = [message for message in messages if message != user_prompt]
                                        messages.insert(
                                            1 if messages and messages[0].get('role') == 'system' else 0,
                                            dict(user_prompt),
                                        )
                                    checkpoint = next(
                                        idx for idx, message in enumerate(messages) if message.get('role') != 'system'
                                    )
                                    messages[checkpoint] = {**messages[checkpoint], 'contextSummary': context_summary}
                                    new_form_data['messages'] = get_request_messages(messages)
                                    output.append(
                                        {
                                            'type': 'open_webui:context_compaction',
                                            'id': output_id('cc'),
                                            'call_id': next(
                                                message['tool_calls'][0]['id']
                                                for message in messages
                                                if message.get('tool_calls')
                                            ),
                                            'summary': context_summary,
                                        }
                                    )
                                    await emit_output()
                            except Exception:
                                log.exception('Tool loop context compaction failed; keeping the current context')

                        new_form_data = await convert_url_images_to_base64(new_form_data, user=user)

                        if filter_functions:
                            new_form_data, _ = await process_filter_functions(
                                request=request,
                                filter_context=filter_context,
                                filter_functions=filter_functions,
                                filter_type='request',
                                form_data=new_form_data,
                                extra_params=extra_params,
                            )

                        final_answer_requested = bool(stop_reason) or (
                            max_tool_call_iterations is not None and tool_call_iterations >= max_tool_call_iterations
                        )
                        if final_answer_requested:
                            new_form_data = final_answer_payload(new_form_data)
                        new_form_data = normalize_messages_for_model(new_form_data)

                        res = await generate_chat_completion(
                            request,
                            new_form_data,
                            user,
                            bypass_system_prompt=True,
                        )

                        if isinstance(res, StreamingResponse):
                            # Reset response-relative output indices; keep prior output as the UI history prefix.
                            prior_output = list(full_output())
                            # Trim the trailing empty placeholder message
                            # so it doesn't persist as a ghost item once
                            # the new stream produces real content.
                            if (
                                prior_output
                                and prior_output[-1].get('type') == 'message'
                                and prior_output[-1].get('status') == 'in_progress'
                            ):
                                msg_parts = prior_output[-1].get('content', [])
                                if not msg_parts or (len(msg_parts) == 1 and not msg_parts[0].get('text', '').strip()):
                                    prior_output.pop()
                            output = []
                            output_start = len(prior_output)
                            try:
                                current_usage = await stream_body_handler(res, new_form_data)
                            finally:
                                log_model_usage(request, user, form_data['model'], current_usage)
                        elif getattr(res, 'status_code', 200) >= 400:
                            await emit_message_error(get_message_error_content(get_response_error_detail(res)))
                            break
                        else:
                            break
                    except Exception as e:
                        error_content = get_message_error_content(e)
                        log.exception('Tool-call continuation failed: %s', error_content)
                        await emit_message_error(error_content)
                        break

                if DETECT_CODE_INTERPRETER:
                    MAX_RETRIES = 5
                    retries = 0

                    while output and output[-1].get('type') == 'open_webui:code_interpreter' and retries < MAX_RETRIES:
                        await event_emitter(
                            {
                                'type': 'chat:completion',
                                'data': {
                                    'output': full_output(),
                                },
                            }
                        )

                        retries += 1
                        log.debug('Attempt count: %s', retries)

                        ci_item = output[-1]
                        ci_output = ''
                        try:
                            if ci_item.get('attributes', {}).get('type') == 'code':
                                code = ci_item.get('code', '')
                                # Sanitize code (strips ANSI codes and markdown fences)
                                code = sanitize_code(code)

                                if CODE_INTERPRETER_BLOCKED_MODULES:
                                    blocking_code = textwrap.dedent(f"""
                                        import builtins
    
                                        BLOCKED_MODULES = {CODE_INTERPRETER_BLOCKED_MODULES}
    
                                        _real_import = builtins.__import__
                                        def restricted_import(name, globals=None, locals=None, fromlist=(), level=0):
                                            if name.split('.')[0] in BLOCKED_MODULES:
                                                importer_name = globals.get('__name__') if globals else None
                                                if importer_name == '__main__':
                                                    raise ImportError(
                                                        f"Direct import of module {{name}} is restricted."
                                                    )
                                            return _real_import(name, globals, locals, fromlist, level)
    
                                        builtins.__import__ = restricted_import
                                    """)
                                    code = blocking_code + '\n' + code

                                ci_engine = await Config.get('code_interpreter.engine')
                                if ci_engine == 'pyodide':
                                    ci_output = await event_caller(
                                        {
                                            'type': 'execute:python',
                                            'data': {
                                                'id': str(uuid4()),
                                                'code': code,
                                                'session_id': metadata.get('session_id', None),
                                                'files': metadata.get('files', []),
                                            },
                                        }
                                    )
                                elif ci_engine == 'jupyter':
                                    ci_output = await execute_code_jupyter(
                                        await Config.get('code_interpreter.jupyter.url'),
                                        code,
                                        (
                                            await Config.get('code_interpreter.jupyter.auth_token')
                                            if await Config.get('code_interpreter.jupyter.auth') == 'token'
                                            else None
                                        ),
                                        (
                                            await Config.get('code_interpreter.jupyter.auth_password')
                                            if await Config.get('code_interpreter.jupyter.auth') == 'password'
                                            else None
                                        ),
                                        await Config.get('code_interpreter.jupyter.timeout'),
                                    )
                                else:
                                    ci_output = {'stdout': 'Code interpreter engine not configured.'}

                                log.debug('Code interpreter output: %s', ci_output)

                                # Handle error responses from event_caller
                                # (e.g. session disconnected, timeout)
                                if isinstance(ci_output, dict) and ci_output.get('error'):
                                    ci_output = {'stderr': ci_output['error']}

                                if isinstance(ci_output, dict):
                                    stdout = ci_output.get('stdout', '')

                                    if isinstance(stdout, str):
                                        stdoutLines = stdout.split('\n')
                                        for idx, line in enumerate(stdoutLines):
                                            if re.match(r'data:image/\w+;base64', line):
                                                image_url = await get_image_url_from_base64(
                                                    request,
                                                    line,
                                                    metadata,
                                                    user,
                                                )
                                                if image_url:
                                                    stdoutLines[idx] = f'![Output Image]({image_url})'

                                        ci_output['stdout'] = '\n'.join(stdoutLines)

                                    result = ci_output.get('result', '')

                                    if isinstance(result, str):
                                        resultLines = result.split('\n')
                                        for idx, line in enumerate(resultLines):
                                            if re.match(r'data:image/\w+;base64', line):
                                                image_url = await get_image_url_from_base64(
                                                    request,
                                                    line,
                                                    metadata,
                                                    user,
                                                )
                                                resultLines[idx] = f'![Output Image]({image_url})'
                                        ci_output['result'] = '\n'.join(resultLines)
                        except Exception as e:
                            ci_output = str(e)

                        ci_item['output'] = ci_output
                        ci_item['status'] = 'completed'

                        output.append(
                            {
                                'type': 'message',
                                'id': output_id('msg'),
                                'status': 'in_progress',
                                'role': 'assistant',
                                'content': [{'type': 'output_text', 'text': ''}],
                            }
                        )

                        await event_emitter(
                            {
                                'type': 'chat:completion',
                                'data': {
                                    'output': full_output(),
                                },
                            }
                        )

                        try:
                            new_form_data = {
                                **form_data,
                                'stream': True,
                                'metadata': metadata,
                                'messages': [
                                    *messages,
                                    *convert_output_to_messages(
                                        strip_tool_images(output, model),
                                        raw=True,
                                        reasoning_format=get_reasoning_format(model),
                                        flatten_tool_images=True,
                                    ),
                                ],
                            }

                            if filter_functions:
                                new_form_data, _ = await process_filter_functions(
                                    request=request,
                                    filter_context=filter_context,
                                    filter_functions=filter_functions,
                                    filter_type='request',
                                    form_data=new_form_data,
                                    extra_params=extra_params,
                                )

                            new_form_data = normalize_messages_for_model(new_form_data)

                            res = await generate_chat_completion(
                                request,
                                new_form_data,
                                user,
                                bypass_system_prompt=True,
                            )

                            if isinstance(res, StreamingResponse):
                                try:
                                    await stream_body_handler(res, new_form_data)
                                finally:
                                    log_model_usage(request, user, form_data['model'], current_usage)
                            elif getattr(res, 'status_code', 200) >= 400:
                                await emit_message_error(get_message_error_content(get_response_error_detail(res)))
                                break
                            else:
                                break
                        except Exception as e:
                            error_content = get_message_error_content(e)
                            log.exception('Code interpreter continuation failed: %s', error_content)
                            await emit_message_error(error_content)
                            break

                # Mark all in-progress items as completed
                for item in [*prior_output, *output]:
                    if item.get('status') == 'in_progress':
                        item['status'] = 'completed'

                current_output = full_output()
                title = await Chats.get_chat_title_by_id(metadata['chat_id']) if save_to_chat else ''
                data = {
                    'done': True,
                    'output': current_output,
                    'title': title,
                    **({'usage': usage} if usage else {}),
                }

                if save_to_chat:
                    # Save final output once. The delta path keeps in-progress
                    # state in response_streams instead of writing tokens to DB.
                    await Chats.upsert_message_to_chat_by_id_and_message_id(
                        metadata['chat_id'],
                        metadata['message_id'],
                        {
                            'done': True,
                            'meta': {'response_ended_at': time.time()},
                            'output': current_output,
                            **({'usage': usage} if usage else {}),
                        },
                    )

                await clear_response_stream(request.app.state.redis, response_stream_task_id)
                await publish_chat_finished_event(
                    request,
                    user,
                    metadata,
                    title,
                    get_output_text(current_output) if continuing else ''.join(content_parts),
                    current_output,
                )

                await event_emitter(
                    {
                        'type': 'chat:completion',
                        'data': data,
                    }
                )

                ctx['assistant_message'] = {
                    'content': get_output_text(current_output)
                    if continuing
                    else ''.join(content_parts) or get_output_text(current_output),
                    'output': current_output,
                    **({'usage': usage} if usage else {}),
                    **({'finish_reason': finish_reason} if finish_reason else {}),
                }
                await outlet_filter_handler(ctx)
                await background_tasks_handler(ctx)
            except asyncio.CancelledError:
                log.warning('Task was cancelled!')

                # Close the iterator so stream_wrapper's finally block releases the upstream connection.
                if hasattr(response, 'body_iterator') and hasattr(response.body_iterator, 'aclose'):
                    try:
                        await asyncio.shield(response.body_iterator.aclose())
                    except (asyncio.CancelledError, Exception):
                        pass

                async def save_cancelled_state():
                    cancelled_output = full_output()
                    result_call_ids = {
                        item.get('call_id')
                        for item in cancelled_output
                        if item.get('type') == 'function_call_output' and item.get('call_id')
                    }
                    for item in cancelled_output:
                        # A tool call is stamped completed when its arguments finish, before the tool runs
                        is_running_tool_call = (
                            item.get('type') == 'function_call'
                            and item.get('status') == 'completed'
                            and (item.get('call_id') or item.get('id')) not in result_call_ids
                        )
                        if item.get('status') == 'in_progress' or is_running_tool_call:
                            item['status'] = 'incomplete'

                    await event_emitter({'type': 'chat:tasks:cancel', 'data': {'output': cancelled_output}})
                    if save_to_chat:
                        await Chats.upsert_message_to_chat_by_id_and_message_id(
                            metadata['chat_id'],
                            metadata['message_id'],
                            {
                                'done': True,
                                'output': cancelled_output,
                            },
                        )
                    await clear_response_stream(request.app.state.redis, response_stream_task_id)

                try:
                    await asyncio.shield(save_cancelled_state())
                except (asyncio.CancelledError, Exception):
                    pass
                raise  # re-raise CancelledError for proper propagation

            if response.background is not None:
                await response.background()

        return await response_handler(response, events)

    else:
        # Fallback to the original response
        async def stream_wrapper(original_generator, events):
            def wrap_item(item):
                return f'data: {item}\n\n'

            audit_enabled = bool(getattr(request.state, 'audit_request_id', None))
            current_usage = None
            try:
                message = {}
                outlet_task = None
                filter_context = FilterContext()
                outlet_enabled = ENABLE_API_OUTLET_FILTERS and bool(filter_functions)
                if ENABLE_API_OUTLET_FILTERS and not outlet_enabled:
                    try:
                        outlet_enabled = bool(
                            (isinstance(model, dict) and 'pipeline' in model)
                            or get_sorted_filters(
                                model.get('id') if isinstance(model, dict) else model, request.app.state.MODELS
                            )
                        )
                    except Exception:
                        outlet_enabled = True

                for event in events:
                    event, _ = await process_filter_functions(
                        request=request,
                        filter_context=filter_context,
                        filter_functions=filter_functions,
                        filter_type='stream',
                        form_data=event,
                        extra_params=extra_params,
                    )

                    if event:
                        yield wrap_item(JSONCodec.dumps(event))

                async for data in original_generator:
                    if filter_functions or audit_enabled:
                        line = data.decode('utf-8', 'replace') if isinstance(data, bytes) else data
                        if isinstance(line, str) and line.startswith('data:'):
                            payload = line.removeprefix('data:').strip()
                            if payload and payload != '[DONE]':
                                try:
                                    event = JSONCodec.loads(payload)
                                except JSONCodec.JSONDecodeError:
                                    event = None

                                if isinstance(event, dict):
                                    if audit_enabled:
                                        form_data['model'] = event.get('selected_model_id') or form_data['model']
                                        payload = (
                                            event.get('response') if isinstance(event.get('response'), dict) else event
                                        )
                                        usage = payload.get('usage') or {}
                                        timings = payload.get('timings') or {}
                                        if isinstance(usage, dict) and isinstance(timings, dict) and (usage or timings):
                                            current_usage = {**usage, **timings}
                                    if filter_functions:
                                        event, _ = await process_filter_functions(
                                            request=request,
                                            filter_context=filter_context,
                                            filter_functions=filter_functions,
                                            filter_type='stream',
                                            form_data=event,
                                            extra_params=extra_params,
                                        )
                                        data = wrap_item(JSONCodec.dumps(event)) if event else None

                    if data:
                        if outlet_enabled:
                            done = update_assistant_message_from_stream(message, data)
                            if done and message and outlet_task is None:
                                ctx['assistant_message'] = message
                                # Start before yielding [DONE], which may trigger a disconnect.
                                outlet_task = asyncio.create_task(outlet_filter_handler(ctx))
                        yield data

                if outlet_task is not None:
                    await asyncio.shield(outlet_task)
                elif outlet_enabled and message:
                    ctx['assistant_message'] = message
                    await outlet_filter_handler(ctx)
            except Exception as e:
                log.exception('Chat completion stream failed mid-response: %s', e)
                # Separate the error frame from any unfinished upstream event.
                yield f'\n\ndata: {JSONCodec.dumps({"error": {"message": "Chat completion stream failed"}})}\n\n'
                yield 'data: [DONE]\n\n'
            finally:
                log_model_usage(request, user, form_data['model'], current_usage)

        return StreamingResponse(
            convert_responses_stream_to_chat_completions(stream_wrapper(response.body_iterator, events)),
            headers=dict(response.headers),
            background=response.background,
        )


async def process_chat_response(response, ctx):
    # Non-streaming response
    if not isinstance(response, StreamingResponse):
        return await non_streaming_chat_response_handler(response, ctx)

    # Non standard response
    if not any(
        content_type in response.headers['Content-Type']
        for content_type in ['text/event-stream', 'application/x-ndjson']
    ):
        return response

    # Streaming response
    return await streaming_chat_response_handler(response, ctx)