huihuiSquare/docs/superpowers/plans/2026-08-07-avatar-takeover-chat-plan.md

# 数字分身接管聊天功能 — 实现计划

> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.

**Goal:** 实现数字分身接管主人在会会平台上的聊天，支持立即接管和延迟接管两种模式，优先使用知识库生成回复。

**Architecture:** 在数字分身应用后端新增 takeover 模块，包含消息监听器、接管决策器、回复执行器。通过轮询会会平台推送消息接口获取未读消息，调用现有分身聊天接口生成回复，通过盒子 IM 接口以主人身份发送。

**Tech Stack:** FastAPI (Python), SQLite, Redis, HTTPX, APScheduler, Vue 3 (frontend)

## 全局约束

- 数据库：SQLite (avatar.db)，使用 `_try_add_columns` 模式添加新列
- 响应格式：统一使用 `ok()` / `fail()` from `responses.py`
- 认证：`Authorization: Bearer <app_token>` header
- 盒子 IM 接口前缀：`/api/im/box/*`（通过会会平台网关）
- 签名机制：复用 `news_service.py` 中的 `_build_form` / `_make_sign` 模式
- YAGNI：本期只做单聊接管，群聊接管标记为后续扩展
- TDD：每个功能模块先写测试再实现

---

## 文件清单

### 新建文件

| 文件 | 职责 |
|------|------|
| `digital-avatar-app/backend/routers/takeover.py` | 接管配置 API + 内部处理接口 |
| `digital-avatar-app/backend/services/takeover_service.py` | 接管核心逻辑（监听、决策、回复） |
| `digital-avatar-app/backend/services/boxim_client.py` | 盒子 IM 客户端（获取凭证、发送消息） |
| `digital-avatar-app/backend/requirements.txt` | 新增 redis、apscheduler 依赖 |
| `digital-avatar-app/src/api/index.ts` | 新增 takeover 相关 API 函数 |
| `digital-avatar-app/src/views/AuthorizationManage.vue` | 添加接管开关 UI |

### 修改文件

| 文件 | 修改内容 |
|------|---------|
| `digital-avatar-app/backend/models.py` | Authorization 表新增 takeover 字段 |
| `digital-avatar-app/backend/database.py` | `_try_add_columns` 添加 takeover 列 |
| `digital-avatar-app/backend/main.py` | 注册 takeover router，启动定时任务 |
| `digital-avatar-app/backend/requirements.txt` | 添加 redis, apscheduler |

---

## 任务分解

### Task 1: 扩展 Authorization 数据模型

**Files:**
- Modify: `digital-avatar-app/backend/models.py`
- Modify: `digital-avatar-app/backend/database.py`
- Test: `digital-avatar-app/backend/tests/test_takeover_model.py`

**Interfaces:**
- 新增字段：`takeover_enabled`, `takeover_mode`, `takeover_delay_seconds`
- `Authorization.to_dict()` 返回新增字段

- [ ] **Step 1: Write the failing test**

```python
# digital-avatar-app/backend/tests/test_takeover_model.py
from database import SessionLocal
from models import Authorization

def test_authorization_takeover_fields():
    db = SessionLocal()
    try:
        auth = db.query(Authorization).first()
        assert auth is not None
        # 检查新字段存在且有默认值
        assert hasattr(auth, 'takeover_enabled')
        assert hasattr(auth, 'takeover_mode')
        assert hasattr(auth, 'takeover_delay_seconds')
        assert auth.takeover_enabled == False
        assert auth.takeover_mode == 'immediate'
        assert auth.takeover_delay_seconds == 30
    finally:
        db.close()

def test_authorization_to_dict_includes_takeover():
    db = SessionLocal()
    try:
        auth = db.query(Authorization).first()
        d = auth.to_dict()
        assert 'takeover_enabled' in d
        assert 'takeover_mode' in d
        assert 'takeover_delay_seconds' in d
    finally:
        db.close()
```

- [ ] **Step 2: Run test to verify it fails**

```bash
cd digital-avatar-app/backend
python -m pytest tests/test_takeover_model.py -v
```
Expected: FAIL with "no such column: takeover_enabled"

- [ ] **Step 3: Update Authorization model**

