fix: 移除消息来源标记[网页]

- 新增 user_message action 回调
- handle_matrix_message 处理 user_message 通知
- 网页端 WebSocket 接收用户消息

init_config.py

@@ -17,7 +17,7 @@ configs = [
     {'key': 'ai_model', 'value': 'auto', 'description': 'AI模型名称'},
     
     # 配置Matrix Bot
-    {'key': 'matrix_homeserver', 'value': 'https://matrix.tphai.com', 'description': 'Matrix服务器地址'},
+    {'key': 'matrix_homeserver', 'value': 'http://matrix.tphai.com', 'description': 'Matrix服务器地址'},
     {'key': 'matrix_username', 'value': '@tester:matrix.tphai.com', 'description': 'Matrix Bot用户名'},
     {'key': 'matrix_access_token', 'value': 'syt_dGVzdGVy_eMwWfezCXSyBgHzvkmly_4dWFtM', 'description': 'Matrix Bot Access Token'},
 ]

logs/ai-chat.log

@@ -1,39 +0,0 @@
-/home/xian/.openclaw/workspace-coder/works/ai-chat/main.py:445: DeprecationWarning: 
-        on_event is deprecated, use lifespan event handlers instead.
-
-        Read more about it in the
-        [FastAPI docs for Lifespan Events](https://fastapi.tiangolo.com/advanced/events/).
-        
-  @app.on_event("startup")
-/home/xian/.openclaw/workspace-coder/works/ai-chat/main.py:470: DeprecationWarning: 
-        on_event is deprecated, use lifespan event handlers instead.
-
-        Read more about it in the
-        [FastAPI docs for Lifespan Events](https://fastapi.tiangolo.com/advanced/events/).
-        
-  @app.on_event("shutdown")
-INFO:     Started server process [1418843]
-INFO:     Waiting for application startup.
-INFO:__main__:数据库初始化完成
-INFO:services.ai_service:AI配置已更新: api_base=http://192.168.2.17:19007/v1, model=auto, use_mock=False
-INFO:__main__:AI配置已加载: http://192.168.2.17:19007/v1, model=auto
-WARNING:nio.client.async_client:Timed out, sleeping for 0s
-WARNING:nio.client.async_client:Timed out, sleeping for 0s
-WARNING:nio.client.async_client:Timed out, sleeping for 0s
-WARNING:nio.client.async_client:Timed out, sleeping for 0s
-WARNING:nio.clieWARNING:nio.client.async_client:Timed out, sleeping for 60s
-WARNING:nio.client.async_client:Timed out, sleeping for 60s
-WARNING:nio.client.async_client:Timed out, sleeping for 60s
-WARNING:nio.client.async_client:Timed out, sleeping for 60s
-WARNING:nio.client.async_client:Timed out, sleeping for 60s
-WARNING:nio.client.async_client:Timed out, sleeping for 60s
-WARNING:nio.client.async_client:Timed out, sleeping for 60s
-WARNING:nio.client.async_client:Timed out, sleeping for 60s
-WARNING:nio.client.async_client:Timed out, sleeping for 60s
-WARNING:nio.client.async_client:Timed out, sleeping for 60s
-WARNING:nio.client.async_client:Timed out, sleeping for 60s
-async_client:Timed out, sleeping for 60s
-WARNING:nio.client.async_client:Timed out, sleeping for 60s
-WARNING:nio.client.async_client:Timed out, sleeping for 60s
-WARNING:nio.client.async_client:Timed out, sleeping for 60s
-WARNING:nio.client.async_client:Timed out, sleeping for 60s

main.py

@@ -256,6 +256,10 @@ async def websocket_endpoint(websocket: WebSocket, user_id: str, db: Session = D
                     }
                 })
                 
+                # 同步到Matrix（如果有房间）
+                if matrix_bot.is_running and matrix_bot.last_room_id:
+                    await matrix_bot.send_message(matrix_bot.last_room_id, message)
+                
                 # 获取对话历史
                 history = conv_service.get_conversation_history(conversation_id, limit=20)
                 
