feat: 修改数据库结构

assets/API.md

@@ -2,8 +2,14 @@
 
 ## 认证 `/api/auth`
 
+| 方法 | 路径 | 说明 |
+|------|------|------|
+| POST | `/api/auth/register` | 用户注册 |
+| POST | `/api/auth/login` | 用户登录 |
+| POST | `/api/auth/logout` | 用户登出 |
+| GET | `/api/auth/me` | 获取当前用户信息 |
+
 ### POST /api/auth/register
-用户注册
 
 **请求体：**
 ```json
@@ -14,20 +20,7 @@
 }
 ```
 
-**响应：**
-```json
-{
-  "success": true,
-  "message": "注册成功",
-  "data": {
-    "id": 1,
-    "username": "string"
-  }
-}
-```
-
 ### POST /api/auth/login
-用户登录
 
 **请求体：**
 ```json
@@ -37,88 +30,20 @@
 }
 ```
 
-**响应：**
-```json
-{
-  "success": true,
-  "message": "登录成功",
-  "data": {
-    "access_token": "eyJ...",
-    "token_type": "bearer",
-    "user": {
-      "id": 1,
-      "username": "string",
-      "role": "user"
-    }
-  }
-}
-```
-
-### POST /api/auth/logout
-用户登出
-
-**请求头：** `Authorization: Bearer <token>`
-
-**响应：**
-```json
-{
-  "success": true,
-  "message": "登出成功"
-}
-```
-
-### GET /api/auth/me
-获取当前用户信息
-
-**请求头：** `Authorization: Bearer <token>`
-
-**响应：**
-```json
-{
-  "success": true,
-  "data": {
-    "id": 1,
-    "username": "string",
-    "email": "user@example.com",
-    "role": "user",
-    "is_active": true
-  }
-}
-```
-
 ---
 
 ## 会话 `/api/conversations`
 
-### GET /api/conversations/
+| 方法 | 路径 | 说明 |
-获取会话列表
+|------|------|------|
+| GET | `/api/conversations/` | 获取会话列表 |
+| POST | `/api/conversations/` | 创建会话 |
+| GET | `/api/conversations/{id}` | 获取会话详情 |
+| PUT | `/api/conversations/{id}` | 更新会话 |
+| DELETE | `/api/conversations/{id}` | 删除会话 |
 
-**查询参数：**
+### 创建会话请求体
-- `project_id` (可选): 项目ID
-- `page` (可选): 页码，默认1
-- `page_size` (可选): 每页数量，默认20
 
-**请求头：** `Authorization: Bearer <token>`
-
-**响应：**
-```json
-{
-  "success": true,
-  "data": {
-    "items": [...],
-    "total": 100,
-    "page": 1,
-    "page_size": 20
-  }
-}
-```
-
-### POST /api/conversations/
-创建会话
-
-**请求头：** `Authorization: Bearer <token>`
-
-**请求体：**
 ```json
 {
   "project_id": "string (可选)",
@@ -131,56 +56,19 @@
 }
 ```
 
-**响应：**
-```json
-{
-  "success": true,
-  "message": "会话创建成功",
-  "data": {
-    "id": "conv_xxx",
-    "user_id": 1,
-    "title": "新会话",
-    "model": "glm-5",
-    ...
-  }
-}
-```
-
-### GET /api/conversations/{id}
-获取会话详情
-
-**路径参数：**
-- `id`: 会话ID
-
-**请求头：** `Authorization: Bearer <token>`
-
-### PUT /api/conversations/{id}
-更新会话
-
-### DELETE /api/conversations/{id}
-删除会话
-
 ---
 
 ## 消息 `/api/messages`
 
-### GET /api/messages/{conversation_id}
+| 方法 | 路径 | 说明 |
-获取消息列表
+|------|------|------|
+| GET | `/api/messages/{conversation_id}` | 获取消息列表 |
+| POST | `/api/messages/` | 发送消息（非流式） |
+| POST | `/api/messages/stream` | 发送消息（流式 SSE） |
+| DELETE | `/api/messages/{id}` | 删除消息 |
 
-**路径参数：**
+### 发送消息请求体
-- `conversation_id`: 会话ID
 
-**查询参数：**
-- `limit` (可选): 返回数量，默认100
-
-**请求头：** `Authorization: Bearer <token>`
-
-### POST /api/messages/
-发送消息（非流式）
-
-**请求头：** `Authorization: Bearer <token>`
-
-**请求体：**
 ```json
 {
   "conversation_id": "conv_xxx",
@@ -189,45 +77,122 @@
 }
 ```
 
-**响应：**
+### SSE 事件类型
+
+| 事件 | 说明 |
+|------|------|
+| `text` | 文本增量 |
+| `tool_call` | 工具调用 |
+| `tool_result` | 工具结果 |
+| `done` | 完成 |
+| `error` | 错误 |
+
+---
+
+## 聊天室 `/api/chat-rooms`
+
+| 方法 | 路径 | 说明 |
+|------|------|------|
+| GET | `/api/chat-rooms/` | 获取聊天室列表 |
+| POST | `/api/chat-rooms/` | 创建聊天室 |
+| GET | `/api/chat-rooms/{room_id}` | 获取聊天室详情 |
+| PUT | `/api/chat-rooms/{room_id}` | 更新聊天室 |
+| DELETE | `/api/chat-rooms/{room_id}` | 删除聊天室 |
+| GET | `/api/chat-rooms/{room_id}/agents` | 获取聊天室中的 Agent 列表 |
+| POST | `/api/chat-rooms/{room_id}/agents` | 添加 Agent 到聊天室 |
+| DELETE | `/api/chat-rooms/{room_id}/agents/{agent_id}` | 从聊天室移除 Agent |
+| GET | `/api/chat-rooms/{room_id}/messages` | 获取聊天室消息历史 |
+| POST | `/api/chat-rooms/{room_id}/messages` | 发送消息到聊天室（SSE 流式响应） |
+
+### 创建聊天室请求体
+
 ```json
 {
-  "success": true,
+  "name": "聊天室名称",
-  "data": {
+  "description": "描述（可选）",
-    "user_message": {...},
+  "agent_ids": ["agent-1", "agent-2"]
-    "assistant_message": {...}
-  }
 }
 ```
 
-### POST /api/messages/stream
+### 添加 Agent 请求体
-发送消息（流式响应）
 
-使用 Server-Sent Events (SSE) 返回流式响应。
+```json
+{
+  "agent_id": "agent-1"
+}
+```
 
-**事件类型：**
+---
-- `text`: 文本增量
-- `tool_call`: 工具调用
-- `tool_result`: 工具结果
-- `done`: 完成
-- `error`: 错误
 
-### DELETE /api/messages/{id}
+## WebSocket `/ws/chat-room/{room_id}`
-删除消息
+
+### 连接参数
+
+| 参数 | 类型 | 说明 | 示例 |
+|------|------|------|------|
+| `participant_type` | string | `user` 或 `agent` | `user` |
+| `participant_id` | string | 用户 ID 或 Agent ID | `1` |
+| `participant_name` | string | 显示名称 | `John` |
+
+### 连接示例
+
+**用户连接：**
+```
+ws://host/ws/chat-room/room-123?participant_type=user&participant_id=1&participant_name=John
+```
+
+**Agent 连接：**
+```
+ws://host/ws/chat-room/room-123?participant_type=agent&participant_id=agent-1&participant_name=Assistant
+```
+
+### 发送消息
+
+```json
+{
+  "action": "send_message",
+  "content": "Hello everyone!"
+}
+```
+
+### 接收事件
+
+| 事件 | 说明 | 数据示例 |
+|------|------|---------|
+| `connected` | 连接成功 | `{"room_id": "xxx", "type": "user"}` |
+| `history` | 历史消息 | `{"messages": [...]}` |
+| `agents` | Agent 列表 | `{"agents": [...]}` |
+| `message` | 新消息 | 消息对象 |
+| `typing` | 打字状态 | `{"agent_id": "xxx", "is_typing": true}` |
+| `process_step` | 处理步骤 | 步骤对象 |
+| `done` | 完成 | `{"message_id": "xxx", "token_count": 100}` |
+| `error` | 错误 | `{"content": "error message"}` |
+| `system` | 系统消息 | `{"content": "John joined", "type": "user_join"}` |
+
+---
+
+## Agent `/api/agents`
+
+| 方法 | 路径 | 说明 |
+|------|------|------|
+| GET | `/api/agents/` | 获取 Agent 列表 |
+| POST | `/api/agents/` | 创建 Agent |
+| GET | `/api/agents/{id}` | 获取 Agent 详情 |
+| PUT | `/api/agents/{id}` | 更新 Agent |
+| DELETE | `/api/agents/{id}` | 删除 Agent |
 
 ---
 
 ## 工具 `/api/tools`
 
-### GET /api/tools/
+| 方法 | 路径 | 说明 |
-获取可用工具列表
+|------|------|------|
+| GET | `/api/tools/` | 获取可用工具列表 |
+| GET | `/api/tools/{name}` | 获取工具详情 |
+| POST | `/api/tools/{name}/execute` | 手动执行工具 |
 
-**查询参数：**
+### 工具列表响应
-- `category` (可选): 工具分类
 
-**请求头：** `Authorization: Bearer <token>`
-
-**响应：**
 ```json
 {
   "success": true,
@@ -236,24 +201,78 @@
     "categorized": {
       "crawler": [...],
       "code": [...],
-      "data": [...],
+      "data": [...]
-      "weather": [...]
     },
     "total": 11
   }
 }
 ```
 
-### GET /api/tools/{name}
+### 执行工具请求体
-获取工具详情
 
-### POST /api/tools/{name}/execute
-手动执行工具
-
-**请求体：**
 ```json
 {
   "arg1": "value1",
   "arg2": "value2"
 }
 ```
+
+---
+
+## LLM 提供商 `/api/providers`
+
+| 方法 | 路径 | 说明 |
+|------|------|------|
+| GET | `/api/providers/` | 获取提供商列表 |
+| POST | `/api/providers/` | 创建提供商 |
+| GET | `/api/providers/{id}` | 获取提供商详情 |
+| PUT | `/api/providers/{id}` | 更新提供商 |
+| DELETE | `/api/providers/{id}` | 删除提供商 |
+| POST | `/api/providers/{id}/test` | 测试提供商连接 |
+
+---
+
+## 健康检查
+
+| 方法 | 路径 | 说明 |
+|------|------|------|
+| GET | `/api/health` | 健康检查 |
+| GET | `/` | 服务信息 |
+
+---
+
+## 通用响应格式
+
+### 成功响应
+
+```json
+{
+  "success": true,
+  "message": "操作成功",
+  "data": {...}
+}
+```
+
+### 错误响应
+
+```json
+{
+  "success": false,
+  "error": "错误信息",
+  "code": 404
+}
+```
+
+### 分页响应
+
+```json
+{
+  "success": true,
+  "data": {
+    "items": [...],
+    "total": 100,
+    "page": 1,
+    "page_size": 20
+  }
+}
+```