```python
# digital-avatar-app/backend/models.py
# 在 Authorization 类中新增字段（在 created_at 之前）
class Authorization(Base):
    # ... 现有字段 ...
    takeover_enabled = Column(Boolean, default=False)  # 是否开启分身接管
    takeover_mode = Column(String, default="immediate")  # immediate | delayed
    takeover_delay_seconds = Column(Integer, default=30)  # 延迟秒数
    created_at = Column(DateTime, server_default=func.now())

    def to_dict(self):
        return {
            # ... 现有字段 ...
            "takeoverEnabled": self.takeover_enabled,
            "takeoverMode": self.takeover_mode,
            "takeoverDelaySeconds": self.takeover_delay_seconds,
            "createdAt": _iso(self.created_at),
        }
```

- [ ] **Step 4: Add column migration**

```python
# digital-avatar-app/backend/database.py
# 在 _try_add_columns 调用中新增：
_try_add_columns(
    # ... 现有列 ...
    ("authorizations", "takeover_enabled", "BOOLEAN DEFAULT 0"),
    ("authorizations", "takeover_mode", "VARCHAR DEFAULT 'immediate'"),
    ("authorizations", "takeover_delay_seconds", "INTEGER DEFAULT 30"),
)
```

- [ ] **Step 5: Run test to verify it passes**

```bash
cd digital-avatar-app/backend
python -m pytest tests/test_takeover_model.py -v
```
Expected: PASS

- [ ] **Step 6: Commit**

```bash
cd /Users/yqq/Works/AiProject/huihuiSquare
git add digital-avatar-app/backend/models.py digital-avatar-app/backend/database.py digital-avatar-app/backend/tests/test_takeover_model.py
git commit -m "feat: add takeover fields to Authorization model"
```

---

### Task 2: 盒子 IM 客户端

**Files:**
- Create: `digital-avatar-app/backend/services/boxim_client.py`
- Test: `digital-avatar-app/backend/tests/test_boxim_client.py`

**Interfaces:**
- `BoxIMClient.get_credentials(user_id: str) -> dict` — 获取网易云信凭证
- `BoxIMClient.send_p2p_message(from_accid: str, to_accid: str, content: str) -> bool` — 发送单聊消息

**Dependencies:**
- 会会平台签名机制（复用 news_service 中的模式）
- HTTPX for async HTTP calls
- 平台配置从环境变量读取

- [ ] **Step 1: Write the failing test**

```python
# digital-avatar-app/backend/tests/test_boxim_client.py
import pytest
from unittest.mock import AsyncMock, patch
from services.boxim_client import BoxIMClient

@pytest.fixture
def mock_config():
    return {
        "HUIHUI_IM_BASE_URL": "http://192.168.1.200:60040",
        "HUIHUI_APP_ID": "test_app",
        "HUIHUI_ACCESS_ID": "test_access",
        "HUIHUI_ACCESS_SECRET": "test_secret",
    }

@patch("services.boxim_client.httpx.AsyncClient")
@pytest.mark.asyncio
async def test_get_credentials(mock_client_class, mock_config):
    mock_response = AsyncMock()
    mock_response.json.return_value = {"code": 200, "data": {"accid": "user123", "token": "tok_xyz"}}
    mock_client_class.return_value.__aenter__.return_value.post.return_value = mock_response

    client = BoxIMClient(mock_config)
    result = await client.get_credentials("user123")

    assert result["accid"] == "user123"
    assert result["token"] == "tok_xyz"

@patch("services.boxim_client.httpx.AsyncClient")
@pytest.mark.asyncio
async def test_send_p2p_message_success(mock_client_class, mock_config):
    mock_response = AsyncMock()
    mock_response.json.return_value = {"code": 200}
    mock_client_class.return_value.__aenter__.return_value.post.return_value = mock_response

    client = BoxIMClient(mock_config)
    result = await client.send_p2p_message("owner_acc", "target_acc", "Hello")

    assert result is True
```

- [ ] **Step 2: Run test to verify it fails**

```bash
cd digital-avatar-app/backend
python -m pytest tests/test_boxim_client.py -v
```
Expected: FAIL with "ModuleNotFoundError: No module named 'services.boxim_client'"

