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, WebSocketState

logger = logging.getLogger(__name__)

from emotion_text_utils import (
    clean_text_for_display,
    strip_emotion_from_text,
)
# 表情符号正则表达式（用于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 ab_experiment_media_config import resolve_llm_model, resolve_tts_params
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 services.llm_task_service import llm_task_service
from scenario_config import (
    SCENE_DEFAULT,
    SCENE_WINDOW,
    SCENE_READING,
    SCENE_EATING,
    SCENE_WEIGHTS,
    SCENARIO_TOPICS,
    get_special_first_line,
    build_topics_payload,
    is_special_ab_experiment_user,
)
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:
                if not self._is_ws_connected():
                    break
                try:
                    data = await self.websocket.receive_text()
                except RuntimeError as exc:
                    # 同用户重连时 bind_connection 会主动关闭旧连接，Starlette 抛 RuntimeError 而非 WebSocketDisconnect
                    msg = str(exc).lower()
                    if "websocket is not connected" in msg or "need to call \"accept\" first" in msg:
                        logger.info(
                            f"[{self.user_id}] ℹ️ WebSocket 已关闭（可能被新连接替换），停止接收消息"
                        )
                        break
                    raise
                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:
            await self._cleanup_on_disconnect()
            close_code = getattr(self.websocket, "close_code", None)
            logger.info(f"[{self.user_id}] ℹ️ WebSocket连接结束，close_code={close_code}")

    def _is_ws_connected(self) -> bool:
        try:
            return self.websocket.client_state == WebSocketState.CONNECTED
        except Exception:
            return False

    async def _cleanup_on_disconnect(self) -> None:
        """连接关闭时释放 STT 会话等资源。"""
        if self.session_store:
            try:
                await self.session_store.unbind_connection(self.user_id, self.websocket)
            except Exception:
                pass
        for request_id in list(self.stt_sessions.keys()):
            try:
                await self.cleanup_request(request_id)
            except Exception as exc:
                logger.warning(
                    f"[{self.user_id}] ⚠️ 断开连接时清理 request={request_id} 失败: {exc}"
                )
        for task in self.preload_tasks.values():
            if not task.done():
                task.cancel()
        self.preload_tasks.clear()
    
    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

        # 说话时长写库改异步，不阻塞 STT → LLM → TTS 主链路（全用户生效）
        if speech_duration_ms > 0:
            asyncio.create_task(
                self._update_speech_duration_async(request_id, speech_duration_ms)
            )

        # 初始化AI响应收集器
        self.ai_responses[request_id] = []

        # 检查是否取消
        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)

            ab_task = None
            if is_special_ab_experiment_user(user_info):
                ab_task = await llm_task_service.get_latest_task_cached()
                if not ab_task:
                    logger.warning(f"[{self.user_id}] ⚠️ A/B 用户未找到任务表 prompt，将回退默认 base prompt")
            
            ab_task_prompt = ab_task.get("task_prompt") if ab_task else None
            ab_task_name = ab_task.get("task_name") if ab_task else None

            # 获取system_prompt和runtime_context用于日志记录
            system_prompt = ai_service._build_system_prompt(
                user_info,
                is_new_user,
                ab_task_prompt=ab_task_prompt,
            )
            runtime_context = ai_service._build_runtime_context(user_info, ab_task_name=ab_task_name)
            
            # 构建对话历史（用于上下文）- 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,
                    ab_task=ab_task,
                ):
                    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
                    
                    # 字幕展示文本：剥离 [EMOTION:xxx] 及 LLM 偶发的裸 emotion 词
                    display_text = clean_text_for_display(sentence)
                    
                    # 🧹 TTS 文本清理（括号/emoji 仅影响语音，不影响字幕）
                    clean_sentence = strip_emotion_from_text(sentence)
                    clean_sentence = clean_sentence.translate(str.maketrans('', '', '()[]')).strip()
                    clean_sentence = EMOJI_PATTERN.sub('', clean_sentence)
                    
                    # 如果清理后为空，跳过TTS（emotion 已通过 send_emotion 下发，无需空字幕）
                    if not clean_sentence:
                        continue
                    
                    # 发送文本给前端（source=assistant 用于字幕拼接）
                    if display_text:
                        await self.send_text_result(
                            request_id, display_text, 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, user_info=user_info
                        ):
                            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 = resolve_tts_params(user_info)["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=resolve_llm_model(user_info),
                                        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 _update_speech_duration_async(
        self, request_id: str, speech_duration_ms: int
    ) -> None:
        """后台更新累积说话时长，并同步等级/能量石（不阻塞对话主链路）。"""
        duration_start = time.time()
        try:
            logger.info(
                f"[{request_id}] ⏱️ 异步更新用户说话时长: {speech_duration_ms}ms..."
            )
            level_update = await asyncio.wait_for(
                user_service.update_speech_duration(self.user_id, speech_duration_ms),
                timeout=5.0,
            )
            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）: "
                f"user_id={self.user_id}, duration_ms={speech_duration_ms}"
            )
            return
        except Exception as e:
            duration_elapsed = (time.time() - duration_start) * 1000
            logger.warning(
                f"[{request_id}] ⚠️ 异步更新用户说话时长失败（耗时: {duration_elapsed:.1f}ms）: "
                f"user_id={self.user_id}, duration_ms={speech_duration_ms}, error={e}"
            )
            return

        if not level_update or not isinstance(level_update, dict):
            return

        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
            )
        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}"
                )
    
    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)
            ab_experiment_user = is_special_ab_experiment_user(user_info)
            
            # 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 ab_experiment_user:
                chosen_scene = SCENE_DEFAULT
                if valid_force and force_scene != SCENE_DEFAULT:
                    logger.info(
                        f"[{self.user_id}] A/B 实验用户忽略强制特殊场景 force_scene={force_scene}，使用 default"
                    )
                else:
                    logger.info(f"[{self.user_id}] A/B 实验用户，跳过特殊情景场景随机")
            elif 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:
                lumi_name = (user_info.get("bot_name") or "").strip() or "Lumi"
                if is_special_ab_experiment_user(user_info):
                    ab_task = await llm_task_service.get_latest_task_cached()
                    lumi_first_line = ab_task.get("lumi_first_line") if ab_task else None
                    greeting_text = ai_service.ab_group_greeting_line(
                        user_name,
                        lumi_name,
                        is_new_user,
                        lumi_first_line=lumi_first_line,
                    )
                    logger.info(f"[{self.user_id}] 🎯 A/B 实验用户首句: {greeting_text}")
                elif is_new_user:
                    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_special_ab_experiment_user(user_info)
                    and 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, user_info=user_info
                ):
                    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, user_info=user_info
            ):
                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, user_info=user_info
            ):
                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()