assets/ARCHITECTURE.md

@@ -17,35 +17,57 @@
 
 ```
 luxx/
-├── __init__.py           # FastAPI 应用工厂
+├── __init__.py              # FastAPI 应用工厂
-├── config.py             # 配置管理（YAML）
+├── config.py                # 配置管理（YAML）
-├── database.py           # 数据库连接
+├── core/
-├── models.py             # ORM 模型
+│   ├── __init__.py
-├── routes/               # API 路由层
+│   ├── config.py           # 配置管理
-│   ├── __init__.py      # 路由聚合
+│   └── database.py         # 数据库连接
-│   ├── auth.py          # 认证 (登录/注册)
+├── models/                  # ORM 模型
-│   ├── conversations.py # 会话管理 (CRUD)
+│   ├── __init__.py
-│   ├── messages.py      # 消息处理 (流式/同步)
+│   ├── user.py             # User, LLMProvider, Project
-│   ├── providers.py     # LLM 提供商管理
+│   ├── chat.py            # Conversation, Message
-│   └── tools.py         # 工具管理
+│   ├── room.py             # ChatRoom, Agent, RoomParticipant, ChatRoomMessage
-├── services/            # 服务层
+│   └── participant.py      # Participant (统一参与者抽象)
-│   ├── chat.py         # 聊天服务 (Agentic Loop)
+├── services/               # 服务层
-│   └── llm_client.py   # LLM 客户端
+│   ├── __init__.py
-├── tools/              # 工具系统
+│   ├── chat.py            # 聊天服务 (Agentic Loop)
-│   ├── core.py         # 核心类 (ToolRegistry, ToolDefinition, ToolResult)
+│   ├── room.py            # 聊天室服务 (多 Agent 编排)
-│   ├── factory.py      # @tool 装饰器
+│   ├── participant.py     # 统一参与者服务
-│   ├── executor.py     # 工具执行器 (缓存/并行)
+│   ├── room_ws.py         # WebSocket 处理
-│   ├── services.py     # 工具服务层
+│   ├── llm_service.py     # LLM 服务
-│   └── builtin/        # 内置工具
+│   ├── message_service.py # 消息服务
-│       ├── __init__.py # 工具注册入口
+│   ├── stream_service.py  # 流式响应服务
-│       ├── code.py     # 代码执行 (python_execute, python_eval)
+│   ├── task.py            # 任务服务
-│       ├── crawler.py  # 网页爬虫 (web_search, web_fetch, batch_fetch)
+│   └── agent.py           # Agent 管理
-│       └── data.py     # 数据处理 (process_data)
+├── agents/                 # Agent 系统
-└── utils/              # 工具函数
+│   ├── __init__.py
-    └── helpers.py     # 密码哈希、ID生成、响应封装
+│   ├── base.py            # BaseAgent 抽象类
+│   └── registry.py         # Agent 注册表
+├── api/                    # API 路由层
+│   ├── __init__.py
+│   ├── auth.py            # 认证 (登录/注册)
+│   ├── conversations.py   # 会话管理 (CRUD)
+│   ├── messages.py        # 消息处理
+│   ├── providers.py       # LLM 提供商管理
+│   ├── tools.py           # 工具管理
+│   ├── agents.py          # Agent 管理
+│   └── rooms.py           # 聊天室管理
+├── tools/                 # 工具系统
+│   ├── __init__.py
+│   ├── core.py            # 核心类 (ToolRegistry, ToolDefinition, ToolResult)
+│   ├── factory.py         # @tool 装饰器
+│   ├── executor.py        # 工具执行器
+│   └── builtin/           # 内置工具
+│       ├── __init__.py
+│       ├── code.py         # 代码执行
+│       ├── crawler.py      # 网页爬虫
+│       └── data.py         # 数据处理
+└── utils/                 # 工具函数
+    └── helpers.py          # 密码哈希、ID生成、响应封装
 
-run.py                  # 应用入口文件
+run.py                      # 应用入口文件
-config.yaml             # 配置文件
+config.yaml                 # 配置文件
 ```
 
 ## 核心组件
@@ -55,144 +77,70 @@ FastAPI 应用入口，使用 lifespan 管理生命周期：
 - 启动：初始化数据库、注册内置工具、创建默认管理员用户
 - 关闭：清理资源
 
-```python
+### 2. 配置管理 (`core/config.py`)
-# 默认管理员账号
-username: admin
-password: admin123
-```
-
-### 2. 配置管理 (`config.py`)
 使用 YAML 文件管理配置：
 - 配置文件：`config.yaml`
 - 环境变量替换：`${VAR_NAME}`
 - 单例模式全局访问
-- 默认值支持
 
-```yaml
+### 3. 数据库 (`core/database.py`)
-# config.yaml 示例
-app:
-  secret_key: ${APP_SECRET_KEY}
-  debug: true
-
-database:
-  type: sqlite
-  url: sqlite:///./chat.db
-
-llm:
-  provider: deepseek
-  api_key: ${DEEPSEEK_API_KEY}
-  api_url: https://api.deepseek.com/v1
-```
-
-### 3. 数据库 (`database.py`)
 - SQLAlchemy 同步支持
 - SQLite 默认数据库
 - 依赖注入获取会话
 
-### 4. ORM 模型 (`models.py`)
+### 4. ORM 模型
+
+#### 统一参与者架构
 
 ```mermaid
-erDiagram
+classDiagram
-    USER {
+    class Participant {
-        int id PK
+        +str participant_id
-        string username UK
+        +str name
-        string email UK
+        +ParticipantType participant_type
-        string password_hash
+        +from_user(User) Participant
-        string role
+        +from_agent(...) Participant
-        boolean is_active
+        +from_participant_id(str) Participant
-        datetime created_at
     }
-    
+
-    PROJECT {
+    class ParticipantType {
-        string id PK
+        <<enumeration>>
-        int user_id FK
+        USER = "user"
-        string name
+        AGENT = "agent"
-        text description
-        datetime created_at
-        datetime updated_at
     }
-    
+
-    CONVERSATION {
+    class RoomParticipant {
-        string id PK
+        +str room_id
-        int user_id FK
+        +str agent_id
-        int provider_id FK "optional"
+        +int user_id
-        string project_id FK "optional"
+        +str role
-        string title
+        +participant_id property
-        string model
+        +participant_type property
-        text system_prompt
-        float temperature
-        int max_tokens
-        boolean thinking_enabled
-        datetime created_at
-        datetime updated_at
     }
-    
+
-    MESSAGE {
+    class ChatRoomMessage {
-        string id PK
+        +str participant_id
-        string conversation_id FK
+        +sender_type property
-        string role
+        +sender_id property
-        longtext content "JSON 格式"
-        int token_count
-        text usage "JSON 格式"
-        datetime created_at
     }
-    
+
-    LLM_PROVIDER {
+    Participant --> ParticipantType
-        int id PK
+    RoomParticipant --> Participant
-        int user_id FK
+    ChatRoomMessage --> Participant
-        string name
-        string provider_type
-        string base_url
-        string api_key
-        string default_model
-        int max_tokens
-        boolean is_default
-        boolean enabled
-        datetime created_at
-        datetime updated_at
-    }
-    
-    USER ||--o{ PROJECT : "has"
-    USER ||--o{ CONVERSATION : "has"
-    USER ||--o{ LLM_PROVIDER : "configures"
-    PROJECT ||--o{ CONVERSATION : "contains"
-    LLM_PROVIDER ||--o{ CONVERSATION : "uses"
-    CONVERSATION ||--o{ MESSAGE : "has"
 ```
 
-### Message Content JSON 结构
+### 5. 聊天室系统
 
-`content` 字段统一使用 JSON 格式存储：
+#### WebSocket 端点
+- `/ws/chat-room/{room_id}?participant_type=user&participant_id=123&participant_name=John`
+- `/ws/chat-room/{room_id}?participant_type=agent&participant_id=agent-1&participant_name=Assistant`
 
-**User 消息：**
+#### ParticipantService
+统一处理用户和 Agent 的消息：
+- `process_message()` - 处理来自任何参与者的消息
+- `send_message()` - 主动发送消息
 
-```json
+### 6. 工具系统
-{
-  "text": "用户输入的文本内容",
-  "attachments": [
-    {"name": "utils.py", "extension": "py", "content": "..."}
-  ]
-}
-```
-
-**Assistant 消息：**
-
-```json
-{
-  "text": "AI 回复的文本内容",
-  "tool_calls": [...],
-  "steps": [
-    {"id": "step-0", "index": 0, "type": "thinking", "content": "..."},
-    {"id": "step-1", "index": 1, "type": "text", "content": "..."},
-    {"id": "step-2", "index": 2, "type": "tool_call", "id_ref": "call_xxx", "name": "...", "arguments": "..."},
-    {"id": "step-3", "index": 3, "type": "tool_result", "id_ref": "call_xxx", "name": "...", "content": "...", "success": true}
-  ]
-}
-```
-
-`steps` 字段是**渲染顺序的唯一数据源**，按 `index` 顺序排列。thinking、text、tool_call、tool_result 可以在多轮迭代中穿插出现。
-
-### 5. 工具系统
 
 ```mermaid
 classDiagram
@@ -219,21 +167,13 @@ classDiagram
         +register(tool) void
         +get(name) ToolDefinition?
         +list_all() List~dict~
-        +list_by_category(category) List~dict~
         +execute(name, arguments) dict
-        +remove(name) bool
     }
 
     class ToolExecutor {
         +enable_cache: bool
-        +cache_ttl: int
-        +max_workers: int
-        +_cache: Dict
-        +_call_history: List
         +process_tool_calls(tool_calls, context) list
         +process_tool_calls_parallel(tool_calls, context) list
-        +clear_cache() void
-        +get_history(limit) List
     }
 ```
 
@@ -250,8 +190,6 @@ classDiagram
 
 #### 工具开发规范
 