- [ ] **Step 3: Write minimal implementation**

```python
# digital-avatar-app/backend/services/boxim_client.py
"""盒子 IM 客户端 — 封装网易云信 IM 接口调用"""
import hashlib
import random
import string
from datetime import datetime
from typing import Optional
import httpx


class BoxIMClient:
    """盒子 IM 客户端，通过会会平台网关调用网易云信 IM"""

    def __init__(self, config: dict):
        self.base_url = config.get("HUIHUI_IM_BASE_URL", "http://192.168.1.200:60040")
        self.app_id = config.get("HUIHUI_APP_ID", "")
        self.access_id = config.get("HUIHUI_ACCESS_ID", "")
        self.access_secret = config.get("HUIHUI_ACCESS_SECRET", "")

    def _build_sign_params(self, extra: dict) -> dict:
        """构建带签名的请求参数"""
        nonce = "".join(random.choices(string.ascii_lowercase + string.digits, k=12))
        timestamp = datetime.now().strftime("%Y%m%d%H%M%S")  # 24小时制
        params = {
            "appId": self.app_id,
            "accessId": self.access_id,
            "nonce": nonce,
            "timestamp": timestamp,
            **extra,
        }
        # 计算签名
        keys = sorted(params.keys())
        sign_parts = []
        for k in keys:
            v = params.get(k)
            if v and v != "" and v != []:
                sign_parts.append(f"{k}={v}")
        sign_str = "&".join(sign_parts) + f"&accessSecret={self.access_secret}"
        signature = hashlib.md5(sign_str.encode("utf-8")).hexdigest().upper()
        params["signature"] = signature
        params["signType"] = "MD5"
        params["signVersion"] = "1.0"
        return params

    async def get_credentials(self, user_id: str) -> Optional[dict]:
        """获取用户的网易云信 IM 凭证 (accid, token)"""
        params = self._build_sign_params({"userId": user_id})
        async with httpx.AsyncClient(timeout=10) as client:
            r = await client.post(
                f"{self.base_url}/box/netease",
                params=params,
            )
        data = r.json()
        if data.get("code") in (0, 200):
            return data.get("data", {})
        return None

    async def send_p2p_message(
        self, from_accid: str, to_accid: str, content: str
    ) -> bool:
        """发送单聊消息（文本）"""
        params = self._build_sign_params({
            "from": from_accid,
            "to": to_accid,
            "msgType": "text",
            "content": content,
        })
        async with httpx.AsyncClient(timeout=10) as client:
            r = await client.post(
                f"{self.base_url}/box/message/send/p2p",
                params=params,
            )
        data = r.json()
        return data.get("code") in (0, 200)
```

- [ ] **Step 4: Add test dependencies**

```bash
cd digital-avatar-app/backend
pip install pytest pytest-asyncio httpx
```

- [ ] **Step 5: Run test to verify it passes**

```bash
cd digital-avatar-app/backend
python -m pytest tests/test_boxim_client.py -v
```
Expected: PASS

- [ ] **Step 6: Commit**

```bash
git add digital-avatar-app/backend/services/boxim_client.py digital-avatar-app/backend/tests/test_boxim_client.py
git commit -m "feat: add Box IM client for Netease Yunxin integration"
```

---

### Task 3: 接管服务核心逻辑

**Files:**
- Create: `digital-avatar-app/backend/services/takeover_service.py`
- Test: `digital-avatar-app/backend/tests/test_takeover_service.py`

**Interfaces:**
- `TakeoverService.check_takeover_enabled(owner_huihui_id: str, from_user_id: str) -> Authorization | None`
- `TakeoverService.generate_reply(avatar_id: str, message: str) -> str`
- `TakeoverService.execute_takeover(auth: Authorization, message: dict) -> bool`
- `TakeoverService.process_delayed_queue()` — 处理延迟队列中到期的消息

**Dependencies:**
- `BoxIMClient` from Task 2
- Existing chat API: `POST /api/avatar/{id}/chat`
- Redis for delayed queue
- `Authorization` model from Task 1

- [ ] **Step 1: Write the failing test**

