Tolan/backend/websocket_handler.py

"""
WebSocket处理器 - 新版本（STT+LLM+TTS流水线，支持流式LLM）
"""
import json
import base64
import asyncio
import time
import re
import logging
import traceback
from typing import Any, Dict, Literal, Optional
from fastapi import WebSocket
from starlette.websockets import WebSocketDisconnect

logger = logging.getLogger(__name__)

# 预编译正则表达式（性能优化）
EMOTION_TAG_PATTERN = re.compile(r'\[EMOTION:\w+\]')
# 表情符号正则表达式（用于TTS前清理）
EMOJI_PATTERN = re.compile(
    "["
    "\U0001F600-\U0001F64F"  # emoticons
    "\U0001F300-\U0001F5FF"  # symbols & pictographs
    "\U0001F680-\U0001F6FF"  # transport & map symbols
    "\U0001F1E0-\U0001F1FF"  # flags (iOS)
    "\U00002702-\U000027B0"
    "\U000024C2-\U0001F251"
    "]+", flags=re.UNICODE
)
from models import (
    AudioChunkMessage,
    AudioEndMessage,
    CancelMessage,
    TextResultMessage,
    EmotionMessage,
    AudioResponseMessage,
    ErrorMessage,
    PingMessage,
    PongMessage,
    EnergyStoneEventMessage,
    EnergyStoneSpawnMessage,
    UserLevelMessage,
)
from ai_service import ai_service
from request_logger import request_logger, PerformanceMetrics
from config import settings
from stt_service import stt_service, STTSession
from services.user_service import user_service
from services.energy_stone_service import energy_stone_service
from services.scenario_topic_service import scenario_topic_service
from services.scenario_gift_service import grant_scene_gift_if_needed
from scenario_config import (
    SCENE_DEFAULT,
    SCENE_WINDOW,
    SCENE_READING,
    SCENE_EATING,
    SCENE_WEIGHTS,
    SCENARIO_TOPICS,
    get_special_first_line,
    build_topics_payload,
)
import random
import uuid
from sqlalchemy import text