-所有工具必须遵循统一的开发规范，确保错误处理和返回格式一致。
-
 **核心原则：装饰器自动处理一切，工具函数只写业务逻辑**
 
 ```python
@@ -268,7 +206,7 @@ def my_tool(arguments: dict):
     # 业务逻辑 - 只管返回数据
     data = fetch_data(arguments["arg1"])
     return {"items": data, "count": len(data)}
-    
+
     # 或者直接抛出异常（装饰器自动捕获并转换）
     if invalid:
         raise ValueError("Invalid input")
@@ -279,7 +217,7 @@ def my_tool(arguments: dict):
 2. 所有异常捕获和转换
 3. 结果格式统一包装
 
-**返回格式转换**
+**返回格式转换：**
 
 | 工具函数返回/抛出 | 装饰器转换为 |
 |-------------------|-------------|
@@ -287,73 +225,52 @@ def my_tool(arguments: dict):
 | `raise ValueError("msg")` | `{"success": false, "data": null, "error": "ValueError: msg"}` |
 | `raise Exception()` | `{"success": false, "data": null, "error": "..."}` |
 
-**工具调用流程**
+#### 工具调用流程
 
-```
+```mermaid
-LLM 请求
+flowchart TD
-    ↓
+    A[LLM 请求] --> B{ToolRegistry.execute}
-ToolRegistry.execute(name, args)
+    B --> C[@tool 装饰器]
-    ↓
+    C --> D{验证 required_params}
-@tool 装饰器
+    D -->|失败| E[返回 error]
-    ├─ 验证 required_params
+    D -->|成功| F[执行工具函数]
-    ├─ 执行工具函数 (try-except 包裹)
+    F --> G{try-except}
-    ├─ 捕获异常 → 转换为 error
+    G -->|成功| H[包装 ToolResult]
-    ├─ 包装返回格式
+    G -->|异常| I[捕获并转换 error]
-    └─ 返回 ToolResult
+    I --> H
-    ↓
+    H --> J[返回 ToolResult]
-ToolExecutor 返回结果
+    J --> K[前端 ProcessBlock 显示]
-    ↓
-前端 ProcessBlock 显示
 ```
 
-### 6. 服务层
+#### Message Content JSON 结构
 
-#### ChatService (`services/chat.py`)
+`content` 字段统一使用 JSON 格式存储：
-核心聊天服务：
-- Agentic Loop 迭代执行（最多 10 轮）
-- 流式 SSE 响应
-- 工具调用编排（并行执行）
-- 消息历史管理
-- 自动重试机制
-- 支持 thinking_content 提取
-- Token 用量追踪
 
-#### LLMClient (`services/llm_client.py`)
+**User 消息：**
-LLM API 客户端：
+```json
-- 多提供商：DeepSeek、GLM、OpenAI
+{
-- 流式/同步调用
+  "text": "用户输入的文本内容",
-- 错误处理和重试
+  "attachments": [
-- Token 计数
+    {"name": "utils.py", "extension": "py", "content": "..."}
+  ]
+}
+```
 
-### 7. 认证系统 (`routes/auth.py`)
+**Assistant 消息：**
-- JWT Bearer Token
+```json
-- Bcrypt 密码哈希
+{
-- 用户注册/登录
+  "text": "AI 回复的文本内容",
+  "tool_calls": [...],
+  "steps": [
+    {"id": "step-0", "index": 0, "type": "thinking", "content": "..."},
+    {"id": "step-1", "index": 1, "type": "text", "content": "..."},
+    {"id": "step-2", "index": 2, "type": "tool_call", "id_ref": "call_xxx", "name": "...", "arguments": "..."},
+    {"id": "step-3", "index": 3, "type": "tool_result", "id_ref": "call_xxx", "name": "...", "content": "...", "success": true}
+  ]
+}
+```
 
-### 8. API 路由
+**steps 字段是渲染顺序的唯一数据源**，按 `index` 顺序排列。
-
-| 路由 | 方法 | 说明 |
-|------|------|------|
-| `/auth/register` | POST | 用户注册 |
-| `/auth/login` | POST | 用户登录 |
-| `/conversations` | GET | 会话列表（分页） |
-| `/conversations` | POST | 创建会话 |
-| `/conversations/{id}` | GET | 会话详情 |
-| `/conversations/{id}` | PUT | 更新会话 |
-| `/conversations/{id}` | DELETE | 删除会话 |
-| `/messages` | GET | 消息列表 |
-| `/messages` | POST | 发送消息（同步） |
-| `/messages/stream` | POST | 发送消息（流式 SSE） |
-| `/messages/{id}` | DELETE | 删除消息 |
-| `/providers` | GET | LLM 提供商列表 |
-| `/providers` | POST | 创建提供商 |
-| `/providers/{id}` | GET | 提供商详情 |
-| `/providers/{id}` | PUT | 更新提供商 |
-| `/providers/{id}` | DELETE | 删除提供商 |
-| `/providers/{id}/test` | POST | 测试提供商连接 |
-| `/tools` | GET | 可用工具列表 |
-| `/health` | GET | 健康检查 |
-| `/` | GET | 服务信息 |
 
 ## 数据流
 
@@ -362,63 +279,43 @@ LLM API 客户端：
 ```mermaid
 sequenceDiagram
     participant Client
-    participant API as POST /messages/stream
+    participant API as WebSocket/API
-    participant CS as ChatService
+    participant PS as ParticipantService
+    participant RS as ChatRoomService
     participant LLM as LLM API
     participant TE as ToolExecutor
-    
+
-    Client->>API: POST {content, tools, thinking_enabled}
+    Client->>API: send_message(content)
-    API->>CS: stream_response()
+    API->>PS: process_message()
-    
+    PS->>RS: save_message()
-    loop MAX_ITERATIONS (10)
+    PS->>RS: dispatch to agents
-        CS->>LLM: call(messages, tools)
+
-        LLM-->>CS: SSE Stream
+    loop Agentic Loop
-        
+        RS->>LLM: call(messages, tools)
+        LLM-->>RS: SSE Stream
         alt tool_calls
-            CS->>TE: process_tool_calls_parallel()
+            RS->>TE: process_tool_calls_parallel()
-            TE-->>CS: tool_results
+            TE-->>RS: tool_results
-            CS->>CS: 追加到 messages
         end
     end
-    
+
-    CS->>CS: _save_message()
+    RS-->>PS: process_step events
-    CS->>API: SSE Stream
+    PS-->>Client: SSE Stream
-    API-->>Client: 流式响应
 ```
 
-## SSE 事件
+## WebSocket 事件
 
 | 事件 | 说明 |
 |------|------|
-| `process_step` | 结构化步骤（thinking/text/tool_call/tool_result），携带 `id`、`index` 确保渲染顺序 |
+| `connected` | 连接成功 |
-| `done` | 响应完成，携带 message_id、token_count、usage |
+| `history` | 历史消息 |
+| `agents` | 聊天室 Agent 列表 |
+| `message` | 新消息 |
+| `typing` | 打字状态 |
+| `process_step` | Agent 处理步骤 |
+| `done` | 处理完成 |
 | `error` | 错误信息 |
-
+| `system` | 系统消息 (join/leave) |
-### process_step 事件格式
-
-```json
-{"type": "process_step", "step": {"id": "step-0", "index": 0, "type": "thinking", "content": "..."}}
-{"type": "process_step", "step": {"id": "step-1", "index": 1, "type": "text", "content": "回复文本..."}}
-{"type": "process_step", "step": {"id": "step-2", "index": 2, "type": "tool_call", "id_ref": "call_abc", "name": "web_search", "arguments": "{\"query\": \"...\"}"}}
-{"type": "process_step", "step": {"id": "step-3", "index": 3, "type": "tool_result", "id_ref": "call_abc", "name": "web_search", "content": "{...}", "success": true}}
-```
-
-| 字段 | 说明 |
-|------|------|
-| `id` | 步骤唯一标识（格式 `step-{index}`） |
-| `index` | 步骤序号，确保按正确顺序显示 |
-| `type` | 步骤类型：`thinking` / `text` / `tool_call` / `tool_result` |
-| `id_ref` | 工具调用引用 ID（仅 tool_call/tool_result） |
-| `name` | 工具名称（仅 tool_call/tool_result） |
-| `arguments` | 工具调用参数 JSON 字符串（仅 tool_call） |
-| `content` | 内容（thinking 的思考内容、text 的文本、tool_result 的返回结果） |
-| `success` | 工具执行是否成功（仅 tool_result） |
-
-### done 事件格式
-
-```json
-{"type": "done", "message_id": "uuid", "token_count": 1234, "usage": {"prompt_tokens": 100, "completion_tokens": 200, "total_tokens": 300}}
-```
 
 ## 配置示例
 
@@ -454,35 +351,3 @@ tools:
 | `APP_SECRET_KEY` | 应用密钥 | `your-secret-key` |
 | `DEEPSEEK_API_KEY` | DeepSeek API | `sk-xxxx` |
 | `DATABASE_URL` | 数据库连接 | `sqlite:///./chat.db` |
-
-## 项目结构说明
-
-### 入口文件
-
-- `run.py` - 启动 Uvicorn 服务器
-
-### 响应格式
-
-所有 API 统一使用响应封装：
-
-```json
-// 成功
-{"success": true, "data": {...}, "message": "操作成功"}
-
-// 错误
-{"success": false, "error": "错误信息", "code": 404}
-```
-
-### 工具缓存机制
-
-ToolExecutor 支持结果缓存：
-- TTL: 5 分钟（可配置）
-- 缓存 Key: `{tool_name}:{sorted_arguments_json}`
-- 调用历史记录最近 1000 条
-
-### 流式响应特点
-
-1. 实时返回 thinking_content（模型思考过程）
-2. 实时返回 text 增量更新
-3. 工具调用串行执行，结果批量返回
-4. 最终 `done` 事件包含完整 message_id 和 token 用量

assets/DATABASE.md

@@ -90,21 +90,13 @@ erDiagram
         int owner_id FK
         bool is_active
         datetime created_at