```python
# digital-avatar-app/backend/tests/test_takeover_service.py
import pytest
from unittest.mock import AsyncMock, patch, MagicMock
from services.takeover_service import TakeoverService

@pytest.fixture
def mock_db():
    db = MagicMock()
    return db

@pytest.fixture
def mock_boxim():
    client = AsyncMock()
    client.get_credentials.return_value = {"accid": "owner_acc", "token": "tok"}
    client.send_p2p_message.return_value = True
    return client

@pytest.fixture
def mock_auth():
    auth = MagicMock()
    auth.takeover_enabled = True
    auth.takeover_mode = "immediate"
    auth.takeover_delay_seconds = 30
    auth.avatar_id = "avatar_123"
    auth.target_id = "target_user_123"
    return auth

def test_check_takeover_enabled_returns_auth_when_enabled(mock_db, mock_auth):
    mock_db.query.return_value.filter.return_value.filter.return_value.first.return_value = mock_auth
    service = TakeoverService(mock_db, None)
    result = service.check_takeover_enabled("owner_123", "target_123")
    assert result == mock_auth

def test_check_takeover_enabled_returns_none_when_disabled(mock_db):
    disabled_auth = MagicMock()
    disabled_auth.takeover_enabled = False
    mock_db.query.return_value.filter.return_value.filter.return_value.first.return_value = disabled_auth
    service = TakeoverService(mock_db, None)
    result = service.check_takeover_enabled("owner_123", "target_123")
    assert result is None
```

- [ ] **Step 2: Run test to verify it fails**

```bash
cd digital-avatar-app/backend
python -m pytest tests/test_takeover_service.py -v
```
Expected: FAIL

- [ ] **Step 3: Write minimal implementation**

```python
# digital-avatar-app/backend/services/takeover_service.py
"""分身接管聊天服务 — 消息监听、决策、回复执行"""
import json
import logging
from typing import Optional
from sqlalchemy.orm import Session
import httpx

from models import Authorization, Avatar
from services.boxim_client import BoxIMClient

logger = logging.getLogger(__name__)


class TakeoverService:
    """分身接管服务"""

    def __init__(self, db: Session, boxim_client: BoxIMClient, redis_client=None):
        self.db = db
        self.boxim = boxim_client
        self.redis = redis_client
        self._chat_api_base = "http://localhost:8000/api"  # 分身应用自己的 API

    def check_takeover_enabled(
        self, owner_huihui_id: str, from_user_id: str
    ) -> Optional[Authorization]:
        """检查是否有接管授权"""
        auth = (
            self.db.query(Authorization)
            .filter(Authorization.target_id == from_user_id)
            .filter(Authorization.takeover_enabled == True)
            .first()
        )
        if auth and auth.takeover_enabled:
            return auth
        return None

    async def generate_reply(self, avatar_id: str, message: str) -> str:
        """调用分身聊天接口生成回复"""
        try:
            async with httpx.AsyncClient(timeout=30) as client:
                r = await client.post(
                    f"{self._chat_api_base}/avatar/{avatar_id}/chat",
                    json={"message": message, "history": []},
                )
            data = r.json()
            if data.get("code") in (0, 200):
                return data.get("data", {}).get("answer", "")
            logger.warning(f"分身聊天接口返回异常: {data}")
            return ""
        except Exception as e:
            logger.error(f"调用分身聊天接口失败: {e}")
            return ""

    async def execute_takeover(self, auth: Authorization, message: dict) -> bool:
        """执行接管：生成回复并以主人身份发送"""
        try:
            # 1. 获取主人的 IM 凭证
            owner_huihui_id = auth.owner_id if hasattr(auth, 'owner_id') else ""
            credentials = await self.boxim.get_credentials(owner_huihui_id)
            if not credentials:
                logger.warning(f"无法获取主人 IM 凭证: {owner_huihui_id}")
                return False

            # 2. 生成回复
            reply = await self.generate_reply(auth.avatar_id, message.get("content", ""))
            if not reply:
                logger.warning("分身未生成回复")
                return False

            # 3. 发送消息
            success = await self.boxim.send_p2p_message(
                from_accid=credentials["accid"],
                to_accid=message.get("from_accid", ""),
                content=reply,
            )
            if success:
                logger.info(f"分身接管回复成功: {reply[:50]}...")
            return success
        except Exception as e:
            logger.error(f"执行接管失败: {e}")
            return False

    def enqueue_delayed_message(self, auth: Authorization, message: dict):
        """将消息写入 Redis 延迟队列"""
        if not self.redis:
            logger.warning("Redis 未配置，降级为立即接管")
            return
        key = f"takeover:delayed:{auth.target_id}:{message.get('msg_id', '')}"
        value = json.dumps({
            "avatar_id": auth.avatar_id,
            "from_accid": message.get("from_accid", ""),
            "content": message.get("content", ""),
            "owner_huihui_id": auth.owner_id if hasattr(auth, 'owner_id') else "",
        })
        self.redis.setex(key, auth.takeover_delay_seconds + 10, value)
        logger.info(f"消息写入延迟队列: {key}")

    async def process_delayed_queue(self):
        """处理延迟队列中到期的消息"""
        if not self.redis:
            return
        # Redis 会自动过期，这里不需要主动处理
        # 实际使用时可以用 Redis keyspace notifications 或定期扫描
        pass
```