class WebSocketHandler:
    """WebSocket连接处理器"""
    
    def __init__(
        self,
        websocket: WebSocket,
        user_id: str = None,
        token: str = None,
        redis_cache = None,
        session_store = None,
    ):
        self.websocket = websocket
        self.user_id = user_id or "guest_user"  # 用户ID（默认游客）
        self.token = token  # 会话 token（单活校验）
        self.cancelled_requests: set = set()  # 已取消的请求ID
        self.stt_sessions: Dict[str, STTSession] = {}  # requestId -> STT会话对象
        self.event_loop: Optional[asyncio.AbstractEventLoop] = None  # 保存事件循环引用
        self.request_start_times: Dict[str, float] = {}  # 请求开始时间
        self.audio_end_times: Dict[str, float] = {}  # requestId -> 录音结束时间
        self.ai_responses: Dict[str, list] = {}  # requestId -> AI响应句子列表（用于日志记录）
        self.performance_metrics: Dict[str, PerformanceMetrics] = {}  # requestId -> 性能指标
        self.energy_stone_service = None
        
        # 初始化服务（延迟加载，避免循环导入）
        self.message_service = None
        self.context_service = None
        self.redis_cache = redis_cache  # 🔥 Redis缓存服务
        self.session_store = session_store  # 会话存储（单活 token）
        
        # 🔥 预加载任务
        self.preload_tasks: Dict[str, asyncio.Task] = {}
        
        # 🔥 异步持久化队列
        self.pending_persistence = []
        self.persistence_task: Optional[asyncio.Task] = None
        self.persistence_interval = 5.0  # 每5秒批量持久化
        
        # 🔥 异步总结队列
        self.pending_summaries = []
        self.summary_task: Optional[asyncio.Task] = None
        
        # 情景对话：首条语音携带的 scene/topic，在进入 LLM 前落库
        self._pending_scenario_log: Dict[str, Dict[str, str]] = {}
        
    async def _init_services(self):
        """初始化服务（延迟加载，避免重复初始化）"""
        start_time = time.time()
        
        # 🔥 如果服务已初始化，直接返回（避免重复初始化导致的阻塞）
        if self.message_service and self.context_service:
            logger.debug(f"[{self.user_id}] 🔍 服务已初始化，跳过重复初始化")
            return
        
        try:
            logger.info(f"[{self.user_id}] 🔧 开始初始化服务...")
            from database import db_manager
            from services import MessageService, ContextService
            
            # 使用延迟初始化的全局实例
            from services.message_service import message_service
            from services.context_service import context_service
            
            # 🔥 只在服务未初始化时才设置（避免重复调用set_db_manager）
            if not message_service.db or not context_service.db:
                logger.debug(f"[{self.user_id}] 🔧 设置数据库管理器...")
                await message_service.set_db_manager(db_manager)
                await context_service.set_db_manager(db_manager)
            if not energy_stone_service.db:
                logger.debug(f"[{self.user_id}] 🔧 设置能量石服务数据库管理器...")
                await energy_stone_service.set_db_manager(db_manager)
            if not scenario_topic_service.db:
                await scenario_topic_service.set_db_manager(db_manager)
            
            self.message_service = message_service
            self.context_service = context_service
            self.energy_stone_service = energy_stone_service
            
            elapsed = (time.time() - start_time) * 1000
            logger.info(f"[{self.user_id}] ✅ 服务初始化完成，耗时: {elapsed:.1f}ms")
        except Exception as e:
            elapsed = (time.time() - start_time) * 1000
            logger.error(f"[{self.user_id}] ❌ 服务初始化失败，耗时: {elapsed:.1f}ms, 错误: {e}")
            logger.error(traceback.format_exc())

    async def _close_with_session_invalid(self, reason: str = "session_invalidated"):
        """发送会话失效错误并关闭连接"""
        try:
            await self.send_error(
                request_id="session",
                code="SESSION_INVALIDATED",
                error_message=reason,
            )
        except Exception:
            pass
        try:
            await self.websocket.close(code=4401, reason=reason)
        except Exception:
            pass

    async def _ensure_valid_session(self) -> bool:
        """校验会话 token，确保单用户单活"""
        if not self.session_store:
            return True
        if not self.user_id or not self.token:
            await self._close_with_session_invalid("missing_session_token")
            return False
        if not await self.session_store.validate(self.user_id, self.token, refresh=True):
            await self._close_with_session_invalid("session_invalidated")
            return False
        await self.session_store.bind_connection(self.user_id, self.token, self.websocket)
        return True
    
    async def handle_connection(self):
        """处理WebSocket连接"""
        await self.websocket.accept()
        
        # 初始化服务
        await self._init_services()
        
        # 校验会话 token
        if not await self._ensure_valid_session():
            return
        
        # 🔥 恢复未过期的能量石（页面切换后重新显示）
        await self._restore_active_stones()
        
        # 保存事件循环引用，供回调线程使用
        self.event_loop = asyncio.get_running_loop()
        
        try:
            while True:
                # 接收消息
                data = await self.websocket.receive_text()
                message_data = json.loads(data)
                
                # 处理消息
                await self.handle_message(message_data)

        except WebSocketDisconnect as e:
            close_code = getattr(e, "code", None)
            reason = getattr(e, "reason", None)
            logger.warning(f"[{self.user_id}] ⚠️ WebSocket断开: code={close_code}, reason={reason}")

        except Exception as e:
            logger.error(f"[{self.user_id}] ❌ WebSocket异常: {e.__class__.__name__}: {e}")
            logger.error(traceback.format_exc())

        finally:
            close_code = getattr(self.websocket, "close_code", None)
            logger.info(f"[{self.user_id}] ℹ️ WebSocket连接结束，close_code={close_code}")
    
    async def handle_message(self, message_data: dict):
        """处理接收到的消息"""
        message_type = message_data.get("type")
        request_id = message_data.get("requestId")
        incoming_token = message_data.get("token")
        if incoming_token:
            self.token = incoming_token

        if not await self._ensure_valid_session():
            return
        
        # 检查是否为已取消的请求
        if request_id in self.cancelled_requests:
            return
        
        if message_type == "audio_chunk":
            await self.handle_audio_chunk(AudioChunkMessage(**message_data))
            
        elif message_type == "audio_end":
            await self.handle_audio_end(AudioEndMessage(**message_data))
            
        elif message_type == "cancel":
            await self.handle_cancel(CancelMessage(**message_data))
            
        elif message_type == "ping":
            await self.handle_ping(PingMessage(**message_data))
        
        elif message_type == "energy_stone_event":
            await self.handle_energy_stone_event(EnergyStoneEventMessage(**message_data))
        
        elif message_type == "start_greeting":
            # 由前端显式触发首句 greeting；支持 data.forceScene 测试用
            data = message_data.get("data") or {}
            force_scene = data.get("forceScene") if isinstance(data, dict) else None
            await self._send_greeting_message(force_scene=force_scene)
        
        elif message_type == "play_topic_first_line":
            # 情景对话：用户切换话题后请求播放该话题首句 TTS
            data = message_data.get("data") or {}
            if isinstance(data, dict) and data.get("topicKey"):
                await self._play_topic_first_line(
                    request_id=message_data.get("requestId") or f"topic_{uuid.uuid4().hex[:8]}",
                    topic_key=data.get("topicKey"),
                )
    
    async def handle_audio_chunk(self, message: AudioChunkMessage):
        """处理音频数据块 - 使用新的STT服务"""
        request_id = message.requestId
        
        # 检查是否取消
        if request_id in self.cancelled_requests:
            return
        
        # Base64解码音频数据
        audio_data = base64.b64decode(message.data)
        chunk_size = len(audio_data)
        
        # 第一个chunk：初始化请求并启动STT会话
        if request_id not in self.request_start_times:
            current_time = time.time()
            self.request_start_times[request_id] = current_time
            
            # 情景对话：首条语音携带的 scene/topic，进入 LLM 前落库
            if getattr(message, "scene", None) and getattr(message, "topic", None):
                self._pending_scenario_log[request_id] = {"scene": message.scene, "topic": message.topic}
            
            # 初始化性能指标
            self.performance_metrics[request_id] = PerformanceMetrics()
            self.performance_metrics[request_id].user_speech_start = current_time
            self.performance_metrics[request_id].stt_first_audio_received = current_time
        
            # 创建STT会话
            session = await self._create_stt_session(request_id)
            if not session or not session.is_connected:
                await self.send_error(request_id, "STT_ERROR", "语音识别服务启动失败")
                return
            
            self.stt_sessions[request_id] = session
            
            # 🔥 启动Redis预加载任务（与录音并行）
            if self.redis_cache:
                self.preload_tasks[request_id] = asyncio.create_task(
                    self._ensure_redis_context(request_id)
                )
        
        # 发送音频帧到STT服务
        session = self.stt_sessions.get(request_id)
        if session:
            success = await session.send_audio(audio_data)
            if not success:
                await self.send_error(request_id, "STT_ERROR", "发送音频数据失败")
            
            # 📝 更新STT音频统计（用于日志）
            if request_id in self.performance_metrics:
                self.performance_metrics[request_id].stt_audio_chunks += 1
                self.performance_metrics[request_id].stt_total_bytes += chunk_size
    
    async def handle_audio_end(self, message: AudioEndMessage):
        """处理音频结束 - 使用新的STT服务"""
        request_id = message.requestId
        
        # 情景对话：若首 chunk 未带 scene/topic，可在此补带（仅首条语音）
        if getattr(message, "scene", None) and getattr(message, "topic", None) and request_id not in self._pending_scenario_log:
            self._pending_scenario_log[request_id] = {"scene": message.scene, "topic": message.topic}
        
        # 记录录音结束时间
        audio_end_time = time.time()
        self.audio_end_times[request_id] = audio_end_time
        
        # 更新性能指标
        if request_id in self.performance_metrics:
            self.performance_metrics[request_id].user_speech_end = audio_end_time
            self.performance_metrics[request_id].stt_last_audio_received = audio_end_time
        
        speech_duration_ms = message.speechDuration or 0
        total_duration_ms = None
        level_update = None

        # 🔥 更新用户累积说话时长（如果有提供，使用超时避免阻塞）
        if speech_duration_ms > 0:
            duration_start = time.time()
            try:
                logger.info(f"[{request_id}] ⏱️ 开始更新用户说话时长: {speech_duration_ms}ms（超时: 1.0s）...")
                # 使用超时，避免长时间阻塞
                level_update = await asyncio.wait_for(
                    user_service.update_speech_duration(self.user_id, speech_duration_ms),
                    timeout=1.0  # 1秒超时
                )
                duration_elapsed = (time.time() - duration_start) * 1000
                logger.info(f"[{request_id}] ✅ 更新用户说话时长成功，耗时: {duration_elapsed:.1f}ms")
            except asyncio.TimeoutError:
                duration_elapsed = (time.time() - duration_start) * 1000
                logger.warning(f"[{request_id}] ⚠️ 更新用户说话时长超时（耗时: {duration_elapsed:.1f}ms）: user_id={self.user_id}, duration_ms={speech_duration_ms}")
            except Exception as e:
                duration_elapsed = (time.time() - duration_start) * 1000
                logger.warning(f"[{request_id}] ⚠️ 更新用户说话时长失败（耗时: {duration_elapsed:.1f}ms）: user_id={self.user_id}, duration_ms={speech_duration_ms}, error={e}")
        
        # 初始化AI响应收集器
        self.ai_responses[request_id] = []

        if level_update and isinstance(level_update, dict):
            total_duration_ms = level_update.get("total_duration_ms")
            user_level = level_update.get("user_level")
            if total_duration_ms is not None:
                await self._maybe_spawn_conversation_stones(total_duration_ms, speech_duration_ms, request_id)
            # 🔥 即使等级没有变化，也发送 user_level 消息，确保前端能同步 total_duration_ms
            # 获取当前用户等级（如果更新时等级没变，user_level 可能为 None，需要查询）
            if user_level is not None:
                await self.send_user_level_update(
                    request_id=request_id,
                    level=user_level,
                    total_duration_ms=total_duration_ms or 0,
                )
            elif total_duration_ms is not None:
                # 等级没变，但总时长更新了，需要查询当前等级并发送消息
                try:
                    user_info = await user_service.get_user(self.user_id)
                    if user_info and user_info.get("user_level") is not None:
                        await self.send_user_level_update(
                            request_id=request_id,
                            level=user_info.get("user_level"),
                            total_duration_ms=total_duration_ms,
                        )
                except Exception as e:
                    logger.warning(f"[{request_id}] ⚠️ 查询用户等级失败，无法发送 user_level 消息: {e}")
        
        # 检查是否取消
        if request_id in self.cancelled_requests:
            self.cancelled_requests.remove(request_id)
            await self.cleanup_request(request_id)
            return
        
        try:
            # 1. STT: 等待完整识别结果
            stt_start = time.time()
            
            session = self.stt_sessions.get(request_id)
            if not session:
                await self.send_error(request_id, "STT_ERROR", "语音识别会话不存在")
                request_logger.log_error(
                    user_id=self.user_id,
                    request_id=request_id,
                    error_code="STT_ERROR",
                    error_message="语音识别会话不存在",
                    stage="stt"
                )
                return
            
            # 🔥 手动提交音频，触发识别（禁用VAD时必需）
            try:
                commit_success = await session.commit_audio()
                if not commit_success:
                    logger.error(
                        f"[STT异常] request_id={request_id}, "
                        f"错误类型=提交识别请求失败, "
                        f"用户ID={self.user_id}, "
                        f"音频块数={message.totalChunks}, "
                        f"会话状态=is_connected={session.is_connected}"
                    )
                    await self.send_error(request_id, "STT_ERROR", "提交识别请求失败")
                    request_logger.log_error(
                        user_id=self.user_id,
                        request_id=request_id,
                        error_code="STT_ERROR",
                        error_message="提交识别请求失败",
                        stage="stt"
                    )
                    await self.cleanup_request(request_id)
                    return
            except Exception as commit_error:
                logger.error(
                    f"[STT异常] request_id={request_id}, "
                    f"错误类型=提交音频异常, "
                    f"异常类型={type(commit_error).__name__}, "
                    f"错误消息={str(commit_error)}, "
                    f"用户ID={self.user_id}, "
                    f"音频块数={message.totalChunks}, "
                    f"堆栈跟踪:\n{traceback.format_exc()}"
                )
                await self.send_error(request_id, "STT_ERROR", f"提交音频异常: {str(commit_error)}")
                request_logger.log_error(
                    user_id=self.user_id,
                    request_id=request_id,
                    error_code="STT_ERROR",
                    error_message=f"提交音频异常: {str(commit_error)}",
                    stage="stt"
                )
                await self.cleanup_request(request_id)
                return
            
            # 等待最终结果（根据音频长度动态调整超时）
            total_chunks = message.totalChunks
            if total_chunks > 80:
                timeout = 5.0
            elif total_chunks > 50:
                timeout = 4.0
            else:
                timeout = 3.0
            
            logger.info(f"[{request_id}] ⏳ 等待STT最终识别结果（超时: {timeout}s）...")
            text = await session.wait_for_final_result(timeout=timeout)
            
            # 获取识别结果
            if not text:
                # 检查语音检测状态，提供更详细的错误信息
                callback = session.callback
                speech_detected = callback.speech_detected if callback else False
                speech_start_count = callback.speech_start_count if callback else 0
                
                if not speech_detected and speech_start_count == 0:
                    error_msg = "未检测到有效语音，请检查麦克风权限或重新录音"
                else:
                    error_msg = f"语音识别超时，已检测到语音但未返回识别结果（speech_detected={speech_detected}, speech_start_count={speech_start_count}）"
                
                logger.error(
                    f"[{request_id}] ❌ STT识别失败: {error_msg}, "
                    f"音频块数={message.totalChunks}, "
                    f"录音时长={message.speechDuration}ms"
                )
                
                await self.send_error(request_id, "STT_ERROR", error_msg)
                request_logger.log_error(
                    user_id=self.user_id,
                    request_id=request_id,
                    error_code="STT_ERROR",
                    error_message=error_msg,
                    stage="stt"
                )
                await self.cleanup_request(request_id)
                return
            
            stt_duration = time.time() - stt_start
            stt_complete_time = time.time()
            
            # 记录STT完成时间（在获取到text之后）
            stt_complete_elapsed = (stt_complete_time - self.request_start_times.get(request_id, stt_complete_time)) * 1000
            logger.info(f"[{request_id}] ✅ STT识别完成，识别文本: '{text[:50]}{'...' if len(text) > 50 else ''}'，耗时: {stt_complete_elapsed:.1f}ms")
            
            # 更新性能指标
            if request_id in self.performance_metrics:
                self.performance_metrics[request_id].stt_completed = stt_complete_time
                self.performance_metrics[request_id].user_text = text
                # 获取STT首个文本生成时间
                first_text_time = session.get_first_text_time()
                if first_text_time:
                    self.performance_metrics[request_id].stt_first_text_generated = first_text_time
            
            # 发送最终识别结果（source=user，前端不用于字幕）
            await self.send_text_result(request_id, text, True, source="user")
            
            if request_id in self.cancelled_requests:
                return
            
            # 🔥 LLM提示词准备开始
            llm_prompt_start = time.time()
            if request_id in self.performance_metrics:
                self.performance_metrics[request_id].llm_prompt_prepare_start = llm_prompt_start
            
            # 🔥 从Redis获取上下文（零延迟）
            context = None
            current_round = 0
            next_round = 0
            
            if self.redis_cache:
                # 尝试从Redis获取上下文（使用超时，避免长时间阻塞）
                redis_start = time.time()
                try:
                    logger.info(f"[{request_id}] 🔍 开始从Redis获取上下文（超时: 2.0s）...")
                    context = await asyncio.wait_for(
                        self._get_context_from_redis(request_id),
                        timeout=2.0  # 2秒超时
                    )
                    redis_elapsed = (time.time() - redis_start) * 1000
                    logger.info(f"[{request_id}] ✅ Redis上下文获取完成，耗时: {redis_elapsed:.1f}ms")
                except asyncio.TimeoutError:
                    redis_elapsed = (time.time() - redis_start) * 1000
                    logger.warning(f"[{request_id}] ⚠️ 从Redis获取上下文超时（耗时: {redis_elapsed:.1f}ms），降级到MySQL")
                    context = None
                
                if context:
                    current_round = context.get('current_round', 0)
                    next_round = current_round + 1
                    logger.debug(f"[{request_id}] 📊 当前轮次: {current_round}，下一轮次: {next_round}")
                    
                    # 🔥 立即写入Redis（用户消息，使用超时）
                    write_start = time.time()
                    try:
                        logger.info(f"[{request_id}] 💾 开始写入Redis（用户消息，超时: 1.0s）...")
                        await asyncio.wait_for(
                            self.redis_cache.append_message(
                                user_id=self.user_id,
                                role='user',
                                content=text,
                                round_number=next_round
                            ),
                            timeout=1.0  # 1秒超时
                        )
                        append_elapsed = (time.time() - write_start) * 1000
                        logger.debug(f"[{request_id}] ✅ Redis写入用户消息完成，耗时: {append_elapsed:.1f}ms")
                        
                        increment_start = time.time()
                        await asyncio.wait_for(
                            self.redis_cache.increment_round(self.user_id),
                            timeout=1.0  # 1秒超时
                        )
                        increment_elapsed = (time.time() - increment_start) * 1000
                        write_total = (time.time() - write_start) * 1000
                        logger.info(f"[{request_id}] ✅ Redis写入完成（增量轮次耗时: {increment_elapsed:.1f}ms，总耗时: {write_total:.1f}ms）")
                    except asyncio.TimeoutError:
                        write_elapsed = (time.time() - write_start) * 1000
                        logger.warning(f"[{request_id}] ⚠️ Redis写入超时（耗时: {write_elapsed:.1f}ms），但继续处理")
                    except Exception as e:
                        write_elapsed = (time.time() - write_start) * 1000
                        logger.warning(f"[{request_id}] ⚠️ Redis写入失败（耗时: {write_elapsed:.1f}ms）: {e}")
                    
                    # 🔥 异步保存user消息到MySQL（不阻塞主流程）
                    await self._init_services()
                    if self.message_service:
                        asyncio.create_task(
                            self.message_service.save_message(
                                user_id=self.user_id,
                                role='user',
                                content=text
                            )
                        )
                    
                else:
                    # Redis未命中，降级到MySQL
                    mysql_start = time.time()
                    logger.info(f"[{request_id}] ⚠️ Redis未命中，降级到MySQL获取上下文...")
                    await self._init_services()
                    
                    if self.message_service and self.context_service:
                        try:
                            # 保存用户消息
                            save_start = time.time()
                            logger.info(f"[{request_id}] 💾 开始保存用户消息到MySQL（超时: 5.0s）...")
                            user_message = await asyncio.wait_for(
                                self.message_service.save_message(
                                    user_id=self.user_id,
                                    role='user',
                                    content=text
                                ),
                                timeout=5.0
                            )
                            save_elapsed = (time.time() - save_start) * 1000
                            next_round = user_message['round_number']
                            current_round = next_round - 1
                            logger.info(f"[{request_id}] ✅ 用户消息保存成功，轮次: {next_round}，耗时: {save_elapsed:.1f}ms")
                            
                            # 📝 记录用户消息ID
                            if request_id in self.performance_metrics:
                                self.performance_metrics[request_id].user_message_id = user_message.get('message_id')
                            
                            # 获取上下文
                            context_start = time.time()
                            logger.info(f"[{request_id}] 🔍 开始从MySQL获取上下文（超时: 3.0s）...")
                            context = await asyncio.wait_for(
                                self.context_service.get_context_for_round(
                                    user_id=self.user_id,
                                    current_round=current_round
                                ),
                                timeout=3.0
                            )
                            context_elapsed = (time.time() - context_start) * 1000
                            mysql_total_elapsed = (time.time() - mysql_start) * 1000
                            if context:
                                messages_count = len(context.get('messages', []))
                                logger.info(f"[{request_id}] ✅ 从MySQL获取上下文成功，消息数: {messages_count}，获取耗时: {context_elapsed:.1f}ms，总耗时: {mysql_total_elapsed:.1f}ms")
                            else:
                                logger.warning(f"[{request_id}] ⚠️ 从MySQL获取上下文为空，获取耗时: {context_elapsed:.1f}ms，总耗时: {mysql_total_elapsed:.1f}ms")
                            
                        except asyncio.TimeoutError as e:
                            elapsed = (time.time() - mysql_start) * 1000
                            logger.error(f"[{request_id}] ❌ MySQL操作超时，耗时: {elapsed:.1f}ms, 错误: {e}")
                        except Exception as e:
                            elapsed = (time.time() - mysql_start) * 1000
                            logger.error(f"[{request_id}] ❌ MySQL操作失败，耗时: {elapsed:.1f}ms, 错误: {e}")
                            logger.error(traceback.format_exc())
            else:
                # 没有Redis，直接使用MySQL
                await self._init_services()
                
                if self.message_service and self.context_service:
                    try:
                        user_message = await asyncio.wait_for(
                            self.message_service.save_message(
                                user_id=self.user_id,
                                role='user',
                                content=text
                            ),
                            timeout=5.0
                        )
                        next_round = user_message['round_number']
                        current_round = next_round - 1
                        
                        # 📝 记录用户消息ID
                        if request_id in self.performance_metrics:
                            self.performance_metrics[request_id].user_message_id = user_message.get('message_id')
                        
                        context = await asyncio.wait_for(
                            self.context_service.get_context_for_round(
                                user_id=self.user_id,
                                current_round=current_round
                            ),
                            timeout=3.0
                        )
                        
                    except asyncio.TimeoutError:
                        pass
                    except Exception as e:
                        pass
            
            # 🔥 LLM提示词准备结束
            llm_prompt_end = time.time()
            if request_id in self.performance_metrics:
                self.performance_metrics[request_id].llm_prompt_prepare_end = llm_prompt_end
            
            llm_start = time.time()
            # 🔥 LLM调用开始
            if request_id in self.performance_metrics:
                self.performance_metrics[request_id].llm_call_start = llm_start
            
            chunk_index = 0
            first_audio_time = None
            final_emotion = "neutral"
            sentence_count = 0
            llm_first_sentence_time = None
            llm_first_text_time = None  # 🔥 LLM首个文本输出时间
            emotion_sent = False  # 标记是否已发送情感
            
            # 情景对话：首条语音若携带 scene/topic，在此落库（仅记录一次）
            pending = self._pending_scenario_log.pop(request_id, None)
            if pending:
                await scenario_topic_service.record_conversation(
                    self.user_id, pending["scene"], pending["topic"]
                )
            
            # 构建Prompt（如果有上下文，使用上下文；否则使用原始文本）
            prompt_build_start = time.time()
            
            # 获取用户信息（用于运行时上下文）
            user_info = await user_service.get_user(self.user_id)
            is_new_user = False
            if user_info:
                # 判断是否为新用户（total_rounds == 0）
                is_new_user = (user_info.get("total_rounds", 0) == 0)
            
            # 获取system_prompt和runtime_context用于日志记录
            system_prompt = ai_service._build_system_prompt(user_info, is_new_user)
            runtime_context = ai_service._build_runtime_context(user_info)
            
            # 构建对话历史（用于上下文）- v0.3: 近20轮对话
            chat_history = []
            if context and context.get("recent_messages"):
                for msg in context.get("recent_messages", []):
                    role = "user" if msg.get("role") == "user" else "assistant"
                    chat_history.append({"role": role, "content": msg.get("content", "")})
            
            # 📝 构建完整的 messages 数组（用于保存 LLM 提示词记录和日志）
            llm_messages = [
                {"role": "system", "content": system_prompt},
                {"role": "system", "content": runtime_context},
            ]
            if chat_history:
                llm_messages.extend(chat_history)
            llm_messages.append({"role": "user", "content": text})  # ✅ v0.3: 使用原始用户输入
            
            # 构建完整提示词用于日志（文本格式）
            full_prompt_parts = [
                f"System (Base): {system_prompt}",
                f"System (Runtime Context): {runtime_context}",
            ]
            if chat_history:
                full_prompt_parts.append(f"Chat History ({len(chat_history)} messages)")
            full_prompt_parts.append(f"User: {text}")
            full_prompt = "\n\n".join(full_prompt_parts)
            
            prompt_build_elapsed = (time.time() - prompt_build_start) * 1000
            
            # 📝 更新LLM输入统计（用于日志）
            if request_id in self.performance_metrics:
                self.performance_metrics[request_id].llm_prompt_length = len(text)
                self.performance_metrics[request_id].llm_context_round = next_round
                self.performance_metrics[request_id].llm_full_prompt = full_prompt
            
            logger.info(f"[{request_id}] 🤖 开始调用LLM，用户输入长度: {len(text)}，构建耗时: {prompt_build_elapsed:.1f}ms，轮次: {next_round}")
            
            # 🔥 流式处理：LLM生成一句 → 立即TTS
            # ✅ v0.3: 传入原始用户输入，generate_response_stream 内部会构建完整的 messages 数组
            try:
                llm_first_chunk_time = None
                
                async for sentence, emotion in ai_service.generate_response_stream(
                    text,  # ✅ v0.3: 原始用户输入，generate_response_stream 内部会构建完整的 messages 数组
                    user_info=user_info,
                    is_new_user=is_new_user,
                    chat_history=chat_history if chat_history else None
                ):
                    if llm_first_chunk_time is None:
                        llm_first_chunk_time = time.time()
                        llm_first_chunk_elapsed = (llm_first_chunk_time - llm_start) * 1000
                        logger.info(f"[{request_id}] ✅ LLM首个响应到达，耗时: {llm_first_chunk_elapsed:.1f}ms")
                    if request_id in self.cancelled_requests:
                        return
                    
                    # 🎯 优先处理情感：第一时间发送给前端
                    if not emotion_sent and emotion:
                        logger.info(f"[{request_id}] 🎭 发送情感标签给前端: {emotion}, 句子内容: {sentence[:100] if sentence else '(空句子)'}")
                        await self.send_emotion(request_id, emotion)
                        emotion_sent = True
                        final_emotion = emotion
                    
                    # ⚡ 如果是空句子（仅传递情感），跳过 TTS 处理
                    if not sentence:
                        continue
                    
                    sentence_count += 1
                    
                    # ✅ v0.3: 保留情感标签在发送给前端的文本中
                    # 发送给前端的文本（保留情感标签）
                    text_for_frontend = sentence
                    
                    # 🧹 清理文本：移除情感标记、特殊字符和表情符号（仅用于TTS）
                    clean_sentence = EMOTION_TAG_PATTERN.sub('', sentence)
                    clean_sentence = clean_sentence.translate(str.maketrans('', '', '()[]')).strip()
                    clean_sentence = EMOJI_PATTERN.sub('', clean_sentence)  # 🔥 移除表情符号
                    
                    # 如果清理后为空，跳过TTS
                    if not clean_sentence:
                        # 但如果有情感标签，仍然发送给前端
                        if "[EMOTION:" in text_for_frontend:
                            await self.send_text_result(
                                request_id, text_for_frontend, is_final=False, source="assistant"
                            )
                        continue
                    
                    # 发送文本给前端（保留情感标签，source=assistant 用于字幕拼接）
                    await self.send_text_result(
                        request_id, text_for_frontend, is_final=False, source="assistant"
                    )
                    
                    # 收集AI响应（用于日志记录）- 使用清理后的文本
                    self.ai_responses[request_id].append(clean_sentence)
                    
                    # 记录第一个句子生成时间
                    if llm_first_sentence_time is None:
                        llm_first_sentence_time = time.time()
                    
                    # 🔥 记录LLM首个文本输出时间
                    if llm_first_text_time is None:
                        llm_first_text_time = time.time()
                        if request_id in self.performance_metrics:
                            self.performance_metrics[request_id].llm_first_text_output = llm_first_text_time
                    
                    tts_sentence_start = time.time()
                    # 🔥 记录TTS调用开始时间（第一个句子）
                    if request_id in self.performance_metrics and self.performance_metrics[request_id].tts_call_start is None:
                        self.performance_metrics[request_id].tts_call_start = tts_sentence_start
                    
                    # 立即TTS合成这个句子（使用清理后的文本 + 情感参数）
                    try:
                        async for audio_chunk in ai_service.text_to_speech_stream(clean_sentence, emotion):
                            if request_id in self.cancelled_requests:
                                return
                            
                            # Base64编码
                            audio_base64 = base64.b64encode(audio_chunk).decode()
                            
                            # 发送音频数据
                            await self.send_audio_response(
                                request_id,
                                audio_base64,
                                chunk_index,
                                False,
                            )
                            
                            # 记录第一个音频块发送时间
                            if chunk_index == 0:
                                first_audio_time = time.time()
                                time_to_first_audio = first_audio_time - audio_end_time
                                # 🔥 记录TTS首个音频输出时间
                                if request_id in self.performance_metrics:
                                    self.performance_metrics[request_id].tts_first_audio_output = first_audio_time
                            
                            # 📝 更新TTS音频统计（用于日志）
                            if request_id in self.performance_metrics:
                                self.performance_metrics[request_id].tts_audio_chunks += 1
                                self.performance_metrics[request_id].tts_total_bytes += len(audio_chunk)
                            
                            chunk_index += 1
                    except Exception as tts_error:
                        # TTS异常已在ai_service中详细记录，这里只记录上下文信息
                        logger.error(
                            f"[WebSocket异常] request_id={request_id}, "
                            f"错误类型=TTS合成异常, "
                            f"异常类型={type(tts_error).__name__}, "
                            f"错误消息={str(tts_error)}, "
                            f"句子内容={clean_sentence[:100]}... (前100字符), "
                            f"情感={emotion}, "
                            f"句子序号={sentence_count}, "
                            f"已发送chunk数={chunk_index}, "
                            f"堆栈跟踪:\n{traceback.format_exc()}"
                        )
                        # 重新抛出异常，让外层处理
                        raise
                    
                    # 更新当前情感（但不再发送，因为已在开头发送）
                    if emotion:
                        final_emotion = emotion
            except Exception as llm_error:
                # LLM异常已在ai_service中详细记录，这里只记录上下文信息
                logger.error(
                    f"[WebSocket异常] request_id={request_id}, "
                    f"错误类型=LLM生成异常, "
                    f"异常类型={type(llm_error).__name__}, "
                    f"错误消息={str(llm_error)}, "
                    f"用户输入={text[:100]}... (前100字符), "
                    f"LLM输入={llm_input[:100]}... (前100字符), "
                    f"上下文轮次={next_round}, "
                    f"已生成句子数={sentence_count}, "
                    f"堆栈跟踪:\n{traceback.format_exc()}"
                )
                # 重新抛出异常，让外层处理
                raise
            
            # 发送 TTS 结束标记：整轮 AI 回复（可能多句话/多段 TTS）全部发完后只发一次 isLast=True，前端据此判定「对话结束」并在全部播完后才切 idle
            await self.send_audio_response(request_id, "", chunk_index, True)
            
            # 记录最后一个音频块发送时间
            last_audio_time = time.time()
            time_to_last_audio = last_audio_time - audio_end_time
            
            # 🔥 更新性能指标：LLM和TTS完成时间
            if request_id in self.performance_metrics:
                self.performance_metrics[request_id].llm_all_text_completed = last_audio_time
                self.performance_metrics[request_id].llm_call_end = last_audio_time
                self.performance_metrics[request_id].tts_all_audio_completed = last_audio_time
                self.performance_metrics[request_id].tts_call_end = last_audio_time
                # 📝 更新LLM输出统计
                self.performance_metrics[request_id].llm_emotion = final_emotion
                self.performance_metrics[request_id].llm_sentences = sentence_count
                # 📝 更新TTS输入统计
                self.performance_metrics[request_id].tts_voice = settings.tts_voice
            
            # 🔥 保存AI响应
            ai_response_text = ' '.join(self.ai_responses.get(request_id, []))
            llm_total_elapsed = (time.time() - llm_start) * 1000
            logger.info(f"[{request_id}] ✅ LLM调用完成，总耗时: {llm_total_elapsed:.1f}ms，句子数: {sentence_count}，情感: {final_emotion}")
            
            if ai_response_text:
                # 🔥 立即写入Redis（不阻塞）
                if self.redis_cache:
                    redis_save_start = time.time()
                    try:
                        logger.debug(f"[{request_id}] 💾 开始保存AI响应到Redis...")
                        await self.redis_cache.append_message(
                            user_id=self.user_id,
                            role='assistant',
                            content=ai_response_text,
                            round_number=next_round
                        )
                        redis_save_elapsed = (time.time() - redis_save_start) * 1000
                        logger.debug(f"[{request_id}] ✅ AI响应保存到Redis完成，耗时: {redis_save_elapsed:.1f}ms")
                    except Exception as e:
                        redis_save_elapsed = (time.time() - redis_save_start) * 1000
                        logger.warning(f"[{request_id}] ⚠️ AI响应保存到Redis失败，耗时: {redis_save_elapsed:.1f}ms, 错误: {e}")
                
                # 🔥 立即保存AI消息到MySQL（获取message_id）
                if self.message_service:
                    mysql_save_start = time.time()
                    try:
                        logger.info(f"[{request_id}] 💾 开始保存AI消息到MySQL（超时: 5.0s）...")
                        assistant_message = await asyncio.wait_for(
                            self.message_service.save_message(
                                user_id=self.user_id,
                                role='assistant',
                                content=ai_response_text,
                                emotion=final_emotion
                            ),
                            timeout=5.0
                        )
                        mysql_save_elapsed = (time.time() - mysql_save_start) * 1000
                        logger.info(f"[{request_id}] ✅ AI消息保存到MySQL成功，耗时: {mysql_save_elapsed:.1f}ms")
                        # 📝 记录AI消息ID
                        if request_id in self.performance_metrics:
                            self.performance_metrics[request_id].assistant_message_id = assistant_message.get('message_id')
                        
                        # 🔥 保存LLM提示词记录（异步，不阻塞主流程）
                        db_message_id = assistant_message.get('id')
                        if db_message_id and self.message_service:
                            try:
                                # 获取LLM调用参数
                                llm_call_metadata = {}
                                if request_id in self.performance_metrics:
                                    metrics = self.performance_metrics[request_id]
                                    llm_call_metadata = {
                                        "llm_call_start": getattr(metrics, 'llm_call_start', None),
                                        "llm_first_chunk_time": getattr(metrics, 'llm_first_chunk_time', None),
                                        "llm_first_text_output": getattr(metrics, 'llm_first_text_output', None),
                                        "llm_prompt_length": getattr(metrics, 'llm_prompt_length', None),
                                    }
                                
                                # 异步保存LLM提示词记录（不阻塞主流程）
                                asyncio.create_task(
                                    self.message_service.save_llm_prompt(
                                        message_id=db_message_id,
                                        user_id=self.user_id,
                                        round_number=next_round,
                                        system_prompt=system_prompt,
                                        runtime_context=runtime_context,
                                        messages=llm_messages,
                                        model_name=settings.llm_model,
                                        temperature="0.7",
                                        max_tokens=500,
                                        stream=True,
                                        metadata=llm_call_metadata
                                    )
                                )
                                logger.debug(f"[{request_id}] 📝 LLM提示词记录保存任务已创建: message_id={db_message_id}")
                            except Exception as e:
                                logger.warning(f"[{request_id}] ⚠️ 创建LLM提示词记录保存任务失败: {e}")
                    except asyncio.TimeoutError:
                        mysql_save_elapsed = (time.time() - mysql_save_start) * 1000
                        logger.warning(f"[{request_id}] ⚠️ AI消息保存到MySQL超时，耗时: {mysql_save_elapsed:.1f}ms")
                    except Exception as e:
                        mysql_save_elapsed = (time.time() - mysql_save_start) * 1000
                        logger.warning(f"[{request_id}] ⚠️ AI消息保存到MySQL失败，耗时: {mysql_save_elapsed:.1f}ms, 错误: {e}")
                        # 保存失败不影响主流程，继续异步持久化
                
                # 🔥 加入异步持久化队列（批量写入MySQL，用于重试或降级场景）
                asyncio.create_task(
                    self._persist_conversation_async(
                        request_id=request_id,
                        user_text=text,
                        ai_text=ai_response_text,
                        round_number=next_round,
                        emotion=final_emotion
                    )
                )
                
                # 首次对话“水晶石”赠送已关闭：不再生成首句奖励（MissionEnergyStone）
                # 也不再更新 `users.crystal_stone_total`。

                # 注意：总结触发逻辑已移至 message_service.py
                # 在保存 assistant 消息时，如果 current_round % 15 == 0，会自动触发总结
            
            # 计算总耗时
            total_end = time.time()
            total_duration = total_end - self.request_start_times.get(request_id, stt_start)
            llm_tts_total = total_end - llm_start
            
            # 📝 记录详细性能日志到文件
            try:
                ai_response_log = ' '.join(self.ai_responses.get(request_id, []))
                
                # 更新最终的AI响应文本
                if request_id in self.performance_metrics:
                    self.performance_metrics[request_id].ai_response_text = ai_response_log
                    
                    # 记录详细性能数据
                    request_logger.log_request(
                        user_id=self.user_id,
                        request_id=request_id,
                        metrics=self.performance_metrics[request_id]
                    )
            except Exception as log_error:
                pass
            
        except Exception as e:
            # 详细记录异常信息
            session = self.stt_sessions.get(request_id)
            user_text_log = session.get_current_text() if session else ""
            
            # 判断异常来源（STT/LLM/TTS）
            error_type = "UNKNOWN_ERROR"
            error_stage = "unknown"
            error_str = str(e).lower()
            
            if "stt" in error_str or "语音识别" in error_str or "transcription" in error_str:
                error_type = "STT_ERROR"
                error_stage = "stt"
            elif "llm" in error_str or "chat" in error_str or "completion" in error_str or "generate" in error_str:
                error_type = "LLM_ERROR"
                error_stage = "llm"
            elif "tts" in error_str or "speech" in error_str or "audio" in error_str:
                error_type = "TTS_ERROR"
                error_stage = "tts"
            else:
                error_type = "NETWORK_ERROR"
                error_stage = "unknown"
            
            logger.error(
                f"[WebSocket异常] request_id={request_id}, "
                f"user_id={self.user_id}, "
                f"错误类型={error_type}, "
                f"异常类型={type(e).__name__}, "
                f"错误消息={str(e)}, "
                f"阶段={error_stage}, "
                f"用户输入={user_text_log[:100]}... (前100字符), "
                f"堆栈跟踪:\n{traceback.format_exc()}"
            )
            
            # 记录错误日志
            try:
                request_logger.log_error(
                    user_id=self.user_id,
                    request_id=request_id,
                    error_code=error_type,
                    error_message=str(e),
                    stage=error_stage,
                    user_text=user_text_log
                )
            except Exception as log_error:
                logger.error(
                    f"[日志记录异常] request_id={request_id}, "
                    f"错误类型=记录错误日志失败, "
                    f"异常类型={type(log_error).__name__}, "
                    f"错误消息={str(log_error)}"
                )
            
            await self.send_error(request_id, error_type, str(e))
        finally:
            # 清理资源
            await self.cleanup_request(request_id)
    
    async def _create_stt_session(self, request_id: str) -> Optional[STTSession]:
        """创建STT会话并设置回调"""
        async def on_text_update(text: str, is_final: bool):
            """STT文本更新回调（source=user，前端不用于字幕）"""
            await self.send_text_result(request_id, text, is_final, source="user")
        
        async def on_error(error_msg: str):
            """STT错误回调"""
            await self.send_error(request_id, "STT_ERROR", error_msg)
        
        # 创建会话
        session = await stt_service.create_session(
            request_id=request_id,
            on_text_update=on_text_update,
            on_error=on_error,
            event_loop=self.event_loop
        )
        
        return session if session.is_connected else None
    
    async def cleanup_request(self, request_id: str):
        """清理请求相关资源"""
        # 清理STT会话
        if request_id in self.stt_sessions:
            session = self.stt_sessions[request_id]
            await session.close()
            del self.stt_sessions[request_id]
        
        # 清理取消标记
        if request_id in self.cancelled_requests:
            self.cancelled_requests.remove(request_id)
        
        # 清理时间记录
        if request_id in self.request_start_times:
            del self.request_start_times[request_id]
        
        if request_id in self.audio_end_times:
            del self.audio_end_times[request_id]
        
        # 清理AI响应收集器
        if request_id in self.ai_responses:
            del self.ai_responses[request_id]
        
        # 清理性能指标
        if request_id in self.performance_metrics:
            del self.performance_metrics[request_id]
    
    async def handle_cancel(self, message: CancelMessage):
        """处理取消请求"""
        request_id = message.requestId
        
        self.cancelled_requests.add(request_id)
        
        # 清理资源（包括关闭STT会话）
        await self.cleanup_request(request_id)
    
    async def handle_ping(self, message: PingMessage):
        """处理Ping"""
        await self.send_pong(message.requestId)

    async def handle_energy_stone_event(self, message: EnergyStoneEventMessage):
        """处理前端能量石事件"""
        if not self.energy_stone_service:
            return

        metadata = {
            "client_ts": message.clientTs,
            "position": message.position,
            "request_id": message.requestId,
        }
        try:
            await self.energy_stone_service.mark_event(
                stone_id=message.stoneId,
                event=message.event,
                metadata={k: v for k, v in metadata.items() if v is not None},
            )
        except Exception as exc:
            logger.warning(
                "[EnergyStone] 记录事件失败: stone_id=%s event=%s error=%s",
                message.stoneId,
                message.event,
                exc,
            )
    
    async def _maybe_spawn_mission_stone(self, request_id: str):
        """首次对话奖励"""
        if not self.energy_stone_service or not self.user_id:
            return
        try:
            payload = await self.energy_stone_service.create_mission_stone_if_needed(
                user_id=self.user_id,
                request_id=request_id,
            )
            if payload:
                await self.send_energy_stone_spawn(payload)
        except Exception as exc:
            logger.warning(
                "[EnergyStone] 任务能量石生成失败: user_id=%s error=%s",
                self.user_id,
                exc,
            )
    
    async def _maybe_spawn_conversation_stones(
        self,
        total_duration_ms: int,
        increment_ms: int,
        request_id: str,
    ):
        """按累计说话时长发放"""
        if (
            not self.energy_stone_service
            or not self.user_id
            or total_duration_ms <= 0
            or increment_ms <= 0
        ):
            return
        try:
            stones = await self.energy_stone_service.create_conversation_stones_if_needed(
                user_id=self.user_id,
                request_id=request_id,
                total_duration_ms=total_duration_ms,
                duration_increment_ms=increment_ms,
            )
            for stone in stones:
                await self.send_energy_stone_spawn(stone)
        except Exception as exc:
            logger.warning(
                "[EnergyStone] 对话能量石生成失败: user_id=%s error=%s",
                self.user_id,
                exc,
            )

    async def _send_greeting_message(self, force_scene: Optional[str] = None):
        """
        发送首句（AI先说话）
        - 若 force_scene 有效则强制该场景；否则按权重随机场景（特殊场景需排除已聊完的）。
        - default：新用户固定首句 / 回归用户动态登录问候。
        - 特殊场景：未完成话题等概率选一，发 greeting_context 后发固定首句+TTS。
        """
        request_id = f"greeting_{uuid.uuid4().hex[:8]}"
        try:
            user_info = await user_service.get_user(self.user_id)
            if not user_info:
                logger.warning(f"[{self.user_id}] ⚠️ 无法获取用户信息，跳过首句发送")
                return
            
            user_name = (user_info.get("username") or "").strip() or "Neo"
            is_new_user = (user_info.get("total_rounds", 0) == 0)
            
            # 1) 决定本次场景
            chosen_scene = SCENE_DEFAULT
            scene_anchor: Optional[Dict[str, Any]] = None
            valid_force = force_scene in (SCENE_DEFAULT, SCENE_WINDOW, SCENE_READING, SCENE_EATING)
            if valid_force:
                if force_scene == SCENE_DEFAULT:
                    chosen_scene = force_scene
                    logger.info(f"[{self.user_id}] 🎯 首句强制场景: {chosen_scene}")
                else:
                    forced_anchor = await self._pick_scene_anchor_for_user(force_scene)
                    if forced_anchor:
                        chosen_scene = force_scene
                        scene_anchor = forced_anchor
                        logger.info(f"[{self.user_id}] 🎯 首句强制场景: {chosen_scene}, anchor_layout_item_id={scene_anchor.get('anchorLayoutItemId')}")
                    else:
                        chosen_scene = SCENE_DEFAULT
                        logger.info(f"[{self.user_id}] ℹ️ 强制场景 {force_scene} 无可用锚点，回退 default")
            else:
                await self._init_services()
                completed = await scenario_topic_service.get_completed_count_per_scenario(self.user_id)
                candidates = []
                for scene_key, weight in SCENE_WEIGHTS:
                    if scene_key == SCENE_DEFAULT:
                        candidates.append((scene_key, weight))
                    else:
                        if completed.get(scene_key, 0) < 3 and await self._scene_has_available_anchor(scene_key):
                            candidates.append((scene_key, weight))
                if not candidates:
                    candidates = [(SCENE_DEFAULT, 70)]
                total = sum(w for _, w in candidates)
                r = random.uniform(0, total)
                acc = 0
                for sk, w in candidates:
                    acc += w
                    if r < acc:
                        chosen_scene = sk
                        break
            
            # 2) 默认场景：沿用原逻辑
            if chosen_scene == SCENE_DEFAULT:
                if is_new_user:
                    lumi_name = (user_info.get("bot_name") or "").strip() or "Lumi"
                    greeting_text = ai_service.first_meeting_line(user_name, lumi_name)
                    logger.info(f"[{self.user_id}] 🎯 新用户首句: {greeting_text}")
                else:
                    await self._init_services()
                    chat_history = []
                    if self.context_service:
                        try:
                            current_round = await self.redis_cache.get_current_round(self.user_id) if self.redis_cache else 1
                            context = await self.context_service.get_context_for_round(
                                user_id=self.user_id, current_round=current_round
                            )
                            if context and context.get("recent_messages"):
                                for msg in context.get("recent_messages", []):
                                    role = "user" if msg.get("role") == "user" else "assistant"
                                    chat_history.append({"role": role, "content": msg.get("content", "")})
                        except Exception as e:
                            logger.warning(f"[{self.user_id}] ⚠️ 获取对话历史失败: {e}")
                    greeting_text = await ai_service.generate_login_greeting(
                        user_info=user_info, chat_history=chat_history if chat_history else None
                    )
                    logger.info(f"[{self.user_id}] 🎯 回归用户登录问候: {greeting_text}")
                
                emotion = "friendly" if is_new_user else "neutral"
                clean_text = greeting_text
                if not is_new_user and "[EMOTION:" in greeting_text:
                    emotion_start = greeting_text.find("[EMOTION:")
                    emotion_end = greeting_text.find("]", emotion_start)
                    if emotion_end > emotion_start:
                        emotion_str = greeting_text[emotion_start + 9:emotion_end].strip().lower()
                        valid_emotions = ["neutral", "happy", "sad", "excited", "angry", "surprised", "thoughtful", "encouraging", "confused", "worried", "satisfied", "friendly", "sorry"]
                        if emotion_str in valid_emotions:
                            emotion = emotion_str
                        clean_text = greeting_text[emotion_end + 1:].strip()
                clean_text = clean_text.translate(str.maketrans('', '', '()[]')).strip()
                clean_text = EMOJI_PATTERN.sub('', clean_text)
                if not clean_text or not clean_text.strip():
                    clean_text = "Welcome back! How are you today?"
                
                await self.send_emotion(request_id, emotion)
                await self.send_text_result(request_id, clean_text, is_final=True, source="assistant")
                chunk_index = 0
                async for audio_chunk in ai_service.text_to_speech_stream(clean_text, emotion):
                    audio_base64 = base64.b64encode(audio_chunk).decode('utf-8')
                    await self.send_audio_response(request_id=request_id, data=audio_base64, chunk_index=chunk_index, is_last=False)
                    chunk_index += 1
                await self.send_audio_response(request_id=request_id, data="", chunk_index=chunk_index, is_last=True)
                logger.info(f"[{self.user_id}] ✅ 首句发送完成: request_id={request_id}")
                return
            
            # 3) 特殊场景：选话题、发 greeting_context、再发首句+TTS
            if not scene_anchor:
                scene_anchor = await self._pick_scene_anchor_for_user(chosen_scene)
            if not scene_anchor:
                logger.warning(f"[{self.user_id}] ⚠️ 特殊场景 {chosen_scene} 无可用锚点，回退默认首句")
                await self._send_greeting_message(force_scene=SCENE_DEFAULT)
                return
            done_topics = await scenario_topic_service.get_done_topic_keys(self.user_id, chosen_scene)
            all_topics = SCENARIO_TOPICS.get(chosen_scene, [])
            available = [t for t in all_topics if t not in done_topics]
            if not available:
                logger.warning(f"[{self.user_id}] ⚠️ 特殊场景 {chosen_scene} 无可用话题，回退默认首句")
                await self._send_greeting_message(force_scene=SCENE_DEFAULT)
                return
            
            chosen_topic = random.choice(available)
            first_line = get_special_first_line(chosen_topic, user_name)
            topics_payload = build_topics_payload(available)
            
            await self._send_greeting_context(
                request_id=request_id,
                scene=chosen_scene,
                topic=chosen_topic,
                topics=topics_payload,
                first_line=first_line,
                scene_anchor=scene_anchor,
            )
            emotion = "friendly"
            await self.send_emotion(request_id, emotion)
            await self.send_text_result(request_id, first_line, is_final=True, source="assistant")
            chunk_index = 0
            async for audio_chunk in ai_service.text_to_speech_stream(first_line, emotion):
                audio_base64 = base64.b64encode(audio_chunk).decode('utf-8')
                await self.send_audio_response(request_id=request_id, data=audio_base64, chunk_index=chunk_index, is_last=False)
                chunk_index += 1
            await self.send_audio_response(request_id=request_id, data="", chunk_index=chunk_index, is_last=True)
            await grant_scene_gift_if_needed(self.user_id, chosen_scene)
            logger.info(f"[{self.user_id}] ✅ 特殊场景首句发送完成: scene={chosen_scene}, topic={chosen_topic}, request_id={request_id}")
        except Exception as e:
            logger.error(f"[{self.user_id}] ❌ 发送首句失败: {e}")
            logger.error(traceback.format_exc())
    
    async def _send_greeting_context(
        self,
        request_id: str,
        scene: str,
        topic: str,
        topics: list,
        first_line: str,
        scene_anchor: Optional[Dict[str, Any]] = None,
    ):
        """发送 greeting_context 消息（仅特殊场景）"""
        payload = {
            "type": "greeting_context",
            "requestId": request_id,
            "timestamp": self._get_timestamp(),
            "scene": scene,
            "topic": topic,
            "topics": topics,
            "firstLine": first_line,
        }
        if scene_anchor:
            payload.update(scene_anchor)
        await self.websocket.send_text(json.dumps(payload))

    async def _scene_has_available_anchor(self, scene: str) -> bool:
        if scene == SCENE_DEFAULT:
            return True
        anchor = await self._pick_scene_anchor_for_user(scene)
        return anchor is not None

    async def _pick_scene_anchor_for_user(self, scene: str) -> Optional[Dict[str, Any]]:
        """为特殊场景在当前布局中随机选取一个锚点（优先新槽位模型）。"""
        anchor_type: Optional[str] = None
        if scene in (SCENE_READING, SCENE_EATING):
            anchor_type = "seat"
        elif scene == SCENE_WINDOW:
            anchor_type = "window"
        if not anchor_type:
            return None

        try:
            from database import db_manager
            async with db_manager.get_session() as session:
                layout_user_type_res = await session.execute(
                    text(
                        """
                        SELECT DATA_TYPE
                        FROM information_schema.columns
                        WHERE table_schema = :db
                          AND table_name = 'chatbot_layouts'
                          AND column_name = 'user_id'
                        """
                    ),
                    {"db": settings.MYSQL_DATABASE},
                )
                layout_user_type = (layout_user_type_res.mappings().first() or {}).get("DATA_TYPE") or ""
                layout_user_id_where_expr = (
                    "(SELECT id FROM users WHERE user_id = :user_id LIMIT 1)"
                    if "bigint" in str(layout_user_type).lower()
                    else ":user_id"
                )

                # 1) 新模型：基于 user_chatbot_slot_placements + chatbot_backdrop_slots 选锚点
                backdrop_user_type_res = await session.execute(
                    text(
                        """
                        SELECT DATA_TYPE
                        FROM information_schema.columns
                        WHERE table_schema = :db
                          AND table_name = 'user_chatbot_backdrops'
                          AND column_name = 'user_id'
                        """
                    ),
                    {"db": settings.MYSQL_DATABASE},
                )
                backdrop_user_type = (backdrop_user_type_res.mappings().first() or {}).get("DATA_TYPE") or ""
                backdrop_user_id_where_expr = (
                    "(SELECT id FROM users WHERE user_id = :user_id LIMIT 1)"
                    if "bigint" in str(backdrop_user_type).lower()
                    else ":user_id"
                )

                backdrop_row_res = await session.execute(
                    text(
                        f"""
                        SELECT backdrop_item_id
                        FROM user_chatbot_backdrops
                        WHERE user_id = {backdrop_user_id_where_expr}
                        ORDER BY updated_at DESC, id DESC
                        LIMIT 1
                        """
                    ),
                    {"user_id": self.user_id},
                )
                backdrop_row = backdrop_row_res.mappings().first()
                backdrop_item_id = (
                    int(backdrop_row["backdrop_item_id"])
                    if backdrop_row and backdrop_row.get("backdrop_item_id") is not None
                    else None
                )

                candidates: list[dict[str, Any]] = []
                if backdrop_item_id is not None:
                    slot_candidates_res = await session.execute(
                        text(
                            f"""
                            SELECT
                                usp.id,
                                s.start_row,
                                s.start_col,
                                s.size_rows,
                                s.size_cols,
                                COALESCE(usp.z_index, s.z_index, si.default_z_index, 0) AS z_index
                            FROM user_chatbot_slot_placements usp
                            JOIN chatbot_backdrop_slots s
                              ON s.backdrop_item_id = usp.backdrop_item_id
                             AND s.slot_code = usp.slot_code
                             AND s.is_active = 1
                            LEFT JOIN shop_items si ON si.id = usp.item_id
                            WHERE usp.user_id = {layout_user_id_where_expr}
                              AND usp.backdrop_item_id = :backdrop_item_id
                              AND s.anchor_type = :anchor_type
                            ORDER BY usp.id ASC
                            """
                        ),
                        {
                            "user_id": self.user_id,
                            "backdrop_item_id": backdrop_item_id,
                            "anchor_type": anchor_type,
                        },
                    )
                    candidates = [dict(r) for r in slot_candidates_res.mappings().all()]

                # 2) 兼容旧模型：新模型未命中时回退 layout_items + scene_anchor_type
                if not candidates:
                    layout_row_res = await session.execute(
                        text(
                            f"""
                            SELECT id
                            FROM chatbot_layouts l
                            WHERE l.user_id = {layout_user_id_where_expr}
                            ORDER BY l.updated_at DESC, l.id DESC
                            LIMIT 1
                            """
                        ),
                        {"user_id": self.user_id},
                    )
                    layout_row = layout_row_res.mappings().first()
                    if not layout_row or layout_row.get("id") is None:
                        return None

                    items_res = await session.execute(
                        text(
                            """
                            SELECT
                                li.id,
                                li.start_row,
                                li.start_col,
                                li.size_rows,
                                li.size_cols,
                                li.z_index
                            FROM chatbot_layout_items li
                            JOIN shop_items si ON si.id = li.item_id
                            WHERE li.layout_id = :layout_id
                              AND si.scene_anchor_type = :anchor_type
                            ORDER BY li.id ASC
                            """
                        ),
                        {"layout_id": layout_row["id"], "anchor_type": anchor_type},
                    )
                    candidates = [dict(r) for r in items_res.mappings().all()]
                if not candidates:
                    return None
                picked = random.choice(candidates)
                anchor_z = int(picked.get("z_index") or 0)
                return {
                    "anchorLayoutItemId": int(picked["id"]),
                    "anchorPlacement": "window_left_bottom" if scene == SCENE_WINDOW else "seat_center_bottom",
                    "anchorZIndex": anchor_z,
                    "specialAvatarZIndex": max(anchor_z + 1, 3),
                }
        except Exception as e:
            logger.warning(f"[{self.user_id}] 获取场景锚点失败: scene={scene}, error={e}")
            return None
    
    async def _play_topic_first_line(self, request_id: str, topic_key: str):
        """情景对话：播放某话题的规范首句 TTS（用户切换话题后调用）"""
        try:
            user_info = await user_service.get_user(self.user_id)
            user_name = (user_info.get("username") or "").strip() or "Neo" if user_info else "Neo"
            first_line = get_special_first_line(topic_key, user_name)
            emotion = "friendly"
            await self.send_emotion(request_id, emotion)
            await self.send_text_result(request_id, first_line, is_final=True, source="assistant")
            chunk_index = 0
            async for audio_chunk in ai_service.text_to_speech_stream(first_line, emotion):
                audio_base64 = base64.b64encode(audio_chunk).decode("utf-8")
                await self.send_audio_response(request_id=request_id, data=audio_base64, chunk_index=chunk_index, is_last=False)
                chunk_index += 1
            await self.send_audio_response(request_id=request_id, data="", chunk_index=chunk_index, is_last=True)
        except Exception as e:
            logger.warning(f"[{self.user_id}] play_topic_first_line failed: topic_key={topic_key}, e={e}")
    
    async def send_user_level_update(self, request_id: str, level: int, total_duration_ms: int):
        """发送用户等级更新消息"""
        message = UserLevelMessage(
            type="user_level",
            requestId=request_id,
            timestamp=self._get_timestamp(),
            level=level,
            totalDurationMs=total_duration_ms,
        )
        await self.websocket.send_text(message.model_dump_json())
    
    async def send_text_result(
        self, request_id: str, text: str, is_final: bool, source: Optional[Literal["user", "assistant"]] = None
    ):
        """发送文本结果。source=user 表示用户STT(不参与字幕)，source=assistant 表示AI回复(参与字幕拼接)。"""
        message = TextResultMessage(
            type="text_result",
            requestId=request_id,
            timestamp=self._get_timestamp(),
            text=text,
            isFinal=is_final,
            source=source,
        )
        await self.websocket.send_text(message.model_dump_json())

    async def send_energy_stone_spawn(self, payload: Dict[str, Any]):
        """发送能量石生成消息"""
        message = EnergyStoneSpawnMessage(
            type="energy_stone_spawn",
            requestId=payload.get("stone_id"),
            timestamp=self._get_timestamp(),
            stoneId=payload.get("stone_id"),
            stoneType=payload.get("stone_type"),
            ttlSeconds=payload.get("ttl_seconds", settings.ENERGY_STONE_TTL_SECONDS),
            appearAt=payload.get("appear_at", self._get_timestamp()),
            metadata=payload.get("metadata"),
        )
        await self.websocket.send_text(message.model_dump_json())
    
    async def _restore_active_stones(self):
        """恢复未过期的能量石（页面切换后重新显示）"""
        if not self.energy_stone_service or not self.user_id:
            return
        
        try:
            active_stones = await self.energy_stone_service.get_active_stones(self.user_id)
            if active_stones:
                logger.info(
                    f"[{self.user_id}] 🔄 恢复 {len(active_stones)} 个未过期的能量石"
                )
                for stone_payload in active_stones:
                    await self.send_energy_stone_spawn(stone_payload)
        except Exception as exc:
            logger.warning(
                f"[{self.user_id}] ⚠️ 恢复能量石失败: {exc}"
            )
    
    async def send_emotion(self, request_id: str, emotion: str):
        """发送情感标签"""
        message = EmotionMessage(
            type="emotion",
            requestId=request_id,
            timestamp=self._get_timestamp(),
            emotion=emotion,
        )
        await self.websocket.send_text(message.model_dump_json())
    
    async def send_audio_response(
        self, request_id: str, data: str, chunk_index: int, is_last: bool,
        format: str = "pcm", sample_rate: int = 24000
    ):
        """发送音频响应"""
        message = AudioResponseMessage(
            type="audio_response",
            requestId=request_id,
            timestamp=self._get_timestamp(),
            data=data,
            sampleRate=sample_rate,
            format=format,
            chunkIndex=chunk_index,
            isLast=is_last,
        )
        await self.websocket.send_text(message.model_dump_json())
    
    async def send_error(self, request_id: str, code: str, error_message: str):
        """发送错误消息"""
        message = ErrorMessage(
            type="error",
            requestId=request_id,
            timestamp=self._get_timestamp(),
            code=code,
            message=error_message,
        )
        await self.websocket.send_text(message.model_dump_json())
    
    async def send_pong(self, request_id: str):
        """发送Pong"""
        message = PongMessage(
            type="pong",
            requestId=request_id,
            timestamp=self._get_timestamp(),
        )
        await self.websocket.send_text(message.model_dump_json())
    
    def _get_timestamp(self) -> int:
        """获取当前时间戳（毫秒）"""
        return int(time.time() * 1000)
    
    def _build_llm_input(self, user_input: str, context: dict) -> str:
        """
        构建包含上下文的LLM输入
        
        Args:
            user_input: 用户当前输入
            context: 上下文信息（包含summary和recent_messages）
        
        Returns:
            完整的LLM输入文本
        """
        
        parts = []
        
        # 1. 最新总结（如果有，已包含所有历史信息）
        summary = context.get('summary')
        if summary:
            parts.append("【历史对话总结】")
            parts.append(summary)
            parts.append("")
        
        # 2. 最近对话（如果有）
        recent_messages = context.get('recent_messages', [])
        if recent_messages:
            parts.append("【最近对话】")
            for msg in recent_messages:
                role_label = "User" if msg['role'] == 'user' else "Assistant"
                parts.append(f"{role_label}: {msg['content']}")
            parts.append("")
        
        # 3. 当前用户输入
        parts.append(f"User: {user_input}")
        
        return "\n".join(parts)
    
    async def _ensure_redis_context(self, request_id: str):
        """
        确保Redis中有最新上下文（预加载）
        
        策略：
        1. 检查Redis是否有缓存
        2. 如果没有，从MySQL加载并更新Redis
        3. 如果有，直接使用（零等待）
        """
        try:
            start = time.time()
            
            # 检查Redis缓存
            current_round = await self.redis_cache.get_current_round(self.user_id)
            cached_context = await self.redis_cache.get_context(self.user_id)
            
            if cached_context:
                # ✅ Redis已有缓存，无需操作
                return
            
            # ❌ Redis缓存未命中，从MySQL加载
            # 确保服务已初始化
            await self._init_services()
            
            if not self.context_service:
                return
            
            # 从MySQL获取上下文
            context = await self.context_service.get_context_for_round(
                user_id=self.user_id,
                current_round=current_round
            )
            
            # 更新Redis缓存
            await self.redis_cache.set_context(self.user_id, context)
            
            # 同时缓存消息列表
            for msg in context.get('recent_messages', []):
                await self.redis_cache.append_message(
                    user_id=self.user_id,
                    role=msg['role'],
                    content=msg['content'],
                    round_number=msg.get('round_number', 0)
                )
            
            # 同步轮次
            if current_round > 0:
                await self.redis_cache.set_current_round(self.user_id, current_round)
            
        except asyncio.CancelledError:
            # 🔥 任务被取消时，确保清理资源
            logger.debug(f"Redis上下文预加载任务被取消: request_id={request_id}")
            raise  # 重新抛出 CancelledError
        except Exception as e:
            import traceback
            traceback.print_exc()
    
    async def _get_context_from_redis(self, request_id: str) -> Optional[dict]:
        """
        从Redis获取上下文（零延迟）
        
        Returns:
            上下文字典 或 None（需降级到MySQL）
        """
        try:
            if not self.redis_cache:
                return None
            
            start = time.time()
            
            # 获取当前轮次
            current_round = await self.redis_cache.get_current_round(self.user_id)
            
            # 从Redis构建上下文（需要传入db_manager以获取所有总结）
            db_manager = None
            if self.context_service:
                db_manager = self.context_service.db
            
            context = await self.redis_cache.build_context_from_cache(
                user_id=self.user_id,
                current_round=current_round,
                db_manager=db_manager
            )
            
            return context
            
        except Exception as e:
            return None
    
    async def _persist_conversation_async(
        self,
        request_id: str,
        user_text: str,
        ai_text: str,
        round_number: int,
        emotion: str
    ):
        """
        异步持久化对话到MySQL（不阻塞主流程）
        
        策略：加入队列，批量写入
        """
        self.pending_persistence.append({
            'request_id': request_id,
            'user_id': self.user_id,
            'user_text': user_text,
            'ai_text': ai_text,
            'round_number': round_number,
            'emotion': emotion,
            'timestamp': time.time()
        })
        
        # 启动持久化任务
        if not self.persistence_task or self.persistence_task.done():
            self.persistence_task = asyncio.create_task(
                self._batch_persistence_loop()
            )
    
    async def _batch_persistence_loop(self):
        """批量持久化循环（后台任务）"""
        while True:
            try:
                await asyncio.sleep(self.persistence_interval)
                
                if not self.pending_persistence:
                    continue
                
                # 批量提取
                batch = self.pending_persistence.copy()
                self.pending_persistence.clear()
                
                start = time.time()
                
                # 确保服务已初始化
                await self._init_services()
                
                if not self.message_service:
                    # 重新加入队列
                    self.pending_persistence.extend(batch)
                    continue
                
                # 批量写入MySQL
                # ⚠️ 注意：用户消息和AI消息都已在主流程中立即保存
                # 这里不再重复保存，避免数据库中出现重复记录
                # 如果立即保存失败，应该通过错误日志和监控告警处理，而不是在这里重试
                for item in batch:
                    # 记录日志，但不保存（避免重复）
                    logger.debug(
                        f"跳过批量持久化（已立即保存）: "
                        f"user={item['user_id']}, round={item['round_number']}"
                    )
                
            except asyncio.CancelledError:
                # 🔥 任务被取消时，确保清理资源
                logger.info("批量持久化任务被取消，正在清理...")
                raise  # 重新抛出 CancelledError
            except Exception as e:
                import traceback
                traceback.print_exc()
    
    async def _generate_summary_async(
        self,
        request_id: str,
        user_id: str,
        start_round: int,
        end_round: int
    ):
        """
        异步生成对话总结（完全不阻塞）
        
        触发时机：
        - 每15轮触发一次（15, 30, 45...）
        """
        self.pending_summaries.append({
            'request_id': request_id,
            'user_id': user_id,
            'start_round': start_round,
            'end_round': end_round,
            'timestamp': time.time()
        })
        
        # 启动总结生成任务
        if not self.summary_task or self.summary_task.done():
            self.summary_task = asyncio.create_task(
                self._summary_generation_loop()
            )
    
    async def _summary_generation_loop(self):
        """总结生成循环（后台任务）"""
        while True:
            try:
                # 等待任务或间隔
                await asyncio.sleep(1.0)
                
                if not self.pending_summaries:
                    continue
                
                # 逐个处理总结任务
                task = self.pending_summaries.pop(0)
                
                start = time.time()
                
                try:
                    # 确保服务已初始化
                    await self._init_services()
                    
                    if not self.context_service or not self.redis_cache:
                        continue
                    
                    # 调用总结服务（耗时2-3秒）
                    # 注意：这里假设有summary_service，如果没有需要创建
                    from services.summary_service import SummaryService
                    summary_service = SummaryService(self.context_service.db)
                    
                    start_round = task['start_round']
                    end_round = task['end_round']
                    
                    # generate_summary 会自动从数据库获取消息并保存总结
                    # 该方法会自动获取之前的总结并进行累积总结
                    await summary_service.generate_summary(
                        user_id=task['user_id'],
                        start_round=start_round,
                        end_round=end_round
                    )
                    
                    # 从数据库获取生成的总结文本（使用 ORM）
                    from database import db_manager
                    from models_db import ConversationSummary
                    from sqlalchemy import select
                    
                    # 🔥 async with 会自动管理 session 生命周期，包括 CancelledError 情况
                    async with db_manager.get_session() as session:
                        result = await session.execute(
                            select(ConversationSummary)
                            .where(
                                ConversationSummary.user_id == task['user_id'],
                                ConversationSummary.start_round == start_round,
                                ConversationSummary.end_round == end_round
                            )
                            .order_by(ConversationSummary.created_at.desc())
                            .limit(1)
                        )
                        summary_obj = result.scalar_one_or_none()
                        summary = summary_obj.summary_text if summary_obj else ""
                    
                    # 🔥 保存到Redis（立即生效）
                    await self.redis_cache.set_summary(
                        user_id=task['user_id'],
                        end_round=end_round,
                        summary=summary
                    )
                    
                    # 🔥 更新Redis上下文缓存（包含新总结）
                    await self.redis_cache.invalidate_context(task['user_id'])
                    
                except Exception as e:
                    import traceback
                    traceback.print_exc()
                    
            except asyncio.CancelledError:
                # 🔥 任务被取消时，确保清理资源
                logger.info("总结生成任务被取消，正在清理...")
                raise  # 重新抛出 CancelledError
            except Exception as e:
                import traceback
                traceback.print_exc()