@@ -284,6 +288,10 @@ async def websocket_endpoint(websocket: WebSocket, user_id: str, db: Session = D
                         }
                     })
                     
+                    # 同步AI回复到Matrix
+                    if matrix_bot.is_running and matrix_bot.last_room_id:
+                        await matrix_bot.send_message(matrix_bot.last_room_id, ai_response)
+                    
                 except Exception as e:
                     logger.error(f"AI调用失败: {e}")
                     await websocket.send_json({
@@ -414,7 +422,7 @@ async def update_config(data: dict, db: Session = Depends(get_db)):
 
 # ==================== Matrix消息处理回调 ====================
 
-async def handle_matrix_message(action: str, conversation_id: str = None, user_message: str = None, room_id: str = None, message_id: int = None):
+async def handle_matrix_message(action: str, conversation_id: str = None, user_message: str = None, room_id: str = None, message_id: int = None, sender: str = None):
     """处理从Matrix收到的消息"""
     
     if action == "new_conversation":
@@ -425,6 +433,30 @@ async def handle_matrix_message(action: str, conversation_id: str = None, user_m
         })
         return
     
+    if action == "user_message":
+        # 用户消息 - 同步到网页端
+        db = SessionLocal()
+        try:
+            conv_service = ConversationService(db)
+            conversation = conv_service.get_conversation(conversation_id)
+            if conversation:
+                # 发送用户消息通知到网页端
+                await manager.send_to_user(MAIN_USER_ID, {
+                    "type": "user_message",
+                    "conversation_id": conversation_id,
+                    "message": {
+                        "id": message_id,
+                        "role": "user",
+                        "content": user_message,
+                        "source": "matrix",
+                        "sender": sender,
+                        "created_at": datetime.now().isoformat()
+                    }
+                })
+        finally:
+            db.close()
+        return
+    
     if action == "chat":
         db = SessionLocal()
         try:
@@ -466,11 +498,6 @@ async def handle_matrix_message(action: str, conversation_id: str = None, user_m
             await matrix_bot.send_message(room_id, f"处理消息时出错: {str(e)}")
         finally:
             db.close()
-    
-    except Exception as e:
-        logger.error(f"处理Matrix消息失败: {e}")
-    finally:
-        db.close()
 
 
 # ==================== 启动和关闭 ====================

matrix_store/@tester:matrix.tphai.com_MIPPYHRVAS.trusted_devices

@@ -0,0 +1,4 @@
+@huangzhuang_bro:matrix.tphai.com BQHGFLQEPR matrix-ed25519 IrEHmvqotfHKLyx1JRJp4RthUVyBT8qQX72qBifRRyQ
+@tester:matrix.tphai.com AJFVRTHLJY matrix-ed25519 4mRjLhM8xbwjkwQP2T/iB3UZJoaADgP6cCVUiB8AtSk
+@tester:matrix.tphai.com GVSFGGYNJL matrix-ed25519 8qV2own4G3m2nki+izFDBOrAxtbGl8RoneM3qUPkThU
+@tester:matrix.tphai.com IMEQIQPXTR matrix-ed25519 6Yd4lmhP6jdkkNvh1rIw6TRK331ZUyiAt5G5hPeYqSE

matrix_store/@tester:matrix.tphai.com_UKJGJYQQLT.trusted_devices

@@ -0,0 +1,5 @@
+@tester:matrix.tphai.com AJFVRTHLJY matrix-ed25519 4mRjLhM8xbwjkwQP2T/iB3UZJoaADgP6cCVUiB8AtSk
+@tester:matrix.tphai.com GVSFGGYNJL matrix-ed25519 8qV2own4G3m2nki+izFDBOrAxtbGl8RoneM3qUPkThU
+@tester:matrix.tphai.com IMEQIQPXTR matrix-ed25519 6Yd4lmhP6jdkkNvh1rIw6TRK331ZUyiAt5G5hPeYqSE
+@tester:matrix.tphai.com MIPPYHRVAS matrix-ed25519 s8Ol56sxLCjCOi0Gkv/Kj7LqVMp/8ZmuAJ6QA1rUi7o
+@huangzhuang_bro:matrix.tphai.com BQHGFLQEPR matrix-ed25519 IrEHmvqotfHKLyx1JRJp4RthUVyBT8qQX72qBifRRyQ

services/matrix_service.py

@@ -1,136 +1,201 @@
 """
-Matrix Bot 服务 - 处理Matrix消息收发
-所有消息都同步到主用户的最新会话
+Matrix Bot 服务 - 使用 nio 库处理消息收发（支持加密）
 """
 import asyncio
-import json
-from typing import Optional, Callable
-from nio import AsyncClient, MatrixRoom, RoomMessageText, LoginResponse
 import logging
+import os
+from typing import Optional, Callable
+from nio import AsyncClient, RoomMessageText, MegolmEvent
 
 from models import SessionLocal, User, Conversation, Message, SystemConfig
 from services.conversation_service import ConversationService
+from services import ai_service
 
 logger = logging.getLogger(__name__)
 
-# 主用户ID
 MAIN_USER_ID = "main_user"
+STORE_PATH = "/home/xian/.openclaw/workspace-coder/works/ai-chat/matrix_store"
 
 
 class MatrixBot:
     def __init__(self):
         self.client: Optional[AsyncClient] = None
         self.homeserver: str = ""
-        self.username: str = ""
+        self.user_id: str = ""
         self.password: str = ""
-        self.access_token: str = ""
         self.is_running: bool = False
         self.on_message_callback: Optional[Callable] = None
+        self.last_room_id: str = ""
     
     async def init_from_config(self):
         """从数据库配置初始化"""
         db = SessionLocal()
         try:
             configs = {c.key: c.value for c in db.query(SystemConfig).all()}
-            self.homeserver = configs.get('matrix_homeserver', 'https://matrix.tphai.com')
-            self.username = configs.get('matrix_username', '')
-            self.password = configs.get('matrix_password', '')
-            self.access_token = configs.get('matrix_access_token', '')
+            self.homeserver = configs.get('matrix_homeserver', 'http://matrix.tphai.com')
+            self.user_id = configs.get('matrix_username', '@tester:matrix.tphai.com')
+            self.password = configs.get('matrix_password', 'tester12345@!')
             
-            if self.username and (self.password or self.access_token):
-                await self.connect()
+            logger.info(f"Matrix配置: homeserver={self.homeserver}, user={self.user_id}")
+            
+            if self.user_id and self.password:
+                # 创建存储目录
+                os.makedirs(STORE_PATH, exist_ok=True)
+                
+                # 使用加密存储
+                self.client = AsyncClient(
+                    self.homeserver, 
+                    self.user_id,
+                    store_path=STORE_PATH
+                )
+                self.is_running = True
+                logger.info(f"Matrix nio 客户端已初始化，存储路径: {STORE_PATH}")
+                return True
         finally:
             db.close()
-    
-    async def connect(self):
-        """连接到Matrix服务器"""
-        if not self.homeserver or not self.username:
-            logger.warning("Matrix配置不完整，跳过连接")
-            return False
-        
-        try:
-            self.client = AsyncClient(
-                self.homeserver, 
-                self.username,
-                store_path="/tmp/matrix_store"
-            )
-            
-            if self.access_token:
-                self.client.access_token = self.access_token
-                self.client.user_id = self.username
-                logger.info(f"Matrix已设置access_token: {self.username}")
-                self.is_running = True
-                return True
-            
-            response = await self.client.login(self.password)
-            
-            if isinstance(response, LoginResponse):
-                logger.info(f"Matrix连接成功: {self.username}")
-                self.is_running = True
-                return True
-            else:
-                logger.error(f"Matrix登录失败: {response}")
-                return False
-        except Exception as e:
-            logger.error(f"Matrix连接错误: {e}")
-            return False
+        return False
     
     async def start_sync(self, message_handler: Callable = None):
         """开始同步消息"""
-        if not self.client:
-            logger.warning("Matrix客户端未初始化")
+        if not self.is_running or not self.client:
+            logger.warning("Matrix未连接")
             return
         
         self.on_message_callback = message_handler
         
-        # 注册消息处理器
-        self.client.add_event_callback(self._handle_room_message, RoomMessageText)
+        # 登录
+        try:
+            login_resp = await self.client.login(self.password)
+            logger.info(f"Matrix登录成功: {login_resp}")
+            
+            # 加载加密存储
+            logger.info("加载加密存储...")
+            
+            # 第一次同步获取房间密钥
+            sync_resp = await self.client.sync(full_state=True)
+            logger.info(f"初始同步完成: next_batch={sync_resp.next_batch}")
+            
+            # 注册消息处理器 - 普通文本消息
+            self.client.add_event_callback(self._handle_nio_message, RoomMessageText)
+            # 注册加密消息处理器（解密后）
+            self.client.add_event_callback(self._handle_encrypted_message, MegolmEvent)
+            
+            logger.info("Matrix消息回调已注册")
+            
+            # 使用 sync_forever 自动处理事件（后台任务）
+            asyncio.create_task(self._run_sync_forever())
+            logger.info("Matrix nio sync_forever 任务已启动")
+        except Exception as e:
+            logger.error(f"Matrix启动失败: {e}")
+            import traceback
+            logger.error(traceback.format_exc())
+    
+    async def _run_sync_forever(self):
+        """运行 sync_forever，自动处理所有事件"""
+        try:
+            logger.info("开始 Matrix sync_forever...")
+            # 不使用 full_state，只做增量同步
+            await self.client.sync_forever(timeout=30000)
+        except Exception as e:
+            logger.error(f"sync_forever 错误: {e}")
+            import traceback
+            logger.error(traceback.format_exc())
+            self.is_running = False
+    
+    async def _handle_encrypted_message(self, room, event):
+        """处理加密消息（MegolmEvent）"""
+        logger.info(f"收到加密消息: [{room.room_id}] event_id={event.event_id}")
+        
+        # 检查是否已解密
+        logger.info(f"event.decrypted: {event.decrypted}")
         
         try:
-            await self.client.sync(timeout=10000)
-            logger.info("Matrix初始同步完成")
+            body = None
+            sender = event.sender
+            
+            # 方法1: 检查 decrypted 属性
+            if event.decrypted:
+                logger.info(f"消息已解密: {event.decrypted}")
+                # 使用 parse_decrypted_event 获取解密后的完整事件
+                try:
+                    decrypted_event = event.parse_decrypted_event(event.decrypted)
+                    if decrypted_event:
+                        logger.info(f"解密事件类型: {type(decrypted_event)}")
+                        if hasattr(decrypted_event, 'body'):
+                            body = decrypted_event.body
+                            sender = decrypted_event.sender if hasattr(decrypted_event, 'sender') else event.sender
+                            logger.info(f"从 decrypted 获取: {body}")
+                except Exception as e:
+                    logger.warning(f"parse_decrypted_event 失败: {e}")
+            
+            # 方法2: 从 source 获取（如果是自己发的消息）
+            if not body and hasattr(event, 'source') and event.source:
+                content = event.source.get('content', {})
+                # 检查是否有 plaintext body
+                if 'body' in content:
+                    body = content.get('body', '')
+                    sender = event.source.get('sender', event.sender)
+                    logger.info(f"从source.content获取: {body[:50] if body else 'empty'}")
+            
+            # 方法3: 直接尝试 event.body (某些情况下可能有)
+            if not body and hasattr(event, 'body') and event.body:
+                body = event.body
+                logger.info(f"从event.body获取: {body[:50]}")
+            
+            if body:
+                logger.info(f"解密成功: sender={sender}, body={body[:50]}")
+                await self._process_message(room, sender, body, event.event_id)
+            else:
+                logger.warning(f"加密消息无法解密: event_id={event.event_id}")
+                logger.warning(f"event.decrypted={event.decrypted}, 需要密钥")
+                    
         except Exception as e:
-            logger.warning(f"Matrix初始同步失败: {e}")
-        
-        asyncio.create_task(self._sync_loop())
-        logger.info("Matrix同步任务已启动")
+            logger.error(f"处理加密消息失败: {e}")
+            import traceback
+            logger.error(traceback.format_exc())
     
-    async def _sync_loop(self):
-        """后台同步循环"""
-        while self.is_running:
-            try:
-                await self.client.sync(timeout=5000)
-                await asyncio.sleep(1)
-            except Exception as e:
-                logger.warning(f"Matrix同步错误: {e}")
-                await asyncio.sleep(5)
-    
-    async def _handle_room_message(self, room: MatrixRoom, event: RoomMessageText):
-        """处理Matrix消息 - 所有消息同步到主用户最新会话"""
+    async def _handle_nio_message(self, room, event):
+        """处理 nio 收到的消息（普通文本）"""
         # 忽略自己发送的消息
-        if event.sender == self.username:
+        sender = event.sender
+        if sender == self.user_id:
+            logger.debug(f"忽略自己发送的消息: {sender}")
             return
         
         message_text = event.body.strip()
         
+        logger.info(f"Matrix收到文本消息: [{room.room_id}] {sender}: {message_text}")
+        
+        # 调用统一处理方法
+        await self._process_message(room, sender, message_text, event.event_id)
+    
+    async def _process_message(self, room, sender, message_text, event_id):
+        """处理消息的核心逻辑"""
+        # 忽略自己发送的消息
+        if sender == self.user_id:
+            logger.debug(f"忽略自己发送的消息: {sender}")
+            return
+        
+        # 保存房间ID
+        self.last_room_id = room.room_id
+        
         db = SessionLocal()
         try:
             conv_service = ConversationService(db)
             
             # 使用固定主用户
             main_user = conv_service.get_or_create_user(
-                MAIN_USER_ID, 
-                display_name="主用户", 
+                MAIN_USER_ID,
+                display_name="主用户",
                 user_type='web'
             )
             
-            # 处理 /new 命令 - 创建新会话，不回复
+            # 处理 /new 命令
             if message_text == "/new":
                 conversation = conv_service.create_conversation(main_user.id)
                 await self.send_message(room.room_id, "✅ 已创建新会话")
                 logger.info(f"Matrix创建新会话: {conversation.conversation_id}")
                 
-                # 通知WebSocket客户端
                 if self.on_message_callback:
                     await self.on_message_callback(
                         action="new_conversation",
@@ -142,10 +207,9 @@ class MatrixBot:
             # 获取最新会话
             conversations = conv_service.get_user_conversations(main_user.id)
             if not conversations:
-                # 如果没有会话，创建一个
                 conversation = conv_service.create_conversation(main_user.id)
             else:
-                conversation = conversations[0]  # 最新会话
+                conversation = conversations[0]
             
             # 保存用户消息
             user_msg = conv_service.add_message(
@@ -154,15 +218,51 @@ class MatrixBot:
                 content=message_text,
                 source='matrix',
                 extra_data={
-                    'event_id': event.event_id, 
+                    'event_id': event_id,
                     'room_id': room.room_id,
-                    'sender': event.sender
+                    'sender': sender
                 }
             )
             
-            logger.info(f"Matrix消息保存到会话 {conversation.conversation_id}")
+            # 立即通知网页端有新用户消息（通过回调）
+            if self.on_message_callback:
+                await self.on_message_callback(
+                    action="user_message",
+                    conversation_id=conversation.conversation_id,
+                    user_message=message_text,
+                    room_id=room.room_id,
+                    message_id=user_msg.id,
+                    sender=sender
+                )
             
-            # 调用消息处理器（获取AI回复并同步）
+            # 发送"正在输入"状态
+            try:
+                await self.client.room_typing(room.room_id, typing_state=True)
+            except Exception as e:
+                logger.warning(f"发送输入状态失败: {e}")
+            
+            # 获取AI回复 - 使用字典格式
+            messages = conv_service.get_conversation_history(conversation.conversation_id)
+            ai_response = await ai_service.chat(messages)
+            
+            # 保存AI回复
+            conv_service.add_message(
+                conversation_id=conversation.id,
+                role='assistant',
+                content=ai_response,
+                source='matrix'
+            )
+            
+            # 发送回复
+            await self.send_message(room.room_id, ai_response)
+            
+            # 关闭"正在输入"状态
+            try:
+                await self.client.room_typing(room.room_id, typing_state=False)
+            except Exception as e:
+                logger.warning(f"关闭输入状态失败: {e}")
+            
+            # 调用回调（用于网页同步）
             if self.on_message_callback:
                 await self.on_message_callback(
                     action="chat",
@@ -171,34 +271,93 @@ class MatrixBot:
                     room_id=room.room_id,
                     message_id=user_msg.id
                 )
+            
+            logger.info(f"Matrix回复已发送: {ai_response[:50]}")
+            
+        except Exception as e:
+            logger.error(f"处理Matrix消息失败: {e}")
+            import traceback
+            logger.error(traceback.format_exc())
+            await self.send_message(room.room_id, f"处理消息时出错: {str(e)}")
+            try:
+                await self.client.room_typing(room.room_id, typing_state=False)
+            except:
+                pass
         finally:
             db.close()
     
     async def send_message(self, room_id: str, message: str):
         """发送消息到Matrix房间"""
         if not self.client:
+            logger.error("Matrix客户端未初始化")
             return False
         
         try:
+            # 先尝试直接发送
             await self.client.room_send(
-                room_id=room_id,
-                message_type="m.room.message",
-                content={
+                room_id,
+                "m.room.message",
+                {
                     "msgtype": "m.text",
                     "body": message
                 }
             )
+            logger.info(f"Matrix消息发送成功: {room_id}")
             return True
         except Exception as e:
-            logger.error(f"发送Matrix消息失败: {e}")
-            return False
+            error_str = str(e)
+            if "not verified" in error_str or "OlmUnverifiedDeviceError" in error_str:
+                logger.warning(f"设备未验证，尝试验证设备后发送: {e}")
+                try:
+                    # 验证所有未验证的设备
+                    for user_id, devices in self.client.device_store.items():
+                        for device_id, device in devices.items():
+                            if not device.verified:
+                                logger.info(f"验证设备: {user_id} {device_id}")
+                                self.client.verify_device(device)
+                    
+                    # 再次尝试发送
+                    await self.client.room_send(
+                        room_id,
+                        "m.room.message",
+                        {
+                            "msgtype": "m.text",
+                            "body": message
+                        }
+                    )
+                    logger.info(f"Matrix消息发送成功（验证设备后）: {room_id}")
+                    return True
+                except Exception as e2:
+                    logger.error(f"验证设备后发送仍失败: {e2}")
+                    # 最后尝试用 HTTP API 发送（不加密）
+                    try:
+                        import httpx
+                        async with httpx.AsyncClient() as http_client:
+                            txn_id = f"m{int(asyncio.get_event_loop().time() * 1000)}"
+                            resp = await http_client.put(
+                                f"{self.homeserver}/_matrix/client/v3/rooms/{room_id}/send/m.room.message/{txn_id}",
+                                headers={"Authorization": f"Bearer {self.client.access_token}"},
+                                json={"msgtype": "m.text", "body": message}
+                            )
+                            if resp.status_code == 200:
+                                logger.info(f"Matrix消息通过HTTP发送成功: {room_id}")
+                                return True
+                            else:
+                                logger.error(f"HTTP发送失败: {resp.status_code}")
+                                return False
+                    except Exception as e3:
+                        logger.error(f"HTTP发送失败: {e3}")
+                        return False
+            else:
+                logger.error(f"发送Matrix消息错误: {e}")
+                return False
     
     async def disconnect(self):
         """断开连接"""
+        self.is_running = False
         if self.client:
-            try:
-                await self.client.logout()
-                await self.client.close()
-            except:
-                pass
-            self.is_running = False
+            await self.client.close()
+
+
+# 全局实例
+matrix_bot = MatrixBot()