- [ ] **Step 4: Run test to verify it passes**

```bash
cd digital-avatar-app/backend
python -m pytest tests/test_takeover_service.py -v
```
Expected: PASS

- [ ] **Step 5: Commit**

```bash
git add digital-avatar-app/backend/services/takeover_service.py digital-avatar-app/backend/tests/test_takeover_service.py
git commit -m "feat: add takeover service core logic"
```

---

### Task 4: 接管配置 API

**Files:**
- Create: `digital-avatar-app/backend/routers/takeover.py`
- Modify: `digital-avatar-app/backend/main.py`

**Interfaces:**
- `PUT /api/avatar/{avatar_id}/authorizations/takeover` — 更新接管配置
- 请求体：`{ authorization_id, takeover_enabled, takeover_mode, takeover_delay_seconds }`

**Dependencies:**
- `Authorization` model from Task 1
- `ok()` / `fail()` from `responses.py`

- [ ] **Step 1: Write the failing test**

```python
# digital-avatar-app/backend/tests/test_takeover_api.py
from fastapi.testclient import TestClient
from main import app
from database import SessionLocal, Base, engine
from models import Authorization, Avatar

def setup_test_db():
    Base.metadata.create_all(bind=engine)
    db = SessionLocal()
    avatar = Avatar(name="test", status="active", config={})
    db.add(avatar)
    db.commit()
    auth = Authorization(avatar_id=avatar.id, target_id="user1", target_name="测试用户")
    db.add(auth)
    db.commit()
    return db, auth.id

def test_update_takeover_config():
    db, auth_id = setup_test_db()
    client = TestClient(app)
    response = client.put(
        f"/api/avatar/test_auth/avatar_id/authorizations/takeover",
        json={
            "authorization_id": auth_id,
            "takeover_enabled": True,
            "takeover_mode": "delayed",
            "takeover_delay_seconds": 60,
        },
    )
    assert response.status_code == 200
    data = response.json()
    assert data["code"] == 200
    # 验证数据库已更新
    auth = db.query(Authorization).filter(Authorization.id == auth_id).first()
    assert auth.takeover_enabled == True
    assert auth.takeover_mode == "delayed"
    assert auth.takeover_delay_seconds == 60
```

- [ ] **Step 2: Run test to verify it fails**

```bash
cd digital-avatar-app/backend
python -m pytest tests/test_takeover_api.py -v
```
Expected: FAIL with 404 (endpoint not found)

- [ ] **Step 3: Write minimal implementation**

