#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
Taiji 会话服务：WebSocket 中继与 session/completions 业务逻辑
与 taiji_as 透传通信，流式结果通过 Socket.IO chat-events 推送。
"""

import logging
import asyncio
import uuid
import time
from dataclasses import dataclass
from typing import Dict, Any, List, Optional

from fastapi import HTTPException, status

from open_webui.env import BYPASS_MODEL_ACCESS_CONTROL
from open_webui.models.models import Models
from open_webui.utils.middleware import process_image_url
from open_webui.utils.misc import get_content_from_message
from open_webui.utils.models import (
    get_all_models,
    check_model_access,
    get_pipeline_model_for_user,
)
from open_webui.models.chats import Chats
from open_webui.utils.chat_storage import (
    extract_chat_fields_from_form,
    save_chat_messages_on_request,
)
from open_webui.tasks import create_task
from open_webui.socket.main import broadcast_to_user
from open_webui.apps.taiji.repositories.employee_repository import employee_repository
from open_webui.env import DATABASE_URL
from open_webui.apps.taiji.services.employee.delivery_policy import (
    should_prefer_direct_delivery,
)
from open_webui.apps.taiji.acl.employee import is_employee_readable
from open_webui.apps.taiji.utils.chat_meta import (
    get_chat_employee_id,
    merged_chat_meta,
    persist_merged_chat_meta,
)
from open_webui.apps.taiji.utils.session_model_redis import (
    remember_session_coding_model,
)
from open_webui.apps.taiji.utils.websocket_relay import relay_manager
from open_webui.apps.taiji.utils import get_session_wait_timeout_seconds
from open_webui.apps.taiji.services.taiji_connection import (
    taiji_manager,
    TaijiConnectionPool,
)

log = logging.getLogger(__name__)

DELIVERY_CHANNEL_REGISTER = "register"
DELIVERY_CHANNEL_DIRECT = "direct"
META_DELIVERY_CHANNEL = "taiji_delivery_channel"
META_DELIVERY_INSTANCE = "taiji_delivery_instance_id"
META_DELIVERY_WS_URL = "taiji_delivery_ws_url"

# 反向注册员工在重启/重连窗口期 direct 回退失败时，短等待重试 register 的总时长与步长。
# 仅对本应走 register（非 Docker 直连）的员工生效，避免硬抛 1060。
REGISTER_RECONNECT_WAIT_SECONDS = 5.0
REGISTER_RECONNECT_POLL_SECONDS = 0.5


@dataclass(frozen=True)
class EmployeeDeliveryRoute:
    """员工 Taiji 消息物理投递通道（register 反向注册 / direct 平台连出）。"""

    channel: str
    instance_id: Optional[str] = None
    ws_url: Optional[str] = None


async def clear_delivery_channels_for_employee(employee_id: str) -> int:
    """员工连接重置时清除其所有会话上的 taiji_delivery_* 绑定（避免 stale ws_url）。"""
    from open_webui.internal.db import get_async_db
    from open_webui.models.chats import Chat

    eid = (employee_id or "").strip()
    if not eid:
        return 0
    keys = (META_DELIVERY_CHANNEL, META_DELIVERY_INSTANCE, META_DELIVERY_WS_URL)
    cleared = 0
    async with get_async_db() as db:
        from sqlalchemy import or_, select, text
        from sqlalchemy.dialects.postgresql import JSONB

        db_url = (DATABASE_URL or "").lower()
        filters = [Chat.instance_id == eid]
        if "sqlite" in db_url:
            filters.append(
                text("json_extract(meta, '$.instance_id') = :eid").bindparams(eid=eid)
            )
        elif "postgres" in db_url:
            # PostgreSQL 通用 JSON 列需 cast 为 JSONB 才能使用 ->> / astext
            filters.append(Chat.meta.cast(JSONB)["instance_id"].astext == eid)
        rows_result = await db.execute(select(Chat).where(or_(*filters)))
        rows = rows_result.scalars().all()
        for row in rows:
            meta = dict(row.meta or {})
            inner = dict((row.chat or {}).get("meta") or {})
            meta.update(inner)
            if not any(meta.get(k) for k in keys):
                continue
            for k in keys:
                meta.pop(k, None)
            row.meta = meta
            chat_data = dict(row.chat or {})
            chat_data["meta"] = meta
            row.chat = chat_data
            cleared += 1
        if cleared:
            await db.commit()
    if cleared:
        log.info(
            "[DeliveryRoute] cleared delivery binding for employee=%s chats=%s",
            eid,
            cleared,
        )
    return cleared


async def resolve_registered_instance_with_policy(
    employee_id: str,
    user_id: str,
    *,
    is_admin: bool = False,
) -> Optional[str]:
    """带 Docker 策略的 register 实例解析。"""
    eid = (employee_id or "").strip()
    if not eid:
        return None
    row = await employee_repository.get_by_id(eid, None if is_admin else user_id)
    if row and await should_prefer_direct_delivery(row, eid, user_id):
        return None
    instance_id = await resolve_registered_instance_for_employee(
        eid,
        user_id,
        is_admin=is_admin,
        skip_employee_id_fallback=False,
    )
    if instance_id and taiji_manager.is_connected(instance_id):
        return instance_id
    return None


async def persist_chat_delivery_channel(
    *,
    user_id: str,
    chat_id: str,
    channel: str,
    instance_id: Optional[str] = None,
    ws_url: Optional[str] = None,
) -> None:
    """记录本轮 chat 实际使用的 Taiji 投递通道，供 question_response/abort/steer 复用。"""
    cid = (chat_id or "").strip()
    uid = (user_id or "").strip()
    ch = (channel or "").strip()
    if not cid or not uid or ch not in (
        DELIVERY_CHANNEL_REGISTER,
        DELIVERY_CHANNEL_DIRECT,
    ):
        return
    try:
        patch: Dict[str, Any] = {META_DELIVERY_CHANNEL: ch}
        remove_keys: tuple = ()
        if ch == DELIVERY_CHANNEL_REGISTER:
            patch[META_DELIVERY_INSTANCE] = (instance_id or "").strip() or None
            remove_keys = (META_DELIVERY_WS_URL,)
        else:
            patch[META_DELIVERY_WS_URL] = (ws_url or "").strip() or None
            remove_keys = (META_DELIVERY_INSTANCE,)
        if not await persist_merged_chat_meta(uid, cid, patch, remove_keys=remove_keys):
            return
        log.debug(
            "[DeliveryRoute] persisted chat=%s channel=%s instance=%s ws=%s",
            cid,
            ch,
            instance_id,
            ws_url,
        )
    except Exception as exc:
        log.debug(
            "[DeliveryRoute] persist failed chat=%s channel=%s: %s",
            cid,
            ch,
            exc,
        )


async def resolve_chat_delivery_from_meta(
    chat_id: str,
    user_id: str,
) -> Optional[EmployeeDeliveryRoute]:
    """从 chat.meta 读取已绑定的投递通道。"""
    cid = (chat_id or "").strip()
    if not cid:
        return None
    chat = await Chats.get_chat_by_id_and_user_id(cid, user_id)
    if not chat:
        return None
    meta = merged_chat_meta(chat)
    channel = (meta.get(META_DELIVERY_CHANNEL) or "").strip()
    if not channel:
        return None
    if channel == DELIVERY_CHANNEL_REGISTER:
        instance_id = (meta.get(META_DELIVERY_INSTANCE) or "").strip()
        if instance_id:
            return EmployeeDeliveryRoute(
                channel=DELIVERY_CHANNEL_REGISTER,
                instance_id=instance_id,
            )
    elif channel == DELIVERY_CHANNEL_DIRECT:
        ws_url = (meta.get(META_DELIVERY_WS_URL) or "").strip()
        if ws_url:
            return EmployeeDeliveryRoute(
                channel=DELIVERY_CHANNEL_DIRECT,
                ws_url=ws_url,
            )
    return None


async def resolve_employee_delivery_route(
    employee_id: str,
    user_id: str,
    chat_id: str = "",
    *,
    is_admin: bool = False,
) -> Optional[EmployeeDeliveryRoute]:
    """
    解析员工 Taiji 投递通道。

    优先级：chat 已绑定通道 → register（Docker 直连员工跳过 register）→ direct。
    """
    eid = (employee_id or "").strip()
    if not eid or eid == "taiji":
        return None

    row = await employee_repository.get_by_id(eid, None if is_admin else user_id)
    if not row:
        return None
    owner_uid = str(row.user_id) if row.user_id else user_id
    prefer_direct = await should_prefer_direct_delivery(row, eid, user_id)

    cid = (chat_id or "").strip()
    if cid:
        bound = await resolve_chat_delivery_from_meta(cid, user_id)
        if bound and prefer_direct and bound.channel == DELIVERY_CHANNEL_REGISTER:
            log.info(
                "[DeliveryRoute] ignore register binding for docker direct employee=%s chat=%s",
                eid,
                cid,
            )
            bound = None
        if bound:
            if bound.channel == DELIVERY_CHANNEL_REGISTER:
                if bound.instance_id and taiji_manager.is_connected(bound.instance_id):
                    return bound
            elif bound.channel == DELIVERY_CHANNEL_DIRECT:
                ws_url = (bound.ws_url or "").strip()
                if ws_url:
                    try:
                        from open_webui.apps.taiji.services.employee.deployment import (
                            endpoint_resolver,
                        )

                        details = await endpoint_resolver.resolve_with_details(
                            eid, owner_uid
                        )
                        current_url = (
                            str(details.get("url") or "").strip() if details else ""
                        )
                        if not current_url or current_url == ws_url:
                            return bound
                        log.info(
                            "[DeliveryRoute] ignore stale direct binding chat=%s employee=%s old=%s new=%s",
                            cid,
                            eid,
                            ws_url,
                            current_url,
                        )
                    except Exception as exc:
                        log.debug(
                            "[DeliveryRoute] direct binding validation failed chat=%s: %s",
                            cid,
                            exc,
                        )

    if not prefer_direct:
        instance_id = await resolve_registered_instance_for_employee(
            eid,
            user_id,
            is_admin=is_admin,
            skip_employee_id_fallback=False,
        )
        if instance_id and taiji_manager.is_connected(instance_id):
            return EmployeeDeliveryRoute(
                channel=DELIVERY_CHANNEL_REGISTER,
                instance_id=instance_id,
            )

    from open_webui.apps.taiji.services.employee.deployment import endpoint_resolver

    try:
        details = await endpoint_resolver.resolve_with_details(eid, owner_uid)
    except Exception:
        details = None
    if details and details.get("url"):
        return EmployeeDeliveryRoute(
            channel=DELIVERY_CHANNEL_DIRECT,
            ws_url=str(details["url"]),
        )
    return None


async def send_payload_via_delivery_route(
    route: EmployeeDeliveryRoute,
    payload: Dict[str, Any],
    *,
    employee_id: str,
    user_id: str,
    is_admin: bool = False,
) -> bool:
    """经指定投递通道 fire-and-forget 发送载荷。"""
    if route.channel == DELIVERY_CHANNEL_REGISTER and route.instance_id:
        return await taiji_manager.send_to_instance(route.instance_id, payload)

    owner_uid = user_id
    row = await employee_repository.get_by_id(employee_id, None if is_admin else user_id)
    if row and row.user_id:
        owner_uid = str(row.user_id)

    client = await TaijiConnectionPool.get_client(
        employee_id=employee_id,
        user_id=owner_uid,
        is_admin=is_admin,
        force_direct=True,
        ws_url=route.ws_url,
    )
    return await client.send_payload(payload)


async def _await_registered_response(instance_id: str, future: asyncio.Future):
    """
    等待注册实例响应：
    - timeout > 0：总超时等待
    - timeout <= 0：无限等待，但每 30 秒检查连接存活
    """
    timeout_seconds = get_session_wait_timeout_seconds(3600)
    if timeout_seconds > 0:
        return await asyncio.wait_for(future, timeout=timeout_seconds)

    start = time.time()
    while True:
        try:
            return await asyncio.wait_for(future, timeout=30)
        except asyncio.TimeoutError:
            if not taiji_manager.is_connected(instance_id):
                elapsed = int(time.time() - start)
                raise Exception(
                    f"Taiji 实例 {instance_id} 连接已断开（等待 {elapsed}s）"
                )
            continue


def _registered_response_text_and_ids(
    result: Any,
) -> tuple[str, Optional[str], Optional[str]]:
    """
    与远端 type=response 的 data 载荷一致（同 taiji_manager 传入的 message[\"data\"]、TaijiAS 终包）：
    content 或 text、session_id、message_id。
    """
    if not isinstance(result, dict):
        return "", None, None

    def _opt_str(v: Any) -> Optional[str]:
        if v is None:
            return None
        s = str(v).strip()
        return s or None

    body = result.get("content")
    if body is None:
        body = result.get("text")
    text = str(body).strip() if body is not None else ""
    return text, _opt_str(result.get("session_id")), _opt_str(result.get("message_id"))


def normalize_workspace_dir(value: Any) -> Optional[str]:
    """非空 workspace_dir；None / 空串 / 纯空白 → None。"""
    if value in (None, ""):
        return None
    s = str(value).strip()
    return s or None


def resolve_workspace_dir_for_chat(
    request_workspace_dir: Optional[str],
    chat,
) -> Optional[str]:
    """
    解析下发用的 workspace_dir：请求优先，否则回退 chat.meta.workspace_dir。
    """
    wd = normalize_workspace_dir(request_workspace_dir)
    if wd:
        return wd
    if not chat:
        return None
    return normalize_workspace_dir(merged_chat_meta(chat).get("workspace_dir"))


def resolve_team_session_run_mode(
    employee_id: Optional[str],
    chat,
) -> Optional[str]:
    """
    组对话下发 data.run_mode：仅主持人为 a2a_supporter，其他组员不下发会话级模式。
    """
    eid = (employee_id or "").strip()
    if not eid or chat is None:
        return None
    meta = merged_chat_meta(chat)
    host_id = (meta.get("team_host_employee_id") or "").strip()
    team_id = (meta.get("team_id") or "").strip()
    if not host_id and not team_id:
        return None
    from open_webui.apps.taiji.services.employee.run_mode import HOST_SESSION_RUN_MODE

    if host_id and eid == host_id:
        return HOST_SESSION_RUN_MODE
    return None


async def resolve_taiji_session_id_for_question_response(
    *,
    chat_id: str,
    user_id: str,
    session_id: str,
) -> str:
    """
    解析 question_response 的 taiji session 路由键。

    优先级：chat.meta.taiji_session_id → 请求体 session_id → chat_id（与 abort 一致）。
    """
    taiji_session_id = ""
    cid = (chat_id or "").strip()
    if cid:
        chat = await Chats.get_chat_by_id_and_user_id(cid, user_id)
        if chat:
            taiji_session_id = (
                merged_chat_meta(chat).get("taiji_session_id") or ""
            ).strip()
    explicit = (session_id or "").strip()
    if not taiji_session_id and explicit:
        taiji_session_id = explicit
    if not taiji_session_id and cid:
        taiji_session_id = cid
    return taiji_session_id


async def resolve_registered_instance_for_employee(
    employee_id: str,
    user_id: str,
    *,
    is_admin: bool = False,
    skip_employee_id_fallback: bool = False,
) -> Optional[str]:
    """
    解析当前用户（或管理员）视角下，该员工是否已有反向注册的 Taiji 实例在线。
    候选顺序：container_name -> container_id -> employee_id（可跳过最后一项，避免 Docker 直连被 Client Engine 抢路由）。
    """
    if not employee_id or employee_id == "taiji":
        return None
    row = await employee_repository.get_by_id(employee_id, None if is_admin else user_id)
    if not row:
        return None
    candidates = [
        (getattr(row, "container_name", None) or "").strip(),
        (getattr(row, "container_id", None) or "").strip(),
    ]
    if not skip_employee_id_fallback:
        candidates.append(str(employee_id).strip())
    for iid in candidates:
        if iid and taiji_manager.is_connected(iid):
            return iid
    return None


async def resolve_registered_instance_for_chat(
    chat_id: str, user_id: str
) -> Optional[str]:
    """
    根据聊天解析当前可用的已注册 Taiji 实例 ID（含 Docker 投递策略）。
    """
    chat = await Chats.get_chat_by_id_and_user_id(chat_id, user_id)
    if not chat:
        return None
    employee_id = get_chat_employee_id(chat)
    if not employee_id:
        return None
    return await resolve_registered_instance_with_policy(
        str(employee_id), user_id, is_admin=False
    )


async def send_abort_for_chat(
    *,
    user_id: str,
    chat_id: str,
    sid: Optional[str],
    taiji_session_id: Optional[str],
) -> bool:
    """按 chat 绑定的投递通道发送 abort（register 或 direct，二选一）。"""
    chat = await Chats.get_chat_by_id_and_user_id(chat_id, user_id)
    if not chat:
        return False
    employee_id = get_chat_employee_id(chat)
    if not employee_id:
        return False

    route = await resolve_employee_delivery_route(
        str(employee_id), user_id, chat_id=chat_id
    )
    if not route:
        return False

    effective_session = taiji_session_id or chat_id
    if route.channel == DELIVERY_CHANNEL_REGISTER and route.instance_id:
        return await send_abort_via_registered_instance(
            instance_id=route.instance_id,
            user_id=user_id,
            chat_id=chat_id,
            taiji_session_id=taiji_session_id,
        )

    abort_payload = {
        "type": "abort",
        "request_id": str(uuid.uuid4()),
        "data": {
            "metadata": {"user_id": user_id, "chat_id": chat_id},
            "session_id": effective_session,
        },
    }
    sent = await send_payload_via_delivery_route(
        route,
        abort_payload,
        employee_id=str(employee_id),
        user_id=user_id,
    )
    if sent:
        return True

    result = await relay_manager.abort_session(
        user_id=user_id,
        chat_id=chat_id,
        sid=sid,
        session_id=effective_session,
    )
    return bool(result.get("status"))


async def send_steer_to_employee(
    *,
    employee_id: str,
    user_id: str,
    chat_id: str,
    sid: Optional[str],
    text: str,
    message_id: Optional[str],
    taiji_session_id: Optional[str],
    sender_name: Optional[str] = None,
    fallback_instance_id: Optional[str] = None,
) -> None:
    """按 chat 绑定的投递通道发送 /steer 注入。"""
    route = await resolve_employee_delivery_route(
        employee_id, user_id, chat_id=chat_id
    )
    if not route and fallback_instance_id:
        fid = fallback_instance_id.strip()
        if fid and taiji_manager.is_connected(fid):
            route = EmployeeDeliveryRoute(
                channel=DELIVERY_CHANNEL_REGISTER,
                instance_id=fid,
            )
    if not route:
        raise Exception(f"员工 {employee_id} 未在线，无法注入消息")

    request_id = str(uuid.uuid4())
    steer_text = text if text.startswith("/steer ") else f"/steer {text}"
    payload = {
        "type": "message",
        "request_id": request_id,
        "data": {
            "text": steer_text,
            "message": steer_text,
            "sender_name": sender_name or user_id,
            "metadata": {
                "user_id": user_id,
                "chat_id": chat_id,
                "message_id": message_id,
            },
        },
    }
    if taiji_session_id:
        payload["data"]["session_id"] = taiji_session_id

    if route.channel == DELIVERY_CHANNEL_REGISTER and route.instance_id:
        if not taiji_manager.is_connected(route.instance_id):
            raise Exception(f"Taiji 实例 {route.instance_id} 未连接")
        success = await taiji_manager.send_to_instance(route.instance_id, payload)
        if not success:
            raise Exception(f"Taiji 实例 {route.instance_id} 发送失败")
        log.info(
            "[Steer] 注入已发送 channel=register instance_id=%s request_id=%s",
            route.instance_id,
            request_id,
        )
        return

    ok = await send_payload_via_delivery_route(
        route,
        payload,
        employee_id=employee_id,
        user_id=user_id,
    )
    if not ok:
        raise Exception(f"Steer 直连发送失败 employee={employee_id}")
    log.info(
        "[Steer] 注入已发送 channel=direct employee=%s request_id=%s ws=%s",
        employee_id,
        request_id,
        route.ws_url or "",
    )


async def send_abort_via_registered_instance(
    *,
    instance_id: str,
    user_id: str,
    chat_id: str,
    taiji_session_id: Optional[str],
) -> bool:
    """
    经反向注册 WebSocket 向实例发送 abort，与 taiji_as 客户端协议一致（单向通知，不等待响应）。
    """
    if not taiji_manager.is_connected(instance_id):
        return False
    rid = str(uuid.uuid4())
    data: Dict[str, Any] = {
        "metadata": {"user_id": user_id, "chat_id": chat_id},
        "session_id": chat_id
    }
    if taiji_session_id:
        data["session_id"] = taiji_session_id

    payload = {"type": "abort", "request_id": rid, "data": data}
    return await taiji_manager.send_to_instance(instance_id, payload)


async def send_steer_via_registered_instance(
    *,
    instance_id: str,
    user_id: str,
    chat_id: str,
    sid: Optional[str],
    text: str,
    message_id: Optional[str],
    taiji_session_id: Optional[str],
    sender_name: Optional[str] = None,
) -> None:
    """
    经反向注册 WebSocket 向实例发送 /steer 注入。

    不注册 relay 流上下文（避免覆盖主生成流的 conn_key），
    不等待业务 ack（taiji-go 不会发送）。WS 层 send_to_instance bool
    即确认消息送达客户端；业务侧的注入结果由 steer_message_injected
    event 经 relay 自然回流。
    """
    if not taiji_manager.is_connected(instance_id):
        raise Exception(f"Taiji 实例 {instance_id} 未连接")

    request_id = str(uuid.uuid4())
    steer_text = text if text.startswith("/steer ") else f"/steer {text}"
    payload = {
        "type": "message",
        "request_id": request_id,
        "data": {
            "text": steer_text,
            "message": steer_text,
            "sender_name": sender_name or user_id,
            "metadata": {
                "user_id": user_id,
                "chat_id": chat_id,
                "message_id": message_id,
            },
        },
    }
    if taiji_session_id:
        payload["data"]["session_id"] = taiji_session_id

    success = await taiji_manager.send_to_instance(instance_id, payload)
    if not success:
        raise Exception(f"Taiji 实例 {instance_id} 发送失败")

    log.info(
        "[Steer] 注入已发送 instance_id=%s request_id=%s",
        instance_id,
        request_id,
    )


async def send_question_response_to_employee(
    employee_id: str,
    user_id: str,
    question_id: str,
    answers: Dict[str, Any],
    *,
    chat_id: str = "",
    session_id: str = "",
    is_admin: bool = False,
) -> None:
    """
    向员工 Taiji 客户端发送 question_response（fire-and-forget）。

    消息格式与 taiji 协议一致::
        {"type": "question_response", "request_id": "...",
         "employee_id": "...",
         "data": {"question_id": "...", "session_id": "...", "answers": {...}}}
    """
    eid = (employee_id or "").strip()
    qid = (question_id or "").strip()
    if not eid or not qid:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="question_id 无效",
        )
    if not isinstance(answers, dict) or not answers:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="answers 不能为空",
        )

    row = await employee_repository.get_by_id(eid, None if is_admin else user_id)
    if not row:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="员工不存在",
        )
    if not is_admin and not await is_employee_readable(
        str(row.user_id),
        row.access_control if isinstance(row.access_control, dict) else None,
        user_id,
        is_admin=False,
    ):
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="无权访问该员工",
        )

    normalized_answers = {
        str(key): str(value)
        for key, value in answers.items()
        if key is not None and str(key).strip()
    }
    if not normalized_answers:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="answers 不能为空",
        )

    taiji_session_id = await resolve_taiji_session_id_for_question_response(
        chat_id=chat_id,
        user_id=user_id,
        session_id=session_id,
    )

    payload = {
        "type": "question_response",
        "request_id": qid,
        "employee_id": eid,
        "data": {
            "question_id": qid,
            "session_id": taiji_session_id,
            "answers": normalized_answers,
        },
    }

    route = await resolve_employee_delivery_route(
        eid,
        user_id,
        chat_id=chat_id,
        is_admin=is_admin,
    )
    if route:
        if await send_payload_via_delivery_route(
            route,
            payload,
            employee_id=eid,
            user_id=user_id,
            is_admin=is_admin,
        ):
            log.info(
                "[QuestionResponse] sent channel=%s employee=%s question=%s session=%s instance=%s ws=%s",
                route.channel,
                eid,
                qid,
                taiji_session_id,
                route.instance_id or "",
                route.ws_url or "",
            )
            return
        log.warning(
            "[QuestionResponse] %s 发送失败 employee=%s question=%s session=%s instance=%s ws=%s",
            route.channel,
            eid,
            qid,
            taiji_session_id,
            route.instance_id or "",
            route.ws_url or "",
        )

    raise HTTPException(
        status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
        detail="员工未在线，无法发送 question_response",
    )


async def _send_message_via_registered_instance(
    *,
    instance_id: str,
    user_id: str,
    chat_id: str,
    sid: Optional[str],
    text: str,
    sender_name: str,
    message_id: Optional[str],
    taiji_session_id: Optional[str],
    workspace_dir: Optional[str] = None,
    run_mode: Optional[str] = None,
) -> Dict[str, Any]:
    """
    通过 taiji 注册 WebSocket 通道发送消息；正文统一走远端 type=event 流式，
    经 taiji_manager → relay 转 Socket.IO；终包由 type=response 经同路径 _forward_to_frontend 推送 done。
    """
    request_id = str(uuid.uuid4())
    loop = asyncio.get_event_loop()
    future = loop.create_future()
    relay_manager.register_registered_stream(
        request_id,
        instance_id,
        user_id,
        chat_id,
        sid or "",
        message_id,
    )
    try:
        taiji_manager.pending_requests[request_id] = future
        try:
            # 与 TaijiASWebSocketClient.send_message 一致；data.message 兼容只识别 message 字段的远端
            payload = {
                "type": "message",
                "request_id": request_id,
                "data": {
                    "text": text,
                    "message": text,
                    "sender_name": sender_name,
                    "metadata": {"user_id": user_id, "chat_id": chat_id},
                },
            }
            if taiji_session_id:
                payload["data"]["session_id"] = taiji_session_id
            if workspace_dir:
                payload["data"]["workspace_dir"] = workspace_dir
            if run_mode:
                payload["data"]["run_mode"] = run_mode

            success = await taiji_manager.send_to_instance(instance_id, payload)
            if not success:
                raise Exception(f"Taiji 实例 {instance_id} 发送失败")

            try:
                result = await _await_registered_response(instance_id, future)
            except Exception as e:
                log.warning(
                    "[WebSocketRelay] 等待反向注册实例 response 失败/超时: instance=%s request_id=%s: %s",
                    instance_id,
                    request_id,
                    e,
                )
                mid = message_id
                if mid:
                    await broadcast_to_user(
                        user_id,
                        "chat-events",
                        {
                            "chat_id": chat_id,
                            "message_id": mid,
                            "data": {
                                "type": "chat:completion",
                                "data": {
                                    "error": str(e),
                                    "content": f"\n[错误: {e}]",
                                },
                            },
                        },
                        sid=sid if sid else None,
                    )
                raise

            content, _resp_sid, resp_mid = _registered_response_text_and_ids(result)
            if not (message_id or resp_mid):
                log.warning(
                    "[WebSocketRelay] 无 message_id（请求与 response 均未提供），"
                    "前端可能无法对齐流式消息: chat_id=%s",
                    chat_id,
                )

            # session_id 以平台下发为准，不采用客户端回传覆盖 meta
            return {
                "message_id": resp_mid,
                "session_id": taiji_session_id,
            }
        finally:
            taiji_manager.pending_requests.pop(request_id, None)
    finally:
        relay_manager.clear_registered_stream(request_id)


async def _resolve_employee_internal_url(employee_id: str, user_id: str) -> Optional[str]:
    """
    根据员工 id 解析 WebSocket 连接地址。

    委托给 EndpointResolver，集中管理优先级逻辑：
    1. 用户配置的 internal_url（最高优先级）
    2. 从 taiji_containers 表查询
       - remote 类型：使用 endpoint_url
       - docker 类型：构造 ws://{name}:19901
    """
    if not employee_id or employee_id == "taiji":
        return None

    try:
        from open_webui.apps.taiji.services.employee.deployment import endpoint_resolver

        url = await endpoint_resolver.resolve(employee_id, user_id)
        return url
    except Exception:
        return None


def _extract_taiji_outbound_text(message: dict) -> str:
    """
    从最后一条用户消息提取发往 Gateway 的纯文本。

    兼容 str content，以及 process_image_url 产生的多模态 list；
    绝不把 list/None 原样当作 data.text。
    """
    if not isinstance(message, dict):
        return ""
    raw = get_content_from_message(message)
    if raw is None and "content" in message:
        raw = message.get("content")
    if isinstance(raw, list):
        # 兜底：get_content_from_message 未命中 text 块时再扫一遍
        parts = []
        for item in raw:
            if isinstance(item, dict) and item.get("type") == "text":
                parts.append(str(item.get("text") or ""))
        raw = "".join(parts)
    if not isinstance(raw, str):
        return ""
    return raw.strip()


async def process_session_with_taiji_websocket(request, form_data, user, metadata, model, events, tasks):
    """
    WebSocket 中继代理模式（透传转发）
    直接通过 Socket.IO 和 WebSocket 透传消息到 taiji_as，参照 client.py 协议。
    支持按员工的 internal_url 连接；未配置时使用环境变量或默认地址。
    员工 id 优先从当前聊天的 instance_id 取（与员工对话的聊天在创建时已绑定），
    避免请求里 model 被自动选模（如 qwen-plus）覆盖导致连错地址。
    失败时直接抛出异常，由调用方决定是否回退到其他流程。
    """
    chat_id = metadata.get("chat_id")
    message_id = metadata.get("message_id")
    user_id = user.id
    sid = metadata.get("session_id")  # Socket.IO session id，用于按 session 隔离连接

    chat = await Chats.get_chat_by_id_and_user_id(chat_id, user_id)
    employee_id_from_chat = get_chat_employee_id(chat) if chat else None
    employee_id = (
        (form_data.get("employee_id") or "").strip() or None
        or employee_id_from_chat
        or (model.get("id") if isinstance(model, dict) else None)
        or form_data.get("model", "taiji")
    )
    messages = form_data.get("messages") if isinstance(form_data.get("messages"), list) else []
    last_user_text = ""
    for msg in reversed(messages):
        if isinstance(msg, dict) and msg.get("role") == "user":
            last_user_text = _extract_taiji_outbound_text(msg)
            break
    if last_user_text:
        from open_webui.apps.taiji.services.chat_slash_skill import (
            ensure_slash_skill_on_employee,
        )

        await ensure_slash_skill_on_employee(employee_id, user_id, last_user_text)
    internal_url = await _resolve_employee_internal_url(employee_id, user_id)
    employee_row = await employee_repository.get_by_id(employee_id, user_id)
    prefer_direct = await should_prefer_direct_delivery(
        employee_row, employee_id, user_id
    )
    registered_instance_id = None
    if not prefer_direct:
        registered_instance_id = await resolve_registered_instance_for_employee(
            employee_id,
            user_id,
            is_admin=False,
            skip_employee_id_fallback=False,
        )

    messages = form_data.get("messages", [])
    if not messages:
        raise Exception("No messages to send")

    last_message = messages[-1]
    # process_image_url 在视觉模型下会把 content 转成多模态 list；Gateway 的 data.text 只接受非空字符串
    text = _extract_taiji_outbound_text(last_message)
    if not text:
        raise Exception("Message text cannot be empty")
    sender_name = last_message.get("role", "User")
    workspace_dir = resolve_workspace_dir_for_chat(
        metadata.get("workspace_dir"),
        chat,
    )
    run_mode = resolve_team_session_run_mode(employee_id, chat)

    taiji_session_id = None
    if chat:
        taiji_session_id = merged_chat_meta(chat).get("taiji_session_id")
    # 首次对话 meta 尚无 taiji_session_id 时回退 chat_id（与 abort / question_response 一致）
    taiji_session_id = (taiji_session_id or "").strip() or (chat_id or "").strip() or None

    await remember_session_coding_model(
        request,
        form_data=form_data,
        model=model,
        chat_id=chat_id,
        taiji_session_id=taiji_session_id,
        employee_id=employee_id,
        employee_meta=(employee_row.meta if employee_row else None),
    )

    # 优先走已注册实例的反向 WebSocket 通道（Docker 直连员工跳过 register）
    if registered_instance_id:
        log.info(
            "[TaijiMessage] sent channel=register employee=%s chat=%s instance=%s",
            employee_id,
            chat_id,
            registered_instance_id,
        )
        result = await _send_message_via_registered_instance(
            instance_id=registered_instance_id,
            user_id=user_id,
            chat_id=chat_id,
            sid=sid,
            text=text,
            sender_name=sender_name,
            message_id=message_id,
            taiji_session_id=taiji_session_id,
            workspace_dir=workspace_dir,
            run_mode=run_mode,
        )
        await persist_chat_delivery_channel(
            user_id=user_id,
            chat_id=chat_id,
            channel=DELIVERY_CHANNEL_REGISTER,
            instance_id=registered_instance_id,
        )
        return {
            "status": "accepted",
            "message": "消息已通过注册实例通道发送",
            "chat_id": chat_id,
            "message_id": message_id,
            "taiji_session_id": taiji_session_id,
            "taiji_message_id": result.get("message_id"),
        }

    actual_ws_url = await relay_manager.resolve_ws_url(
        user_id, internal_url, employee_id
    )
    client = await relay_manager.get_client(
        user_id,
        chat_id,
        sid=sid,
        internal_url=internal_url,
        employee_id=employee_id,
        ws_url=actual_ws_url,
    )
    if not client:
        # 对本应走 register 的反向注册员工：direct 回退失败大概率是重启/重连窗口期
        # （实例收到 config_sync 后断开 WS 重载技能）。短等待 register 恢复后重试一次，
        # 仍不可达再抛错，避免窗口期内首条消息硬失败。
        if not prefer_direct and employee_id and employee_id != "taiji":
            log.info(
                "[TaijiMessage] direct 失败，等待 register 恢复重试 employee=%s chat=%s",
                employee_id,
                chat_id,
            )
            deadline = time.time() + REGISTER_RECONNECT_WAIT_SECONDS
            recovered_instance = None
            while time.time() < deadline:
                await asyncio.sleep(REGISTER_RECONNECT_POLL_SECONDS)
                recovered_instance = await resolve_registered_instance_for_employee(
                    employee_id,
                    user_id,
                    is_admin=False,
                    skip_employee_id_fallback=False,
                )
                if recovered_instance and taiji_manager.is_connected(recovered_instance):
                    break
                recovered_instance = None
            if recovered_instance:
                log.info(
                    "[TaijiMessage] register 恢复，改走 register employee=%s chat=%s instance=%s",
                    employee_id,
                    chat_id,
                    recovered_instance,
                )
                result = await _send_message_via_registered_instance(
                    instance_id=recovered_instance,
                    user_id=user_id,
                    chat_id=chat_id,
                    sid=sid,
                    text=text,
                    sender_name=sender_name,
                    message_id=message_id,
                    taiji_session_id=taiji_session_id,
                    workspace_dir=workspace_dir,
                    run_mode=run_mode,
                )
                await persist_chat_delivery_channel(
                    user_id=user_id,
                    chat_id=chat_id,
                    channel=DELIVERY_CHANNEL_REGISTER,
                    instance_id=recovered_instance,
                )
                return {
                    "status": "accepted",
                    "message": "消息已通过注册实例通道发送",
                    "chat_id": chat_id,
                    "message_id": message_id,
                    "taiji_session_id": taiji_session_id,
                    "taiji_message_id": result.get("message_id"),
                }
        raise Exception("Failed to connect to taiji_as WebSocket")

    log.info(
        "[TaijiMessage] sent channel=direct employee=%s chat=%s ws=%s",
        employee_id,
        chat_id,
        actual_ws_url or "",
    )

    result = await relay_manager.send_message(
        user_id=user_id,
        chat_id=chat_id,
        text=text,
        session_id=taiji_session_id,
        sender_name=sender_name,
        message_id=message_id,
        sid=sid,
        internal_url=internal_url,
        employee_id=employee_id,
        ws_url=actual_ws_url,
        workspace_dir=workspace_dir,
        run_mode=run_mode,
    )

    # 发送是否成功的判定只看 result 是否为 None（relay_manager.send_message 在
    # 内部已 try/except，失败时返回 None）。result.message_id 缺失不代表发送失败：
    # event-stream 协议下 taiji-go 用 type=event 回报流式输出，不再回 type=response
    # 包，因此 message_id 永远为 None，但消息确实送达、agent 也通过反向 WS 推送内容。
    if result is None:
        raise Exception("Failed to send message to taiji_as")

    await persist_chat_delivery_channel(
        user_id=user_id,
        chat_id=chat_id,
        channel=DELIVERY_CHANNEL_DIRECT,
        ws_url=actual_ws_url,
    )

    # result_message_id 可能为 None（见上方注释），属正常；返回给调用方时如实透传
    result_message_id = result.get("message_id") if result else None

    return {
        "status": "accepted",
        "message": "消息已发送，流式响应通过 WebSocket 推送",
        "chat_id": chat_id,
        "message_id": message_id,
        "taiji_session_id": taiji_session_id,
        "taiji_message_id": result_message_id,
    }


async def verify_employee_for_taiji_session(request, form_data: dict, user) -> None:
    """
    对话前校验：若会话指向 taiji_employee 中的员工且已软删除，则 400「此员工已经被解雇」；
    若该 id 在表中有记录且未删除但当前用户无权使用，则 403。
    非员工表主键（如普通 LLM model id）不校验。
    """
    chat_id = form_data.get("chat_id")
    user_id = user.id
    model_id = form_data.get("model", "taiji")
    model = (
        getattr(request.app.state, "MODELS", None) and request.app.state.MODELS.get(model_id)
    )
    if model is None:
        model = {"id": "taiji", "name": "Taiji"}

    chat = await Chats.get_chat_by_id_and_user_id(chat_id, user_id) if chat_id else None
    employee_id_from_chat = (
        (getattr(chat, "instance_id", None) or (chat.meta or {}).get("instance_id"))
        if chat
        else None
    )
    employee_id = (
        (form_data.get("employee_id") or "").strip() or None
        or employee_id_from_chat
        or (model.get("id") if isinstance(model, dict) else None)
        or form_data.get("model", "taiji")
    )
    if not employee_id or employee_id == "taiji":
        return

    row_any = await employee_repository.get_by_id_including_deleted(employee_id)
    if row_any is None:
        return
    if row_any.is_deleted:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="此员工已经被解雇",
        )
    if not await employee_repository.get_by_id(employee_id, user_id):
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="无权与此员工对话",
        )


async def _resolve_file_docker_paths(files_list: list) -> Optional[list]:
    """
    从文件列表中解析出已在 Taiji 容器内的文件（有 docker_path 元数据）。

    Args:
        files_list: 文件列表，每个元素为 {id, ...} 或 id 字符串

    Returns:
        [{id, name, docker_path}, ...] 或 None
    """
    if not files_list:
        return None
    from open_webui.models.files import Files

    result = []
    for f in files_list:
        fid = f.get("id") if isinstance(f, dict) else f
        if not fid:
            continue
        file_record = await Files.get_file_by_id(fid)
        if file_record and file_record.meta and file_record.meta.get("docker_path"):
            result.append({
                "id": file_record.id,
                "name": file_record.filename,
                "docker_path": file_record.meta["docker_path"],
            })
    return result or None


def _file_entry_id(entry: Any) -> Optional[str]:
    """从顶层 files 或 message.files 单条解析 file id。"""
    if isinstance(entry, str):
        s = entry.strip()
        return s or None
    if not isinstance(entry, dict):
        return None
    fid = entry.get("id")
    if fid:
        s = str(fid).strip()
        return s or None
    inner = entry.get("file")
    if isinstance(inner, dict) and inner.get("id"):
        s = str(inner.get("id")).strip()
        return s or None
    return None


def _collect_file_ids_from_taiji_form(form_data: dict) -> List[str]:
    """汇总本次请求中的附件 id：顶层 files + 最后一条 user 消息的 files，去重保序。"""
    seen: set[str] = set()
    ordered: List[str] = []

    def add(fid: Optional[str]) -> None:
        if not fid:
            return
        if fid in seen:
            return
        seen.add(fid)
        ordered.append(fid)

    for entry in form_data.get("files") or []:
        add(_file_entry_id(entry))

    messages = form_data.get("messages") or []
    if messages:
        last = messages[-1]
        if isinstance(last, dict) and last.get("role") == "user":
            for entry in last.get("files") or []:
                add(_file_entry_id(entry))

    return ordered


async def prepend_taiji_file_paths_to_messages(
    form_data: dict, user, metadata: Optional[Dict[str, Any]] = None
) -> None:
    """
    将当前用户可见的附件路径拼到最后一条 user 消息的 content 前面（仅 str content）。
    路径以 file 表为准（get_file_by_id_and_user_id），不采用客户端上报 path。
    若 meta.docker_path 存在（Taiji Docker 上传时 file_sync 写入），优先使用该容器内路径；
    否则使用 file.path 作为服务器本地路径。
    若传入 metadata 且 user_message 与最后一条 user 消息 id 相同，则同步改写 user_message.content，
    便于 save_chat_messages_on_request 落库与发往 Taiji 一致。
    """
    from open_webui.models.files import Files

    messages = form_data.get("messages") or []
    if not messages:
        return
    last = messages[-1]
    if not isinstance(last, dict) or last.get("role") != "user":
        return
    content = last.get("content")
    if not isinstance(content, str):
        log.debug(
            "[TaijiSession] 跳过路径前缀：最后一条 user 消息的 content 非字符串"
        )
        return

    ids = _collect_file_ids_from_taiji_form(form_data)
    if not ids:
        return

    user_id = getattr(user, "id", None) or str(user)
    lines: List[str] = []
    for fid in ids:
        record = await Files.get_file_by_id_and_user_id(fid, user_id)
        if not record:
            continue
        name = record.filename or fid
        meta = record.meta if isinstance(record.meta, dict) else {}
        docker_path = str(meta.get("docker_path") or "").strip()
        server_path = (record.path or "").strip()
        display_path = docker_path or server_path
        if display_path:
            lines.append(f"- {name}: {display_path}")
        else:
            lines.append(f"- {name}")

    if not lines:
        return

    prefix = "附件路径：\n" + "\n".join(lines)
    new_content = prefix + "\n\n" + content
    last["content"] = new_content

    if metadata:
        um = metadata.get("user_message")
        if (
            isinstance(um, dict)
            and um.get("id")
            and um.get("id") == last.get("id")
            and isinstance(um.get("content"), str)
        ):
            um["content"] = new_content


def _knowledge_collection_lines(entry: Any) -> List[tuple[str, str]]:
    """
    从单条 knowledge / collection 配置解析 (显示名, collection_id)。
    collection_id 与 taiji-knowledge-search skill 的 --collection 一致。
    """
    if not isinstance(entry, dict):
        return []
    if entry.get("type") == "file" and not entry.get("legacy"):
        return []

    name = (
        str(entry.get("name") or entry.get("title") or "").strip() or "知识库"
    )

    if entry.get("type") == "collection":
        if entry.get("legacy") and entry.get("collection_names"):
            return [
                (name, str(cn).strip())
                for cn in entry.get("collection_names") or []
                if str(cn).strip()
            ]
        kid = entry.get("id")
        if kid:
            s = str(kid).strip()
            return [(name, s)] if s else []
        return []

    cn = entry.get("collection_name")
    if cn:
        s = str(cn).strip()
        return [(name, s)] if s else []

    if entry.get("collection_names"):
        return [
            (name, str(cn).strip())
            for cn in entry.get("collection_names") or []
            if str(cn).strip()
        ]

    if entry.get("legacy") and entry.get("id"):
        s = str(entry.get("id")).strip()
        return [(name, s)] if s else []

    return []


def _collect_knowledge_collections_from_taiji_form(
    form_data: dict, model: Optional[dict]
) -> List[tuple[str, str]]:
    """汇总模型绑定与本次请求中的知识库 collection，去重保序。"""
    seen: set[str] = set()
    ordered: List[tuple[str, str]] = []

    def add_lines(entry: Any) -> None:
        for display_name, kid in _knowledge_collection_lines(entry):
            if kid in seen:
                continue
            seen.add(kid)
            ordered.append((display_name, kid))

    model_knowledge = ((model or {}).get("info") or {}).get("meta", {}).get("knowledge")
    if model_knowledge:
        for item in model_knowledge:
            add_lines(item)

    for entry in form_data.get("files") or []:
        add_lines(entry)

    messages = form_data.get("messages") or []
    if messages:
        last = messages[-1]
        if isinstance(last, dict) and last.get("role") == "user":
            for entry in last.get("files") or []:
                add_lines(entry)

    return ordered


def prepend_taiji_knowledge_ids_to_messages(
    form_data: dict, metadata: Optional[Dict[str, Any]] = None
) -> None:
    """
    将知识库 collection id 拼到最后一条 user 消息 content 前（与附件路径前缀方式一致）。
    不在服务端做向量检索；由太极智能体通过 taiji-knowledge-search skill 按 id 查询。
    """
    messages = form_data.get("messages") or []
    if not messages:
        return
    last = messages[-1]
    if not isinstance(last, dict) or last.get("role") != "user":
        return
    content = last.get("content")
    if not isinstance(content, str):
        log.debug(
            "[TaijiSession] 跳过知识库前缀：最后一条 user 消息的 content 非字符串"
        )
        return

    model = (metadata or {}).get("model") if metadata else None
    collections = _collect_knowledge_collections_from_taiji_form(form_data, model)
    if not collections:
        return

    lines = [f"- {name}" for name, _kid in collections]
    prefix = (
        "可用知识库（请使用知识库检索 skill，--knowledge 参数，填下列名称即可）：\n"
        + "\n".join(lines)
    )
    new_content = prefix + "\n\n" + content
    last["content"] = new_content

    if metadata:
        um = metadata.get("user_message")
        if (
            isinstance(um, dict)
            and um.get("id")
            and um.get("id") == last.get("id")
            and isinstance(um.get("content"), str)
        ):
            um["content"] = new_content


async def run_taiji_session_completion(request, form_data, user) -> Dict[str, Any]:
    """
    执行 Taiji 对话：当前未调用 process_chat_payload；
    知识库由 prepend_taiji_knowledge_ids_to_messages 将 collection id 前缀到用户消息（由智能体 skill 检索）；
    附件由 prepend_taiji_file_paths_to_messages 将路径前缀到用户消息；
    再 process_image_url，最后 WebSocket 发往 taiji；按需 create_task 或直接执行。
    """
    if not request.app.state.MODELS or getattr(user, "role", "user") == "admin":
        await get_all_models(request, user=user)

    model_item = form_data.pop("model_item", {})
    tasks_bg = form_data.pop("background_tasks", None)
    model_id = form_data.get("model", None)
    metadata: Dict[str, Any] = {}


    try:
        pipeline_model_id = await get_pipeline_model_for_user(form_data, user)
        if pipeline_model_id:
            model_id = pipeline_model_id
            form_data["model"] = model_id

        model_info = None
        if not model_item.get("direct", False):
            if model_id not in request.app.state.MODELS:
                # 员工对话 model 常为 UUID，不在 MODELS：按 direct 走 payload 处理
                request.state.direct = True
                model = {
                    "id": model_id if model_id else "taiji",
                    "name": "Taiji",
                    "info": {"capabilities": ["vision"]},
                }
                request.state.model = model
            else:
                model = request.app.state.MODELS[model_id]
                model_info = await Models.get_model_by_id(model_id)
                if not BYPASS_MODEL_ACCESS_CONTROL and getattr(user, "role", "user") == "user":
                    await check_model_access(user, model)
        else:
            model = model_item
            model_info = None
            request.state.direct = True
            request.state.model = model

        is_new_chat, message_ids, user_message_obj, parent_id = extract_chat_fields_from_form(form_data, model_id)

        raw_workspace_dir = form_data.pop("workspace_dir", None)
        workspace_dir = normalize_workspace_dir(raw_workspace_dir)

        metadata = {
            "user_id": user.id,
            "chat_id": form_data.pop("chat_id", None),
            "message_id": list(message_ids.values())[0] if message_ids else None,
            "user_message": user_message_obj,
            "user_message_id": user_message_obj.get("id") if user_message_obj else None,
            "session_id": form_data.pop("session_id", None),
            "folder_id": form_data.pop("folder_id", None),
            "filter_ids": form_data.pop("filter_ids", []),
            "tool_ids": form_data.get("tool_ids", None),
            "tool_servers": form_data.pop("tool_servers", None),
            "files": form_data.get("files", None),
            "features": form_data.get("features", {}),
            "variables": form_data.get("variables", {}),
            "workspace_dir": workspace_dir,
            "model": model,
            "direct": model_item.get("direct", False)
            or getattr(request.state, "direct", False),
            **(
                {"function_calling": "native"}
                if form_data.get("params", {}).get("function_calling") == "native"
                or (
                    model_info
                    and model_info.params.model_dump().get("function_calling")
                    == "native"
                )
                else {}
            ),
        }
        
        await save_chat_messages_on_request(form_data, metadata, message_ids, is_new_chat, user)

        if workspace_dir and metadata.get("chat_id"):
            await persist_merged_chat_meta(
                user.id,
                metadata["chat_id"],
                {"workspace_dir": workspace_dir},
            )

        request.state.metadata = metadata
        form_data["metadata"] = metadata

        # 知识库：仅带入 collection id，由太极智能体 skill 检索（不做服务端 RAG）
        prepend_taiji_knowledge_ids_to_messages(form_data, metadata)

        # 附件：将文件路径前缀到最后一条用户消息（优先 meta.docker_path，否则 path）
        await prepend_taiji_file_paths_to_messages(form_data, user, metadata)
    except Exception as e:
        log.debug(f"[TaijiSession] Error preparing taiji session completion: {e}")
        if metadata.get("chat_id") and metadata.get("message_id"):
            await Chats.upsert_message_to_chat_by_id_and_message_id(
                metadata["chat_id"],
                metadata["message_id"],
                {"error": {"content": str(e)}},
            )
        if isinstance(e, HTTPException):
            raise e
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=str(e),
        )

    events: List[Any] = []
    model = metadata.get("model", model)

    async def process_taiji_chat(request, form_data, user, metadata, model):
        try:
            await process_image_url(form_data["messages"], metadata)
            return await process_session_with_taiji_websocket(
                request, form_data, user, metadata, model, events, tasks_bg
            )
        except Exception as e:
            log.exception(
                "[TaijiSession] 后台会话任务失败 chat_id=%s: %s",
                metadata.get("chat_id"),
                e,
            )
            cid, mid = metadata.get("chat_id"), metadata.get("message_id")
            if cid and mid:
                await Chats.upsert_message_to_chat_by_id_and_message_id(
                    cid,
                    mid,
                    {"error": {"content": str(e)}},
                )
            raise

    if (
        metadata.get("session_id")
        and metadata.get("chat_id")
        and metadata.get("message_id")
    ):
        task_metadata = {
            "user_id": user.id,
            "chat_id": metadata.get("chat_id"),
            "sid": metadata.get("session_id"),
            "session_id": None,
        }

        chat = await Chats.get_chat_by_id_and_user_id(metadata.get("chat_id"), user.id)
        if chat and chat.meta:
            task_metadata["session_id"] = chat.meta.get("taiji_session_id")

        task_id, _ = await create_task(
            request.app.state.redis,
            process_taiji_chat(request, form_data, user, metadata, model),
            id=metadata["chat_id"],
            metadata=task_metadata,
        )
        return {"status": True, "task_id": task_id, "model": model_id}

    return await process_taiji_chat(request, form_data, user, metadata, model)