-    }
+        datetime updated_at
-
-    CHAT_ROOM_AGENT {
-        string id PK
-        string chat_room_id FK
-        string agent_id FK
-        bool is_active
-        datetime joined_at
     }
 
     CHAT_ROOM_MESSAGE {
         string id PK
         string room_id FK
-        string sender_type
+        string participant_id
-        string sender_id
         string sender_name
         text content
         string mentions
@@ -122,8 +114,6 @@ erDiagram
     LLM_PROVIDER ||--o{ AGENT : configures
     CONVERSATION ||--o{ MESSAGE : contains
     CHAT_ROOM ||--o{ CHAT_ROOM_MESSAGE : contains
-    CHAT_ROOM ||--o{ CHAT_ROOM_AGENT : has
-    AGENT ||--o{ CHAT_ROOM_AGENT : joins
 ```
 
 ## 3. 表设计
@@ -227,24 +217,13 @@ erDiagram
 | is_active | BOOLEAN | DEFAULT 1 | 是否启用 |
 | created_at | DATETIME | - | 创建时间 |
 
-### chat_room_agents 聊天室成员表
-
-| 字段 | 类型 | 约束 | 说明 |
-|------|------|------|------|
-| id | VARCHAR(64) | PK | 关联 ID |
-| chat_room_id | VARCHAR(64) | FK(chat_rooms.id) | 聊天室 ID |
-| agent_id | VARCHAR(64) | FK(agents.id) | Agent ID |
-| is_active | BOOLEAN | DEFAULT 1 | 是否启用 |
-| joined_at | DATETIME | - | 加入时间 |
-
 ### chat_room_messages 聊天室消息表
 
 | 字段 | 类型 | 约束 | 说明 |
 |------|------|------|------|
 | id | VARCHAR(64) | PK | 消息 ID |
 | room_id | VARCHAR(64) | FK(chat_rooms.id) | 聊天室 ID |
-| sender_type | VARCHAR(16) | NOT NULL | user/agent/system |
+| participant_id | VARCHAR(64) | NOT NULL | 统一参与者 ID (格式: "user:123" 或 "agent:abc") |
-| sender_id | VARCHAR(64) | NOT NULL | 发送者 ID |
 | sender_name | VARCHAR(50) | NOT NULL | 发送者名称 |
 | content | TEXT | NOT NULL | 消息内容 |
 | mentions | TEXT | - | @ 提及列表(JSON) |
@@ -252,32 +231,41 @@ erDiagram
 | token_count | INTEGER | DEFAULT 0 | Token 数 |
 | created_at | DATETIME | - | 创建时间 |
 
+**participant_id 格式:**
+- 用户: `user:{user_id}` (如 `user:123`)
+- Agent: `agent:{agent_id}` (如 `agent:abc-001`)
+
 ## 4. 流程图
 
-### 消息流
+### 消息流（统一参与者）
 
 ```mermaid
 sequenceDiagram
-    participant U as 用户
+    participant U as User
     participant WS as WebSocket
-    participant R as ChatRoomService
+    participant PS as ParticipantService
+    participant RS as ChatRoomService
     participant A as Agent
     participant LLM as LLM
 
     U->>WS: 发送消息
-    WS->>R: dispatch
+    WS->>PS: process_message()
-    R->>A: 触发 Agent
+    PS->>RS: save_message()
+    RS->>RS: 创建 participant_id
+    RS-->>PS: 消息保存成功
+    PS->>A: 触发 Agent
     A->>LLM: stream_call
     LLM-->>A: SSE 流
-    A-->>WS: process_step
+    A-->>PS: process_step
-    WS-->>U: 实时推送
+    PS-->>WS: 实时推送
+    WS-->>U: 显示
 ```
 
 ### Agent 调度
 
 ```mermaid
 flowchart TD
-    A[用户消息] --> B{有 @ 提及?}
+    A[用户/Agent 消息] --> B{有 @ 提及?}
 
     B -->|是| C[触发指定 Agent]
     B -->|否| D[触发 auto_response Agent]

luxx/__init__.py

@@ -18,7 +18,7 @@ async def lifespan(app: FastAPI):
     # Import all models to ensure they are registered with Base
     from luxx.models.user import User, LLMProvider, Project
     from luxx.models.chat import Conversation, Message
-    from luxx.models.room import ChatRoom, Agent, ChatRoomAgent, ChatRoomMessage
+    from luxx.models.room import ChatRoom, Agent
     init_db()
 
     # Create default test user if not exists
@@ -71,7 +71,25 @@ def create_app() -> FastAPI:
     from luxx.services.room_ws import websocket_handler
 
     @app.websocket("/ws/chat-room/{room_id}")
-    async def chat_room_websocket(websocket: WebSocket, room_id: str):
+    async def chat_room_websocket(
+        websocket: WebSocket,
+        room_id: str,
+        participant_id: str = None,
+        participant_type: str = "user",
+        participant_name: str = None
+    ):
+        """
+        WebSocket endpoint for chat room participation.
+
+        Query parameters:
+        - participant_id: ID of the user or agent connecting
+        - participant_type: "user" or "agent" (default: "user")
+        - participant_name: Display name for the participant
+
+        Example:
+        - User: ws://host/ws/chat-room/{room_id}?participant_id=123&participant_type=user&participant_name=John
+        - Agent: ws://host/ws/chat-room/{room_id}?participant_id=agent-1&participant_type=agent&participant_name=Assistant
+        """
         await websocket_handler(websocket, room_id)
 
     # Health check

luxx/agents/builtins/__init__.py

@@ -1 +0,0 @@
-# Builtins package - user-defined agent templates can be placed here

luxx/models/__init__.py

@@ -1,10 +1,12 @@
 """Models package"""
 from luxx.models.user import User, LLMProvider, Project
 from luxx.models.chat import Conversation, Message
-from luxx.models.room import ChatRoom, Agent, ChatRoomAgent, ChatRoomMessage
+from luxx.models.room import ChatRoom, Agent
+from luxx.models.participant import Participant, ParticipantType
 
 __all__ = [
     "User", "LLMProvider", "Project",
     "Conversation", "Message",
-    "ChatRoom", "Agent", "ChatRoomAgent", "ChatRoomMessage"
+    "ChatRoom", "Agent",
+    "Participant", "ParticipantType",
 ]

luxx/models/chat.py

@@ -1,4 +1,5 @@
 """Chat-related models"""
+import json
 from datetime import datetime
 from typing import Optional, List
 from sqlalchemy import String, Integer, Boolean, Float, Text, DateTime, ForeignKey
@@ -18,17 +19,16 @@ class Conversation(Base):
     id: Mapped[str] = mapped_column(String(64), primary_key=True)
     user_id: Mapped[int] = mapped_column(Integer, ForeignKey("users.id"), nullable=False)
     provider_id: Mapped[Optional[int]] = mapped_column(Integer, ForeignKey("llm_providers.id"), nullable=True)
-    project_id: Mapped[Optional[str]] = mapped_column(String(64), ForeignKey("projects.id"), nullable=True)
+    project_id: Mapped[Optional[str]] = mapped_column(String(64), nullable=True)
     title: Mapped[str] = mapped_column(String(255), nullable=False)
     model: Mapped[str] = mapped_column(String(64), nullable=False, default="deepseek-chat")
-    system_prompt: Mapped[str] = mapped_column(Text, nullable=False, default="You are a helpful assistant.")
+    system_prompt: Mapped[str] = mapped_column(Text, nullable=False, default="You are helpful.")
     temperature: Mapped[float] = mapped_column(Float, default=0.7)
     max_tokens: Mapped[int] = mapped_column(Integer, default=2000)
     thinking_enabled: Mapped[bool] = mapped_column(Boolean, default=False)
     created_at: Mapped[datetime] = mapped_column(DateTime, default=local_now)
     updated_at: Mapped[datetime] = mapped_column(DateTime, default=local_now, onupdate=local_now)
 
-    # Relationships
     user: Mapped["User"] = relationship("User", back_populates="conversations")
     provider: Mapped[Optional["LLMProvider"]] = relationship("LLMProvider")
     messages: Mapped[List["Message"]] = relationship(
@@ -53,31 +53,43 @@ class Conversation(Base):
 
 
 class Message(Base):
-    """Message model.
+    """Unified Message model for Conversation and ChatRoom.
 
-    content 字段统一使用 JSON 格式存储：
+    role: user/assistant/system/tool
+    content: JSON format with text, attachments, tool_calls, steps
     """
     __tablename__ = "messages"
 
     id: Mapped[str] = mapped_column(String(64), primary_key=True)
-    conversation_id: Mapped[str] = mapped_column(String(64), ForeignKey("conversations.id"), nullable=False)
+    conversation_id: Mapped[Optional[str]] = mapped_column(String(64), ForeignKey("conversations.id"), nullable=True)
+    room_id: Mapped[Optional[str]] = mapped_column(String(64), nullable=True)
     role: Mapped[str] = mapped_column(String(16), nullable=False)
     content: Mapped[str] = mapped_column(Text, nullable=False, default="")
+    sender_name: Mapped[str] = mapped_column(String(50), nullable=False, default="")
+    mentions: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
     token_count: Mapped[int] = mapped_column(Integer, default=0)
     usage: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
+
+    conversation: Mapped[Optional["Conversation"]] = relationship("Conversation", back_populates="messages")
     created_at: Mapped[datetime] = mapped_column(DateTime, default=local_now)
 
-    # Relationships
+    @property
-    conversation: Mapped["Conversation"] = relationship("Conversation", back_populates="messages")
+    def target_type(self) -> str:
+        return "conversation" if self.conversation_id else "room"
+
+    @property
+    def target_id(self) -> str:
+        return self.conversation_id or self.room_id or ""
 
     def to_dict(self):
-        """Convert to dictionary, extracting process_steps for frontend"""
-        import json
-
         result = {
             "id": self.id,
             "conversation_id": self.conversation_id,
+            "room_id": self.room_id,
+            "target_type": self.target_type,
+            "target_id": self.target_id,
             "role": self.role,
+            "sender_name": self.sender_name,
             "token_count": self.token_count,
             "created_at": self.created_at.isoformat() if self.created_at else None
         }
@@ -89,11 +101,19 @@ class Message(Base):
             except json.JSONDecodeError:
                 result["usage"] = None
 
+        # Parse mentions JSON
+        if self.mentions:
+            try:
+                result["mentions"] = json.loads(self.mentions)
+            except json.JSONDecodeError:
+                result["mentions"] = []
+        else:
+            result["mentions"] = []
+
         # Parse content JSON
         try:
             content_obj = json.loads(self.content) if self.content else {}
         except json.JSONDecodeError:
-            result["content"] = self.content
             result["text"] = self.content
             result["attachments"] = []
             result["tool_calls"] = []
@@ -105,7 +125,7 @@ class Message(Base):
         result["tool_calls"] = content_obj.get("tool_calls", [])
         result["process_steps"] = content_obj.get("steps", [])
 
-        if "content" not in result:
+        if "content" not in content_obj:
             result["content"] = result["text"]
 
         return result

luxx/models/participant.py

@@ -0,0 +1,72 @@
+"""Participant model - unified participant for users and agents."""
+from datetime import datetime
+from typing import Optional, Dict, Any, TYPE_CHECKING
+from dataclasses import dataclass
+from enum import Enum
+
+if TYPE_CHECKING:
+    from luxx.models.user import User
+
+
+class ParticipantType(Enum):
+    USER = "user"
+    AGENT = "agent"
+
+
+@dataclass
+class Participant:
+    """Unified participant abstraction for users and agents."""
+    participant_id: str
+    name: str
+    participant_type: ParticipantType
+    avatar: Optional[str] = None
+    role: Optional[str] = None  # ChatRoom role: owner/admin/member
+    permission_level: int = 1
+    is_active: bool = True
+    created_at: Optional[datetime] = None
+
+    @property
+    def is_agent(self) -> bool:
+        return self.participant_type == ParticipantType.AGENT
+
+    @property
+    def is_user(self) -> bool:
+        return self.participant_type == ParticipantType.USER
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "id": self.participant_id,
+            "name": self.name,
+            "type": self.participant_type.value,
+            "avatar": self.avatar,
+            "role": self.role,
+            "permission_level": self.permission_level,
+            "is_active": self.is_active,
+            "created_at": self.created_at.isoformat() if self.created_at else None
+        }
+
+    @classmethod
+    def from_user(cls, user: "User") -> "Participant":
+        return cls(
+            participant_id=f"user:{user.id}",
+            name=user.username,
+            participant_type=ParticipantType.USER,
+            permission_level=user.permission_level,
+            is_active=user.is_active,
+            created_at=user.created_at
+        )
+
+    @classmethod
+    def from_agent(
+        cls, agent_id: str, name: str, role: str = None,
+        avatar: str = None, permission_level: int = 1, is_active: bool = True
+    ) -> "Participant":
+        return cls(
+            participant_id=f"agent:{agent_id}",
+            name=name,
+            participant_type=ParticipantType.AGENT,
+            role=role,
+            avatar=avatar,
+            permission_level=permission_level,
+            is_active=is_active
+        )

luxx/models/room.py

@@ -1,18 +1,21 @@
-"""ChatRoom models"""
+"""ChatRoom models - unified participant architecture"""
 from datetime import datetime
-from typing import Optional, List
+from typing import Optional, List, TYPE_CHECKING
 from sqlalchemy import String, Integer, Boolean, Text, DateTime, ForeignKey
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from luxx.core.database import Base
 
+if TYPE_CHECKING:
+    from luxx.models.user import User
+
 
 def local_now():
     return datetime.now()
 
 
 class ChatRoom(Base):
-    """Chat Room model - like a group chat for multiple agents"""
+    """Chat Room - group chat for multiple participants"""
     __tablename__ = "chat_rooms"
 
     id: Mapped[str] = mapped_column(String(64), primary_key=True)
@@ -23,39 +26,26 @@ class ChatRoom(Base):
     created_at: Mapped[datetime] = mapped_column(DateTime, default=local_now)
     updated_at: Mapped[datetime] = mapped_column(DateTime, default=local_now, onupdate=local_now)
 
-    # Relationships
     owner: Mapped["User"] = relationship("User", backref="chat_rooms")
-    agents: Mapped[List["ChatRoomAgent"]] = relationship(
-        "ChatRoomAgent", back_populates="chat_room", cascade="all, delete-orphan"
-    )
-    messages: Mapped[List["ChatRoomMessage"]] = relationship(
-        "ChatRoomMessage", back_populates="chat_room", cascade="all, delete-orphan"
-    )
 
-    def to_dict(self, include_agents: bool = False):
+    def to_dict(self):
-        result = {
+        return {
-            "id": self.id,
+            "id": self.id, "name": self.name, "description": self.description,
-            "name": self.name,
+            "owner_id": self.owner_id, "is_active": self.is_active,
-            "description": self.description,
-            "owner_id": self.owner_id,
-            "is_active": self.is_active,
             "created_at": self.created_at.isoformat() if self.created_at else None,
             "updated_at": self.updated_at.isoformat() if self.updated_at else None
         }
-        if include_agents and self.agents:
-            result["agents"] = [ca.to_dict() for ca in self.agents]
-        return result
 
 
 class Agent(Base):
-    """Agent model - defines an AI agent with specific role"""
+    """Agent model - defines an AI agent"""
     __tablename__ = "agents"
 
     id: Mapped[str] = mapped_column(String(64), primary_key=True)
     name: Mapped[str] = mapped_column(String(50), nullable=False)
     role: Mapped[str] = mapped_column(String(50), nullable=False, default="helper")
     avatar: Mapped[Optional[str]] = mapped_column(String(500), nullable=True)
-    system_prompt: Mapped[str] = mapped_column(Text, nullable=False, default="You are a helpful assistant.")
+    system_prompt: Mapped[str] = mapped_column(Text, nullable=False, default="You are helpful.")
     provider_id: Mapped[Optional[int]] = mapped_column(Integer, ForeignKey("llm_providers.id"), nullable=True)
     model: Mapped[Optional[str]] = mapped_column(String(100), nullable=True)
     tools: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
@@ -68,105 +58,24 @@ class Agent(Base):
     created_at: Mapped[datetime] = mapped_column(DateTime, default=local_now)
     updated_at: Mapped[datetime] = mapped_column(DateTime, default=local_now, onupdate=local_now)
 
-    # Relationships
     provider: Mapped[Optional["LLMProvider"]] = relationship("LLMProvider")
-    chat_rooms: Mapped[List["ChatRoomAgent"]] = relationship(
-        "ChatRoomAgent", back_populates="agent", cascade="all, delete-orphan"
-    )
 
     def to_dict(self, include_secrets: bool = False):
         import json
         result = {
-            "id": self.id,
+            "id": self.id, "name": self.name, "role": self.role, "avatar": self.avatar,
-            "name": self.name,
+            "system_prompt": self.system_prompt, "provider_id": self.provider_id, "model": self.model,
-            "role": self.role,
+            "is_active": self.is_active, "priority": self.priority,
-            "avatar": self.avatar,
+            "auto_response": self.auto_response, "mention_trigger": self.mention_trigger,
-            "system_prompt": self.system_prompt,
-            "provider_id": self.provider_id,
-            "model": self.model,
-            "is_active": self.is_active,
-            "priority": self.priority,
-            "auto_response": self.auto_response,
-            "mention_trigger": self.mention_trigger,
             "temperature": float(self.temperature) if self.temperature else 0.7,
             "max_tokens": self.max_tokens,
             "created_at": self.created_at.isoformat() if self.created_at else None,
             "updated_at": self.updated_at.isoformat() if self.updated_at else None
         }
-        if self.tools:
+        try:
-            try:
+            result["tools"] = json.loads(self.tools) if self.tools else []
-                result["tools"] = json.loads(self.tools)
+        except json.JSONDecodeError:
-            except json.JSONDecodeError:
-                result["tools"] = []
-        else:
             result["tools"] = []
-
         if include_secrets and self.provider:
             result["provider"] = self.provider.to_dict(include_key=True)
         return result
-
-
-class ChatRoomAgent(Base):
-    """Association table for ChatRoom and Agent"""
-    __tablename__ = "chat_room_agents"
-
-    id: Mapped[str] = mapped_column(String(64), primary_key=True)
-    chat_room_id: Mapped[str] = mapped_column(String(64), ForeignKey("chat_rooms.id"), nullable=False)
-    agent_id: Mapped[str] = mapped_column(String(64), ForeignKey("agents.id"), nullable=False)
-    is_active: Mapped[bool] = mapped_column(Boolean, default=True)
-    joined_at: Mapped[datetime] = mapped_column(DateTime, default=local_now)
-
-    # Relationships
-    chat_room: Mapped["ChatRoom"] = relationship("ChatRoom", back_populates="agents")
-    agent: Mapped["Agent"] = relationship("Agent", back_populates="chat_rooms")
-
-    def to_dict(self):
-        return {
-            "id": self.id,
-            "chat_room_id": self.chat_room_id,
-            "agent_id": self.agent_id,
-            "is_active": self.is_active,
-            "joined_at": self.joined_at.isoformat() if self.joined_at else None,
-            "agent": self.agent.to_dict() if self.agent else None
-        }
-
-
-class ChatRoomMessage(Base):
-    """Chat Room Message model"""
-    __tablename__ = "chat_room_messages"
-
-    id: Mapped[str] = mapped_column(String(64), primary_key=True)
-    room_id: Mapped[str] = mapped_column(String(64), ForeignKey("chat_rooms.id"), nullable=False)
-    sender_type: Mapped[str] = mapped_column(String(16), nullable=False)
-    sender_id: Mapped[str] = mapped_column(String(64), nullable=False)
-    sender_name: Mapped[str] = mapped_column(String(50), nullable=False)
-    content: Mapped[str] = mapped_column(Text, nullable=False, default="")
-    mentions: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
-    parent_id: Mapped[Optional[str]] = mapped_column(String(64), nullable=True)
-    token_count: Mapped[int] = mapped_column(Integer, default=0)
-    created_at: Mapped[datetime] = mapped_column(DateTime, default=local_now)
-
-    # Relationships
-    chat_room: Mapped["ChatRoom"] = relationship("ChatRoom", back_populates="messages")
-
-    def to_dict(self):
-        import json
-        result = {
-            "id": self.id,
-            "room_id": self.room_id,
-            "sender_type": self.sender_type,
-            "sender_id": self.sender_id,
-            "sender_name": self.sender_name,
-            "content": self.content,
-            "parent_id": self.parent_id,
-            "token_count": self.token_count,
-            "created_at": self.created_at.isoformat() if self.created_at else None
-        }
-        if self.mentions:
-            try:
-                result["mentions"] = json.loads(self.mentions)
-            except json.JSONDecodeError:
-                result["mentions"] = []
-        else:
-            result["mentions"] = []
-        return result

luxx/models/user.py

@@ -1,11 +1,15 @@
 """User-related models"""
 from datetime import datetime
-from typing import Optional, List
+from typing import Optional, List, TYPE_CHECKING
 from sqlalchemy import String, Integer, Boolean, Text, DateTime, ForeignKey
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from luxx.core.database import Base
 
+# Avoid circular import at runtime
+if TYPE_CHECKING:
+    from luxx.models.chat import Conversation
+
 
 def local_now():
     return datetime.now()

luxx/services/__init__.py

@@ -5,11 +5,14 @@ from luxx.services.agent import agent_manager
 from luxx.services.llm_service import llm_service, LLMService
 from luxx.services.message_service import message_service, MessageService
 from luxx.services.stream_service import stream_service, StreamService
+from luxx.services.participant import participant_service, ParticipantService
 
 __all__ = [
     "chat_service",
     "chat_room_service",
     "agent_manager",
+    "participant_service",
+    "ParticipantService",
     "ChatService",
     "llm_service",
     "LLMService",

luxx/services/participant.py

@@ -0,0 +1,128 @@
+"""Participant Service - unified service for users and agents in chat rooms."""
+import logging
+from typing import Dict, Any, Optional, AsyncGenerator
+
+from luxx.agents.base import BaseAgent
+from luxx.agents.registry import agent_registry
+from luxx.models.participant import Participant
+from luxx.services.room import chat_room_service
+
+logger = logging.getLogger(__name__)
+
+
+class ParticipantService:
+    """Unified service for managing participants in chat rooms."""
+
+    def __init__(self):
+        self._active_agents: Dict[str, BaseAgent] = {}
+        self._active_users: Dict[int, Any] = {}
+
+    def _cm(self):
+        """Lazy import connection manager."""
+        from luxx.services.room_ws import connection_manager
+        return connection_manager
+
+    # ==================== Agent ====================
+
+    def register_agent(self, agent: BaseAgent) -> Participant:
+        self._active_agents[agent.agent_id] = agent
+        agent_registry.register(agent)
+        return Participant.from_agent(
+            agent.agent_id, agent.name, agent.role, agent.avatar,
+            agent.auto_response, agent.mention_trigger, agent.priority
+        )
+
+    def unregister_agent(self, agent_id: str) -> bool:
+        if agent_id in self._active_agents:
+            del self._active_agents[agent_id]
+            agent_registry.unregister(agent_id)
+            return True
+        return False
+
+    def get_agent_participant(self, agent_id: str) -> Optional[Participant]:
+        agent = self._active_agents.get(agent_id) or chat_room_service.get_agent(agent_id)
+        if agent:
+            if agent_id not in self._active_agents:
+                self._active_agents[agent_id] = agent
+            return Participant.from_agent(
+                agent.agent_id, agent.name, agent.role, agent.avatar,
+                agent.auto_response, agent.mention_trigger, agent.priority
+            )
+        return None
+
+    # ==================== User ====================
+
+    def register_user(self, user) -> Participant:
+        self._active_users[user.id] = user
+        return Participant.from_user(user)
+
+    def get_user_participant(self, user_id: int) -> Optional[Participant]:
+        user = self._active_users.get(user_id)
+        if not user:
+            from luxx.core.database import SessionLocal
+            from luxx.models.user import User
+            db = SessionLocal()
+            try:
+                user = db.query(User).filter(User.id == user_id).first()
+                if user:
+                    self._active_users[user_id] = user
+            finally:
+                db.close()
+        return Participant.from_user(user) if user else None
+
+    # ==================== Unified ====================
+
+    def get_participant(self, participant_id: str, ptype: str = "user") -> Optional[Participant]:
+        if ptype == "agent":
+            return self.get_agent_participant(participant_id)
+        try:
+            return self.get_user_participant(int(participant_id))
+        except (ValueError, TypeError):
+            return None
+
+    # ==================== Messages ====================
+
+    async def process_message(
+        self, room_id: str, content: str, sender_id: str,
+        sender_name: str, sender_type: str = "user", context: Dict = None
+    ) -> AsyncGenerator[Dict[str, Any], None]:
+        cm = self._cm()
+
+        msg = chat_room_service.save_message(room_id, sender_type, sender_id, sender_name, content)
+        await cm.broadcast_to_room(room_id, {"event": "message", "data": msg})
+
+        room_agents = chat_room_service.get_room_agents(room_id)
+        if sender_type == "agent":
+            room_agents = [a for a in room_agents if a.agent_id != sender_id]
+
+        for agent in room_agents:
+            await cm.broadcast_to_room(room_id, {
+                "event": "typing",
+                "data": {"agent_id": agent.agent_id, "agent_name": agent.name, "is_typing": True}
+            })
+
+        ctx = (context or {})
+        ctx.update({"sender_type": sender_type, "sender_id": sender_id, "username": sender_name})
+
+        async for event in chat_room_service.process_message(room_id, content, sender_id, sender_name, ctx):
+            yield event
+
+        for agent in room_agents:
+            await cm.broadcast_to_room(room_id, {
+                "event": "typing",
+                "data": {"agent_id": agent.agent_id, "agent_name": agent.name, "is_typing": False}
+            })
+
+    async def send_message(
+        self, room_id: str, participant_id: str,
+        participant_type: str, participant_name: str, content: str
+    ):
+        cm = self._cm()
+        msg = chat_room_service.save_message(
+            room_id, participant_type, participant_id, participant_name, content
+        )
+        await cm.broadcast_to_room(room_id, {"event": "message", "data": msg})
+        return msg
+
+
+participant_service = ParticipantService()

luxx/services/room.py

@@ -1,177 +1,19 @@
 """Chat Room Service - orchestrates multi-agent chat"""
 import json
-import re
 import uuid
-import asyncio
 import logging
 from typing import List, Dict, Any, Optional, AsyncGenerator
-from dataclasses import dataclass
 
 from luxx.core.database import SessionLocal
-from luxx.models.room import ChatRoom, Agent, ChatRoomAgent, ChatRoomMessage
+from luxx.models.room import ChatRoom, Agent
+from luxx.models.chat import Message
 from luxx.agents.base import BaseAgent
 
 logger = logging.getLogger(__name__)
 
 
-# ==================== Dispatcher ====================
-
-@dataclass
-class DispatchResult:
-    """Result of message dispatch"""
-    triggered_agents: List[BaseAgent]
-    mentions: List[str]
-    should_respond: bool
-
-
-class MessageDispatcher:
-    """Dispatcher for routing messages to agents"""
-
-    @staticmethod
-    def parse_mentions(content: str) -> List[str]:
-        """Parse @mentions from message content"""
-        pattern = r'@(\w+)'
-        return re.findall(pattern, content)
-
-    @staticmethod
-    def get_agents_by_names(names: List[str], room_agents: List[BaseAgent]) -> List[BaseAgent]:
-        """Get agents by their names (case-insensitive)"""
-        name_lower_map = {a.name.lower(): a for a in room_agents}
-        matched = []
-        for name in names:
-            agent = name_lower_map.get(name.lower())
-            if agent:
-                matched.append(agent)
-        return matched
-
-    @staticmethod
-    def get_agents_by_ids(agent_ids: List[str], room_agents: List[BaseAgent]) -> List[BaseAgent]:
-        """Get agents by their IDs"""
-        id_set = set(agent_ids)
-        return [a for a in room_agents if a.agent_id in id_set]
-
-    def dispatch(self, content: str, room_agents: List[BaseAgent], sender_id: str, sender_type: str = "user") -> DispatchResult:
-        """Dispatch a message to appropriate agents."""
-        available_agents = [a for a in room_agents if a.agent_id != sender_id]
-        mentions = self.parse_mentions(content)
-
-        if mentions:
-            triggered = self.get_agents_by_names(mentions, available_agents)
-            logger.info(f"Message with mentions: {mentions} -> triggered: {[a.name for a in triggered]}")
-            return DispatchResult(triggered_agents=triggered, mentions=mentions, should_respond=len(triggered) > 0)
-
-        auto_agents = [a for a in available_agents if a.auto_response]
-        auto_agents.sort(key=lambda a: a.priority)
-        logger.info(f"Auto-response agents triggered: {[a.name for a in auto_agents]}")
-        return DispatchResult(triggered_agents=auto_agents, mentions=[], should_respond=len(auto_agents) > 0)
-
-
-# ==================== Aggregator ====================
-
-class ResponseAggregator:
-    """Aggregates responses from multiple agents"""
-
-    def __init__(self, room_id: str):
-        self.room_id = room_id
-        self._agent_responses: Dict[str, Dict[str, Any]] = {}
-
-    async def aggregate_stream(self, agent_streams: Dict[str, AsyncGenerator]) -> AsyncGenerator[Dict[str, Any], None]:
-        """Aggregate streaming responses from multiple agents."""
-        if not agent_streams:
-            return
-
-        import asyncio
-
-        def parse_sse(event_str: str) -> Dict[str, Any]:
-            """Parse SSE string to dict."""
-            lines = event_str.strip().split('\n')
-            result = {"event": None, "data": {}}
-            for line in lines:
-                if line.startswith('event: '):
-                    result["event"] = line[7:].strip()
-                elif line.startswith('data: '):
-                    try:
-                        result["data"] = json.loads(line[6:].strip())
-                    except json.JSONDecodeError:
-                        result["data"] = {"content": line[6:].strip()}
-            return result
-
-        async def collect_agent_stream(agent_id: str, stream):
-            """Collect all events from a single agent stream."""
-            try:
-                async for event in stream:
-                    # Event is SSE string from BaseAgent
-                    parsed = parse_sse(event)
-                    parsed["agent_id"] = agent_id
-                    yield parsed
-            except Exception as e:
-                logger.error(f"Agent {agent_id} stream error: {e}")
-                yield {"event": "error", "agent_id": agent_id, "data": {"content": str(e)}}
-
-        # Use a queue-based approach for merging
-        queue = asyncio.Queue()
-
-        async def producer(agent_id: str, stream):
-            try:
-                async for event in stream:
-                    # Parse SSE string to dict if needed
-                    if isinstance(event, str):
-                        parsed = parse_sse(event)
-                        parsed["agent_id"] = agent_id
-                        await queue.put((agent_id, parsed))
-                    else:
-                        # Already a dict, just add agent_id
-                        if isinstance(event, dict):
-                            event["agent_id"] = agent_id
-                        await queue.put((agent_id, event))
-            except Exception as e:
-                logger.error(f"Agent {agent_id} stream error: {e}")
-                await queue.put((agent_id, {"event": "error", "agent_id": agent_id, "data": {"content": str(e)}}))
-            finally:
-                await queue.put((agent_id, None))  # Signal done
-
-        # Start all producers
-        producers = [
-            asyncio.create_task(producer(agent_id, stream))
-            for agent_id, stream in agent_streams.items()
-        ]
-
-        active = len(producers)
-        while active > 0:
-            agent_id, event = await queue.get()
-            if event is None:
-                active -= 1
-            else:
-                yield event
-
-        # Wait for all producers to complete
-        await asyncio.gather(*producers, return_exceptions=True)
-
-    def aggregate_final(self, responses: Dict[str, Dict[str, Any]]) -> List[Dict[str, Any]]:
-        """Aggregate final responses from agents."""
-        results = []
-        for agent_id, response in responses.items():
-            if response.get("event") == "done":
-                results.append({
-                    "agent_id": agent_id,
-                    "agent_name": response.get("agent_name"),
-                    "message_id": response.get("message_id"),
-                    "content": response.get("content"),
-                    "token_count": response.get("token_count", 0)
-                })
-        return results
-
-
-# ==================== Chat Room Service ====================
-
 class ChatRoomService:
-    """Service for managing chat rooms with multi-agent support"""
-
-    def __init__(self):
-        self.dispatcher = MessageDispatcher()
-
     def get_room(self, room_id: str) -> Optional[ChatRoom]:
-        """Get a chat room by ID"""
         db = SessionLocal()
         try:
             return db.query(ChatRoom).filter(ChatRoom.id == room_id).first()
@@ -179,50 +21,60 @@ class ChatRoomService:
             db.close()
 
     def get_room_agents(self, room_id: str) -> List[BaseAgent]:
-        """Get all active agents in a chat room"""
+        """Get active agents in a room from Message table"""
         db = SessionLocal()
         try:
-            room_agents = db.query(ChatRoomAgent).filter(
+            # Query distinct agent records from messages
-                ChatRoomAgent.chat_room_id == room_id,
+            messages = db.query(Message).filter(
-                ChatRoomAgent.is_active == True
+                Message.room_id == room_id,
-            ).all()
+                Message.role == "agent"
+            ).distinct().all()
+            
+            agent_ids = []
+            seen = set()
+            for msg in messages:
+                # Extract agent_id from content JSON
+                try:
+                    content = json.loads(msg.content) if msg.content else {}
+                    agent_id = content.get("agent_id")
+                    if agent_id and agent_id not in seen:
+                        seen.add(agent_id)
+                        agent_ids.append(agent_id)
+                except json.JSONDecodeError:
+                    pass
 
             agents = []
-            for ra in room_agents:
+            for agent_id in agent_ids:
-                agent_db = db.query(Agent).filter(Agent.id == ra.agent_id, Agent.is_active == True).first()
+                agent_db = db.query(Agent).filter(
+                    Agent.id == agent_id,
+                    Agent.is_active == True
+                ).first()
                 if agent_db:
                     agents.append(BaseAgent.from_model(agent_db))
-
+            
-            agents.sort(key=lambda a: a.priority)
+            return sorted(agents, key=lambda a: a.priority)
-            return agents
         finally:
             db.close()
 
     def get_agent(self, agent_id: str) -> Optional[BaseAgent]:
-        """Get an agent by ID"""
         db = SessionLocal()
         try:
             agent_db = db.query(Agent).filter(Agent.id == agent_id).first()
-            if agent_db:
+            return BaseAgent.from_model(agent_db) if agent_db else None
-                return BaseAgent.from_model(agent_db)
-            return None
         finally:
             db.close()
 
-    def list_rooms(self, user_id: int = None, include_agents: bool = True) -> List[Dict]:
+    def list_rooms(self, user_id: int = None) -> List[Dict]:
-        """List all chat rooms"""
         db = SessionLocal()
         try:
-            query = db.query(ChatRoom)
+            q = db.query(ChatRoom)
             if user_id:
-                query = query.filter(ChatRoom.owner_id == user_id)
+                q = q.filter(ChatRoom.owner_id == user_id)
-            rooms = query.order_by(ChatRoom.updated_at.desc()).all()
+            return [r.to_dict() for r in q.order_by(ChatRoom.updated_at.desc()).all()]
-            return [r.to_dict(include_agents=include_agents) for r in rooms]
         finally:
             db.close()
 
     def create_room(self, name: str, owner_id: int, description: str = None, agent_ids: List[str] = None) -> Dict:
-        """Create a new chat room"""
         db = SessionLocal()
         try:
             room = ChatRoom(
@@ -233,183 +85,174 @@ class ChatRoomService:
             )
             db.add(room)
 
-            if agent_ids:
+            # Record agents as join messages
-                for agent_id in agent_ids:
+            for agent_id in (agent_ids or []):
-                    room_agent = ChatRoomAgent(
+                msg = Message(
-                        id=str(uuid.uuid4()),
+                    id=str(uuid.uuid4()),
-                        chat_room_id=room.id,
+                    room_id=room.id,
-                        agent_id=agent_id
+                    role="agent",
-                    )
+                    content=json.dumps({"type": "join", "agent_id": agent_id}),
-                    db.add(room_agent)
+                    sender_name=agent_id
+                )
+                db.add(msg)
 
             db.commit()
-            return room.to_dict(include_agents=True)
+            return room.to_dict()
         finally:
             db.close()
 
-    def update_room(self, room_id: str, name: str = None, description: str = None, is_active: bool = None) -> Optional[Dict]:
+    def update_room(self, room_id: str, **kwargs) -> Optional[Dict]:
-        """Update a chat room"""
         db = SessionLocal()
         try:
             room = db.query(ChatRoom).filter(ChatRoom.id == room_id).first()
             if not room:
                 return None
-
+            for key, value in kwargs.items():
-            if name is not None:
+                if value is not None and hasattr(room, key):
-                room.name = name
+                    setattr(room, key, value)
-            if description is not None:
-                room.description = description
-            if is_active is not None:
-                room.is_active = is_active
-
             db.commit()
-            return room.to_dict(include_agents=True)
+            return room.to_dict()
         finally:
             db.close()
 
     def delete_room(self, room_id: str) -> bool:
-        """Delete a chat room"""
         db = SessionLocal()
         try:
             room = db.query(ChatRoom).filter(ChatRoom.id == room_id).first()
-            if not room:
+            if room:
-                return False
+                db.delete(room)
-            db.delete(room)
-            db.commit()
-            return True
-        finally:
-            db.close()
-
-    def add_agent_to_room(self, room_id: str, agent_id: str) -> bool:
-        """Add an agent to a chat room"""
-        db = SessionLocal()
-        try:
-            existing = db.query(ChatRoomAgent).filter(
-                ChatRoomAgent.chat_room_id == room_id,
-                ChatRoomAgent.agent_id == agent_id
-            ).first()
-
-            if existing:
-                existing.is_active = True
-            else:
-                room_agent = ChatRoomAgent(
-                    id=str(uuid.uuid4()),
-                    chat_room_id=room_id,
-                    agent_id=agent_id
-                )
-                db.add(room_agent)
-
-            db.commit()
-            return True
-        finally:
-            db.close()
-
-    def remove_agent_from_room(self, room_id: str, agent_id: str) -> bool:
-        """Remove an agent from a chat room"""
-        db = SessionLocal()
-        try:
-            room_agent = db.query(ChatRoomAgent).filter(
-                ChatRoomAgent.chat_room_id == room_id,
-                ChatRoomAgent.agent_id == agent_id
-            ).first()
-
-            if room_agent:
-                room_agent.is_active = False
                 db.commit()
                 return True
             return False
         finally:
             db.close()
 
-    def get_messages(self, room_id: str, limit: int = 50, before_id: str = None) -> List[Dict]:
+    def add_participant(
-        """Get messages from a chat room"""
+        self, room_id: str, agent_id: str = None, user_id: int = None
+    ) -> bool:
         db = SessionLocal()
         try:
-            query = db.query(ChatRoomMessage).filter(
+            role = "agent" if agent_id else "user"
-                ChatRoomMessage.room_id == room_id
+            sender_name = agent_id or f"user_{user_id}"
-            ).order_by(ChatRoomMessage.created_at.desc())
+            content_data = {"type": "join", "agent_id": agent_id} if agent_id else {"type": "join", "user_id": user_id}
-
+            
-            if before_id:
+            msg = Message(
-                before_msg = db.query(ChatRoomMessage).filter(
+                id=str(uuid.uuid4()),
-                    ChatRoomMessage.id == before_id
+                room_id=room_id,
-                ).first()
+                role=role,
-                if before_msg:
+                sender_name=sender_name,
-                    query = query.filter(ChatRoomMessage.created_at < before_msg.created_at)
+                content=json.dumps(content_data)
-
+            )
-            messages = query.limit(limit).all()
+            db.add(msg)
-            return [m.to_dict() for m in reversed(messages)]
+            db.commit()
+            return True
         finally:
             db.close()
 
-    def save_message(self, room_id: str, sender_type: str, sender_id: str, sender_name: str, content: str,
+    def remove_participant(self, room_id: str, participant_id: str) -> bool:
-                     mentions: List[str] = None, parent_id: str = None, token_count: int = 0) -> Dict:
-        """Save a message to a chat room"""
         db = SessionLocal()
         try:
-            msg = ChatRoomMessage(
+            db.query(Message).filter(
+                Message.id == participant_id
+            ).delete()
+            db.commit()
+            return True
+        finally:
+            db.close()
+
+    def get_messages(self, room_id: str, limit: int = 50, before_id: str = None) -> List[Dict]:
+        db = SessionLocal()
+        try:
+            q = db.query(Message).filter(Message.room_id == room_id).order_by(Message.created_at.desc())
+            if before_id:
+                before = db.query(Message).filter(Message.id == before_id).first()
+                if before:
+                    q = q.filter(Message.created_at < before.created_at)
+            return [m.to_dict() for m in reversed(q.limit(limit).all())]
+        finally:
+            db.close()
+
+    def save_message(
+        self,
+        room_id: str,
+        role: str,
+        sender_name: str,
+        content: str,
+        mentions: List[str] = None,
+        token_count: int = 0
+    ) -> Dict:
+        db = SessionLocal()
+        try:
+            msg = Message(
                 id=str(uuid.uuid4()),
                 room_id=room_id,
-                sender_type=sender_type,
+                role=role,
-                sender_id=sender_id,
                 sender_name=sender_name,
                 content=content,
                 mentions=json.dumps(mentions) if mentions else None,
-                parent_id=parent_id,
                 token_count=token_count
             )
             db.add(msg)
-
+            
+            # Update room updated_at
             room = db.query(ChatRoom).filter(ChatRoom.id == room_id).first()
             if room:
                 from datetime import datetime
                 room.updated_at = datetime.now()
-
+            
             db.commit()
             return msg.to_dict()
         finally:
             db.close()
 
-    async def process_message(self, room_id: str, user_message: str, user_id: str, user_name: str, context: Dict = None) -> AsyncGenerator[Dict[str, Any], None]:
+    async def process_message(
-        """Process a user message and dispatch to appropriate agents."""
+        self, room_id: str, user_message: str, sender_id: str, sender_name: str = None
+    ):
         room = self.get_room(room_id)
         if not room:
             yield {"event": "error", "data": {"content": "Chat room not found"}}
             return
 
-        room_agents = self.get_room_agents(room_id)
+        agents = self.get_room_agents(room_id)
-        if not room_agents:
+        if not agents:
-            yield {"event": "error", "data": {"content": "No agents available in this room"}}
+            yield {"event": "error", "data": {"content": "No agents available"}}
             return
 
-        dispatch_result = self.dispatcher.dispatch(
+        # Check if sender is agent
-            content=user_message,
+        from luxx.agents.registry import agent_registry
-            room_agents=room_agents,
+        sender_is_agent = agent_registry.get(sender_id) is not None
-            sender_id=user_id,
+        
-            sender_type="user"
+        # Filter out sender if agent
-        )
+        if sender_is_agent:
-
+            agents = [a for a in agents if a.agent_id != sender_id]
-        if not dispatch_result.should_respond:
+        
+        # Check mentions
+        import re
+        mentions = re.findall(r'@(\w+)', user_message)
+        triggered = []
+        
+        if mentions:
+            name_map = {a.name.lower(): a for a in agents}
+            triggered = [name_map[n.lower()] for n in mentions if n.lower() in name_map]
+        
+        if not triggered:
+            triggered = [a for a in agents if a.auto_response]
+            triggered.sort(key=lambda a: a.priority)
+        
+        if not triggered:
             yield {"event": "no_response", "data": {"message": "No agents triggered"}}
             return
 
+        # Get history
         messages = self.get_messages(room_id, limit=20)
-
+        
-        agent_streams = {}
+        # Stream responses
-        for agent in dispatch_result.triggered_agents:
+        for agent in triggered:
-            stream = agent.stream_response(
+            async for event in agent.stream_response(user_message, messages):
-                user_message=user_message,
+                yield event
-                conversation_history=messages,
+        
-                context=context
+        self.save_message(room_id, "user", sender_id, sender_name, user_message)
-            )
-            agent_streams[agent.agent_id] = stream
-
-        aggregator = ResponseAggregator(room_id)
-        async for event in aggregator.aggregate_stream(agent_streams):
-            yield event
 
 
-# Global service instance
+# Global instance
 chat_room_service = ChatRoomService()
-
-# Export for backward compatibility
-dispatcher = chat_room_service.dispatcher

luxx/services/room_ws.py

@@ -1,103 +1,82 @@
-"""WebSocket handler for Chat Rooms"""
+"""WebSocket handler for Chat Rooms - unified user and agent participants."""
-import json
-import asyncio
 import logging
 from typing import Dict, Set
 from fastapi import WebSocket, WebSocketDisconnect
 
 from luxx.services.room import chat_room_service
+from luxx.services.participant import participant_service
 
 logger = logging.getLogger(__name__)
 
 
-class ChatRoomConnectionManager:
+class ConnectionManager:
-    """Manages WebSocket connections for chat rooms"""
-
     def __init__(self):
-        self._room_connections: Dict[str, Set[WebSocket]] = {}
+        self._rooms: Dict[str, Set[WebSocket]] = {}
-        self._connection_rooms: Dict[WebSocket, str] = {}
+        self._info: Dict[WebSocket, Dict] = {}
 
-    async def connect(self, websocket: WebSocket, room_id: str):
+    async def connect(self, ws: WebSocket, room_id: str, ptype: str, pid: str, pname: str):
-        """Connect a WebSocket to a chat room"""
+        await ws.accept()
-        await websocket.accept()
+        self._rooms.setdefault(room_id, set()).add(ws)
+        self._info[ws] = {"type": ptype, "id": pid, "name": pname}
+        await ws.send_json({"event": "connected", "data": {"room_id": room_id, "type": ptype}})
 
-        if room_id not in self._room_connections:
+    def disconnect(self, ws: WebSocket):
-            self._room_connections[room_id] = set()
+        info = self._info.pop(ws, {})
+        room = self._rooms.get(self._info.get(ws, {}).get("id"))
+        if room:
+            room.discard(ws)
+            if not room:
+                del self._rooms[room]
+        return info
 
-        self._room_connections[room_id].add(websocket)
+    async def broadcast(self, room_id: str, msg: dict, exclude: WebSocket = None):
-        self._connection_rooms[websocket] = room_id
+        for ws in self._rooms.get(room_id, set()):
+            if ws != exclude:
+                try:
+                    await ws.send_json(msg)
+                except:
+                    self.disconnect(ws)
 
-        logger.info(f"WebSocket connected to room: {room_id}")
+    def size(self, room_id: str) -> int:
-
+        return len(self._rooms.get(room_id, set()))
-        await websocket.send_json({
-            "event": "connected",
-            "data": {"room_id": room_id, "message": "Connected to chat room"}
-        })
-
-    def disconnect(self, websocket: WebSocket):
-        """Disconnect a WebSocket from its room"""
-        room_id = self._connection_rooms.pop(websocket, None)
-        if room_id and room_id in self._room_connections:
-            self._room_connections[room_id].discard(websocket)
-            if not self._room_connections[room_id]:
-                del self._room_connections[room_id]
-        logger.info(f"WebSocket disconnected from room: {room_id}")
-
-    async def broadcast_to_room(self, room_id: str, message: dict, exclude: WebSocket = None):
-        """Broadcast a message to all connections in a room"""
-        if room_id not in self._room_connections:
-            return
-
-        disconnected = []
-        for connection in self._room_connections[room_id]:
-            if connection == exclude:
-                continue
-            try:
-                await connection.send_json(message)
-            except Exception as e:
-                logger.error(f"Failed to send to connection: {e}")
-                disconnected.append(connection)
-
-        for conn in disconnected:
-            self.disconnect(conn)
-
-    async def send_to_room(self, room_id: str, message: dict):
-        """Send a message to all connections in a room"""
-        await self.broadcast_to_room(room_id, message)
-
-    def get_room_size(self, room_id: str) -> int:
-        """Get the number of connections in a room"""
-        return len(self._room_connections.get(room_id, set()))
 
 
-# Global connection manager
+cm = ConnectionManager()
-connection_manager = ChatRoomConnectionManager()
 
 
-async def websocket_handler(websocket: WebSocket, room_id: str):
+async def websocket_handler(ws: WebSocket, room_id: str):
-    """Handle WebSocket connection for a chat room."""
+    params = dict(ws.query_params)
-    await connection_manager.connect(websocket, room_id)
+    ptype = params.get("participant_type", "user")
+    pid = params.get("participant_id", "")
+    pname = params.get("participant_name", "Anonymous")
+
+    await cm.connect(ws, room_id, ptype, pid, pname)
 
     room = chat_room_service.get_room(room_id)
     if not room:
-        await websocket.send_json({"event": "error", "data": {"content": "Chat room not found"}})
+        await ws.send_json({"event": "error", "data": {"content": "Room not found"}})
-        await websocket.close()
+        await ws.close()
         return
 
+    if ptype == "agent" and pid:
+        agent = chat_room_service.get_agent(pid)
+        if agent:
+            participant_service.register_agent(agent)
+
     try:
-        messages = chat_room_service.get_messages(room_id, limit=50)
+        # Send history
-        await websocket.send_json({"event": "history", "data": {"messages": messages}})
+        await ws.send_json({"event": "history", "data": {"messages": chat_room_service.get_messages(room_id)}})
+        await ws.send_json({"event": "agents", "data": {
+            "agents": [a.to_dict() for a in chat_room_service.get_room_agents(room_id)]
+        }})
 
-        agents = chat_room_service.get_room_agents(room_id)
+        await cm.broadcast(room_id, {
-        await websocket.send_json({"event": "agents", "data": {"agents": [a.to_dict() for a in agents]}})
-
-        await connection_manager.broadcast_to_room(room_id, {
             "event": "system",
-            "data": {"content": "User joined the room", "type": "join"}
+            "data": {"content": f"{pname} joined", "type": f"{ptype}_join"}
-        }, exclude=websocket)
+        }, exclude=ws)
 
         while True:
-            data = await websocket.receive_json()
+            data = await ws.receive_json()
             action = data.get("action")
 
             if action == "send_message":
@@ -105,87 +84,29 @@ async def websocket_handler(websocket: WebSocket, room_id: str):
                 if not content:
                     continue
 
-                user_id = str(data.get("user_id", "anonymous"))
+                sid = pid if ptype == "agent" else str(data.get("user_id", pid or "anonymous"))
-                user_name = data.get("user_name", "Anonymous")
+                sname = pname if ptype == "agent" else data.get("user_name", pname or "Anonymous")
 
-                msg = chat_room_service.save_message(
+                async for event in participant_service.process_message(
-                    room_id=room_id, sender_type="user", sender_id=user_id,
+                    room_id, content, sid, sname, ptype
-                    sender_name=user_name, content=content
-                )
-
-                await connection_manager.broadcast_to_room(room_id, {"event": "message", "data": msg})
-
-                agents = chat_room_service.get_room_agents(room_id)
-                for agent in agents:
-                    await connection_manager.broadcast_to_room(room_id, {
-                        "event": "typing",
-                        "data": {"agent_id": agent.agent_id, "agent_name": agent.name, "is_typing": True}
-                    })
-
-                context = {"user_id": user_id, "username": user_name}
-                logger.info(f"[ROOM_WS] Starting process_message, agents count: {len(agents)}")
-
-                event_count = 0
-                async for event in chat_room_service.process_message(
-                    room_id=room_id, user_message=content, user_id=user_id,
-                    user_name=user_name, context=context
                 ):
-                    event_count += 1
-                    logger.info(f"[ROOM_WS] Received event {event_count}: {event.get('event')}")
-                    
                     if event.get("event") in ["process_step", "done", "error"]:
-                        # Find agent_name from agents list
+                        await cm.broadcast(room_id, {
-                        agent_name = None
+                            "event": event["event"],
-                        for agent in agents:
-                            if agent.agent_id == event.get("agent_id"):
-                                agent_name = agent.name
-                                break
-                        await connection_manager.broadcast_to_room(room_id, {
-                            "event": event.get("event"),
                             "data": event.get("data", {}),
-                            "agent_id": event.get("agent_id"),
+                            "agent_id": event.get("agent_id")
-                            "agent_name": agent_name
                         })
 
-                        if event.get("event") == "done":
-                            # Find agent_name from agents list
-                            agent_name = None
-                            for agent in agents:
-                                if agent.agent_id == event.get("agent_id"):
-                                    agent_name = agent.name
-                                    break
-                            chat_room_service.save_message(
-                                room_id=room_id, sender_type="agent",
-                                sender_id=event.get("agent_id"),
-                                sender_name=agent_name,
-                                content=event.get("data", {}).get("content", "") if isinstance(event.get("data"), dict) else "",
-                                token_count=event.get("data", {}).get("token_count", 0) if isinstance(event.get("data"), dict) else 0
-                            )
-                    else:
-                        logger.info(f"[ROOM_WS] Skipping event: {event.get('event')}")
-                
-                logger.info(f"[ROOM_WS] process_message completed, total events: {event_count}")
-
-                for agent in agents:
-                    await connection_manager.broadcast_to_room(room_id, {
-                        "event": "typing",
-                        "data": {"agent_id": agent.agent_id, "agent_name": agent.name, "is_typing": False}
-                    })
-
             elif action == "ping":
-                await websocket.send_json({"event": "pong", "data": {}})
+                await ws.send_json({"event": "pong", "data": {}})
 
     except WebSocketDisconnect:
-        logger.info(f"WebSocket disconnected from room: {room_id}")
+        await cm.broadcast(room_id, {"event": "system", "data": {"content": f"{pname} left", "type": "leave"}})
-        connection_manager.disconnect(websocket)
-        await connection_manager.broadcast_to_room(room_id, {
-            "event": "system",
-            "data": {"content": "User left the room", "type": "leave"}
-        })
     except Exception as e:
-        logger.error(f"WebSocket error in room {room_id}: {e}")
+        logger.error(f"WebSocket error: {e}")
-        connection_manager.disconnect(websocket)
+        await cm.broadcast(room_id, {"event": "error", "data": {"content": str(e)}})
-        await connection_manager.broadcast_to_room(room_id, {
+    finally:
-            "event": "error",
+        cm.disconnect(ws)
-            "data": {"content": str(e)}
+
-        })
+
+connection_manager = cm