```python
# digital-avatar-app/backend/routers/takeover.py
"""分身接管配置 API"""
from fastapi import APIRouter, Depends, Body
from sqlalchemy.orm import Session

from database import get_db
from models import Authorization
from responses import ok, fail

router = APIRouter(tags=["分身接管"])


@router.put("/avatar/{avatar_id}/authorizations/takeover")
def update_takeover_config(
    avatar_id: str,
    payload: dict = Body(...),
    db: Session = Depends(get_db),
):
    """更新分身接管配置"""
    auth_id = payload.get("authorization_id")
    if not auth_id:
        return fail("缺少 authorization_id", 400)

    auth = db.query(Authorization).filter(Authorization.id == auth_id).first()
    if not auth:
        return fail("授权不存在", 404)

    # 更新接管配置
    if "takeover_enabled" in payload:
        auth.takeover_enabled = payload["takeover_enabled"]
    if "takeover_mode" in payload:
        mode = payload["takeover_mode"]
        if mode not in ("immediate", "delayed"):
            return fail("takeover_mode 必须是 immediate 或 delayed", 400)
        auth.takeover_mode = mode
    if "takeover_delay_seconds" in payload:
        delay = payload["takeover_delay_seconds"]
        if not isinstance(delay, int) or delay < 5:
            return fail("takeover_delay_seconds 必须 >= 5", 400)
        auth.takeover_delay_seconds = delay

    db.commit()
    return ok(auth.to_dict())
```

- [ ] **Step 4: Register router in main.py**

```python
# digital-avatar-app/backend/main.py
# 在 import 部分添加：
import routers.takeover

# 在 router 注册部分添加：
app.include_router(routers.takeover.router, prefix="/api")
```

- [ ] **Step 5: Run test to verify it passes**

```bash
cd digital-avatar-app/backend
python -m pytest tests/test_takeover_api.py -v
```
Expected: PASS

- [ ] **Step 6: Commit**

```bash
git add digital-avatar-app/backend/routers/takeover.py digital-avatar-app/backend/main.py digital-avatar-app/backend/tests/test_takeover_api.py
git commit -m "feat: add takeover config API endpoint"
```

---

### Task 5: 定时消息监听任务

**Files:**
- Modify: `digital-avatar-app/backend/main.py`
- Modify: `digital-avatar-app/backend/requirements.txt`

**Interfaces:**
- APScheduler 定时任务：每 10 秒轮询一次未读消息
- 调用 `TakeoverService.check_takeover_enabled` 判断是否接管
- 根据模式调用立即执行或延迟队列

**Dependencies:**
- `TakeoverService` from Task 3
- `BoxIMClient` from Task 2
- `APScheduler` for scheduling
- `redis` for delayed queue

- [ ] **Step 1: Add dependencies**

```
# digital-avatar-app/backend/requirements.txt
# 新增：
redis>=5.0
apscheduler>=3.10
```

- [ ] **Step 2: Add scheduler setup in main.py**

```python
# digital-avatar-app/backend/main.py
# 在 import 部分添加：
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.interval import IntervalTrigger
import redis as redis_lib
import os

# 在 on_startup 函数中添加：
@app.on_event("startup")
def on_startup():
    init_db()
    seed()

    # 初始化 Redis（可选）
    redis_client = None
    redis_url = os.getenv("REDIS_URL", "")
    if redis_url:
        try:
            redis_client = redis_lib.from_url(redis_url)
            redis_client.ping()
        except Exception:
            logger.warning("Redis 连接失败，延迟接管将降级为立即接管")

    # 初始化 Box IM 客户端
    boxim_config = {
        "HUIHUI_IM_BASE_URL": os.getenv("HUIHUI_IM_BASE_URL", "http://192.168.1.200:60040"),
        "HUIHUI_APP_ID": os.getenv("HUIHUI_APP_ID", ""),
        "HUIHUI_ACCESS_ID": os.getenv("HUIHUI_ACCESS_ID", ""),
        "HUIHUI_ACCESS_SECRET": os.getenv("HUIHUI_ACCESS_SECRET", ""),
    }
    boxim_client = BoxIMClient(boxim_config)

    # 初始化接管服务
    from services.takeover_service import TakeoverService
    takeover_service = TakeoverService(SessionLocal(), boxim_client, redis_client)

    # 启动定时任务
    scheduler = BackgroundScheduler()
    scheduler.add_job(
        takeover_service.poll_and_process_messages,
        trigger=IntervalTrigger(seconds=10),
        id="takeover_message_poll",
    )
    scheduler.start()
```

- [ ] **Step 3: Add poll method to TakeoverService**

```python
# digital-avatar-app/backend/services/takeover_service.py
# 在 TakeoverService 类中新增：
    async def poll_and_process_messages(self):
        """定时轮询未读消息并处理"""
        # 这里简化处理，实际需要调用会会平台的消息推送接口
        # 伪代码：
        # messages = await self.fetch_unread_messages()
        # for msg in messages:
        #     await self.process_message(msg)
        pass

    async def process_message(self, message: dict):
        """处理单条消息"""
        owner_id = message.get("owner_huihui_id", "")
        from_id = message.get("from_accid", "")

        auth = self.check_takeover_enabled(owner_id, from_id)
        if not auth:
            return

        if auth.takeover_mode == "immediate":
            await self.execute_takeover(auth, message)
        else:
            self.enqueue_delayed_message(auth, message)
```

- [ ] **Step 4: Commit**

```bash
git add digital-avatar-app/backend/main.py digital-avatar-app/backend/requirements.txt digital-avatar-app/backend/services/takeover_service.py
git commit -m "feat: add scheduled message polling for takeover"
```

---

### Task 6: 前端授权管理 UI 增强

**Files:**
- Modify: `digital-avatar-app/src/views/AuthorizationManage.vue`
- Modify: `digital-avatar-app/src/api/index.ts`

**Interfaces:**
- 授权卡片新增接管开关和配置
- 开关开启时显示模式选择和延迟秒数输入
- 调用 `PUT /api/avatar/{id}/authorizations/takeover` 保存配置

**Dependencies:**
- Existing authorization API functions
- Takeover config API from Task 4

- [ ] **Step 1: Add API function**

```typescript
// digital-avatar-app/src/api/index.ts
// 新增：
export const updateTakeoverConfig = (avatarId: string, data: {
  authorizationId: string
  takeoverEnabled: boolean
  takeoverMode?: string
  takeoverDelaySeconds?: number
}) => {
  return api.put(`/api/avatar/${avatarId}/authorizations/takeover`, data)
}
```

- [ ] **Step 2: Update AuthorizationManage.vue**

```vue
<!-- digital-avatar-app/src/views/AuthorizationManage.vue -->
<!-- 在授权卡片的权限标签后添加接管配置 -->
<div v-if="auth.permissions?.includes('takeover') || auth.takeoverEnabled" class="takeover-config">
  <div class="takeover-header">
    <span class="takeover-label">分身接管聊天</span>
    <button class="toggle-btn" :class="{ active: auth.takeoverEnabled }" @click="toggleTakeover(auth)">
      <span class="toggle-dot"></span>
    </button>
  </div>

  <div v-if="auth.takeoverEnabled" class="takeover-options">
    <div class="option-row">
      <label>接管模式</label>
      <select v-model="auth.takeoverMode" @change="saveTakeoverConfig(auth)">
        <option value="immediate">立即接管</option>
        <option value="delayed">延迟接管</option>
      </select>
    </div>

    <div v-if="auth.takeoverMode === 'delayed'" class="option-row">
      <label>延迟秒数</label>
      <input type="number" v-model.number="auth.takeoverDelaySeconds"
             min="5" max="300" @change="saveTakeoverConfig(auth)" />
    </div>
  </div>
</div>
```

- [ ] **Step 3: Add methods to AuthorizationManage.vue**

```vue
<script setup lang="ts">
// 新增 import
import { updateTakeoverConfig } from '@/api'

// 新增方法
const toggleTakeover = async (auth: any) => {
  auth.takeoverEnabled = !auth.takeoverEnabled
  await saveTakeoverConfig(auth)
}

const saveTakeoverConfig = async (auth: any) => {
  try {
    await updateTakeoverConfig(currentAvatarId.value, {
      authorizationId: auth.id,
      takeoverEnabled: auth.takeoverEnabled,
      takeoverMode: auth.takeoverMode || 'immediate',
      takeoverDelaySeconds: auth.takeoverDelaySeconds || 30,
    })
  } catch (e) {
    console.error('保存接管配置失败', e)
  }
}
</script>
```

- [ ] **Step 4: Add styles**

```vue
<style scoped>
.takeover-config {
  margin-top: 12px;
  padding: 10px;
  background: rgba(88, 166, 255, 0.05);
  border-radius: 8px;
}
.takeover-header {
  display: flex;
  justify-content: space-between;
  align-items: center;
  margin-bottom: 8px;
}
.takeover-label {
  font-size: 13px;
  font-weight: 600;
}
.takeover-options {
  display: flex;
  flex-direction: column;
  gap: 8px;
}
.option-row {
  display: flex;
  align-items: center;
  gap: 8px;
  font-size: 12px;
}
.option-row label {
  width: 70px;
  flex-shrink: 0;
  color: var(--color-text-muted);
}
.option-row select, .option-row input {
  flex: 1;
  padding: 4px 8px;
  border: 1px solid var(--color-border);
  border-radius: 4px;
  background: var(--color-bg);
  color: var(--color-text);
}
</style>
```

- [ ] **Step 5: Commit**

```bash
git add digital-avatar-app/src/views/AuthorizationManage.vue digital-avatar-app/src/api/index.ts
git commit -m "feat: add takeover config UI to authorization management"
```

---

### Task 7: 集成测试和部署

**Files:**
- Modify: `docker-compose.yml` (digital-avatar-app)
- Create: `digital-avatar-app/.env.example`

**Interfaces:**
- 添加 Redis 容器（可选，用于延迟队列）
- 添加环境变量配置

- [ ] **Step 1: Add Redis to docker-compose**

```yaml
# digital-avatar-app/docker-compose.yml
services:
  avatar-backend:
    # ... 现有配置 ...
    environment:
      - REDIS_URL=redis://avatar-redis:6379
      - HUIHUI_IM_BASE_URL=http://192.168.1.200:60040
      - HUIHUI_APP_ID=${HUIHUI_APP_ID}
      - HUIHUI_ACCESS_ID=${HUIHUI_ACCESS_ID}
      - HUIHUI_ACCESS_SECRET=${HUIHUI_ACCESS_SECRET}
    depends_on:
      - avatar-redis

  avatar-redis:
    image: redis:7-alpine
    container_name: avatar-redis
    restart: unless-stopped
    volumes:
      - redis_data:/data
    networks:
      - avatar-net
```

- [ ] **Step 2: Create .env.example**

```
# digital-avatar-app/.env.example
HUIHUI_IM_BASE_URL=http://192.168.1.200:60040
HUIHUI_APP_ID=your_app_id
HUIHUI_ACCESS_ID=your_access_id
HUIHUI_ACCESS_SECRET=your_access_secret
REDIS_URL=redis://avatar-redis:6379
```

- [ ] **Step 3: Manual testing checklist**

```markdown
## 测试清单

### 功能测试
- [ ] 授权管理页面显示接管开关
- [ ] 开关开启后显示模式选择
- [ ] 延迟模式显示延迟秒数输入
- [ ] 保存配置后刷新页面配置保留
- [ ] 立即接管：模拟消息到达后分身立即回复
- [ ] 延迟接管：N 秒内主人回复则不分身接管
- [ ] 延迟接管：超时后分身自动回复

### 边界测试
- [ ] 主人有多个分身时使用第一个 active 分身
- [ ] 知识库为空时使用 Qwen 兜底
- [ ] IM 接口异常时记录日志不崩溃
- [ ] Redis 不可用时降级为立即接管

### 性能测试
- [ ] 轮询间隔 10 秒时 CPU 占用正常
- [ ] 延迟队列消息过期自动清理
```

- [ ] **Step 4: Commit**

```bash
git add digital-avatar-app/docker-compose.yml digital-avatar-app/.env.example
git commit -m "feat: add Redis for takeover delayed queue and env config"
```

---

## 执行顺序

1. Task 1 → Task 2 → Task 3 → Task 4 → Task 5 → Task 6 → Task 7
2. 每个 Task 完成后运行测试验证
3. Task 5 和 Task 6 可以并行（后端定时任务和前端 UI 无依赖）

## 风险与缓解

| 风险 | 缓解措施 |
|------|---------|
| 会会平台 IM 接口文档不完整 | 先做接口探索，用 Postman 手动验证 |
| 盒子 IM 替换云信的 API 差异 | 保留云信客户端作为 fallback |
| Redis 延迟队列复杂性 | 第一期可简化为内存队列，Redis 作为优化 |
| 分身聊天接口响应慢 | 设置超时，失败时不阻塞轮询 |