fix: accept camelCase authorizationId from frontend

PUT /api/avatar/{avatar_id}/authorizations/takeover to update authorization
takeover settings (enabled, mode, delay_seconds) with validation.

digital-avatar-app/backend/database.py

@@ -35,6 +35,9 @@ def init_db():
         ("knowledge_docs", "chunk_count", "INTEGER DEFAULT 0"),
         ("knowledge_docs", "vectorized_at", "TIMESTAMP"),
         ("avatars", "owner_id", "VARCHAR DEFAULT ''"),
+        ("authorizations", "takeover_enabled", "BOOLEAN DEFAULT 0"),
+        ("authorizations", "takeover_mode", "VARCHAR DEFAULT 'immediate'"),
+        ("authorizations", "takeover_delay_seconds", "INTEGER DEFAULT 30"),
     )
 
 

digital-avatar-app/backend/main.py

@@ -2,6 +2,11 @@ from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 
 import os
+import logging
+
+from apscheduler.schedulers.background import BackgroundScheduler
+from apscheduler.triggers.interval import IntervalTrigger
+import redis as redis_lib
 
 from database import init_db, SessionLocal
 from models import Avatar, Authorization, Organization, TokenAccount, TokenPlan
@@ -13,8 +18,11 @@ import routers.organizations
 import routers.knowledge
 import routers.huihui_auth
 import routers.chat
+import routers.takeover
 from responses import ok
 
+logger = logging.getLogger(__name__)
+
 app = FastAPI(title="会会数字分身 API", version="1.0.0")
 
 app.add_middleware(
@@ -32,6 +40,7 @@ app.include_router(routers.organizations.router, prefix="/api")
 app.include_router(routers.knowledge.router, prefix="/api")
 app.include_router(routers.huihui_auth.router, prefix="/api")
 app.include_router(routers.chat.router, prefix="/api")
+app.include_router(routers.takeover.router, prefix="/api")
 
 UPLOAD_DIR = routers.knowledge.UPLOAD_DIR
 os.makedirs(UPLOAD_DIR, exist_ok=True)
@@ -103,3 +112,41 @@ def seed():
 def on_startup():
     init_db()
     seed()
+
+    # --- Takeover scheduler ---
+    try:
+        # Initialize Redis (optional)
+        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 as e:
+                logger.warning(f"Redis connection failed, delayed takeover will degrade to immediate: {e}")
+
+        # Initialize Box IM client
+        from services.boxim_client import BoxIMClient
+        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)
+
+        # Initialize takeover service
+        from services.takeover_service import TakeoverService
+        takeover_service = TakeoverService(SessionLocal(), boxim_client, redis_client)
+
+        # Start periodic polling job
+        scheduler = BackgroundScheduler()
+        scheduler.add_job(
+            takeover_service.poll_and_process_messages,
+            trigger=IntervalTrigger(seconds=10),
+            id="takeover_message_poll",
+        )
+        scheduler.start()
+        logger.info("Takeover message polling scheduler started (interval=10s)")
+    except Exception as e:
+        logger.warning(f"Failed to initialize takeover scheduler, app will continue without it: {e}")

digital-avatar-app/backend/models.py

@@ -51,6 +51,9 @@ class Authorization(Base):
     target_name = Column(String, default="")
     permissions = Column(JSON, default=list)
     status = Column(String, default="active")  # active | inactive
+    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):
@@ -62,6 +65,9 @@ class Authorization(Base):
             "targetName": self.target_name,
             "permissions": self.permissions or [],
             "status": self.status,
+            "takeoverEnabled": self.takeover_enabled,
+            "takeoverMode": self.takeover_mode,
+            "takeoverDelaySeconds": self.takeover_delay_seconds,
             "createdAt": _iso(self.created_at),
         }
 

digital-avatar-app/backend/requirements.txt

@@ -7,3 +7,5 @@ httpx
 pypdf
 python-docx
 openpyxl
+redis>=5.0
+apscheduler>=3.10

digital-avatar-app/backend/routers/takeover.py

@@ -0,0 +1,41 @@
+"""分身接管配置 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("authorizationId") or 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())

digital-avatar-app/backend/services/boxim_client.py

@@ -0,0 +1,76 @@
+"""盒子 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:
+        """构建带签名的请求参数（复用 news_service 签名模式）"""
+        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,
+        }
+        # 计算签名 — 排序 key, 过滤空值, 拼接后加 accessSecret, MD5 大写
+        keys = sorted(params.keys())
+        sign_parts = []
+        for k in keys:
+            if k in ("signature", "accessSecret"):
+                continue
+            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)

digital-avatar-app/backend/services/takeover_service.py

@@ -0,0 +1,174 @@
+"""Takeover service — message listening, decision, reply execution."""
+import json
+import logging
+import os
+from typing import Optional
+
+import httpx
+from sqlalchemy.orm import Session
+
+from models import Avatar, Authorization
+from services.boxim_client import BoxIMClient
+
+logger = logging.getLogger(__name__)
+
+
+class TakeoverService:
+    """Service for handling avatar takeover — generating replies and sending them via IM."""
+
+    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 = os.getenv(
+            "TAKEOVER_CHAT_API_BASE", "http://localhost:8000/api"
+        )
+
+    def check_takeover_enabled(
+        self, owner_huihui_id: str, from_user_id: str
+    ) -> Optional[Authorization]:
+        """Check whether takeover is enabled for the given target user."""
+        avatar = self.db.query(Avatar).filter(Avatar.owner_id == owner_huihui_id).first()
+        if not avatar:
+            return None
+
+        auth = (
+            self.db.query(Authorization)
+            .filter(Authorization.avatar_id == avatar.id)
+            .filter(Authorization.target_id == from_user_id)
+            .filter(Authorization.takeover_enabled == True)
+            .first()
+        )
+        return auth if auth and auth.takeover_enabled else None
+
+    async def generate_reply(self, avatar_id: str, message: str) -> str:
+        """Call the avatar chat endpoint to generate a reply."""
+        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"Avatar chat API returned error code: {data}")
+            return ""
+        except Exception as e:
+            logger.error(f"Failed to call avatar chat API: {e}")
+            return ""
+
+    async def execute_takeover(self, auth: Authorization, message: dict) -> bool:
+        """Execute takeover: generate a reply and send it as the owner via IM."""
+        try:
+            # Resolve owner through Avatar model
+            avatar = self.db.query(Avatar).filter(Avatar.id == auth.avatar_id).first()
+            if not avatar:
+                logger.warning(f"Avatar not found: {auth.avatar_id}")
+                return False
+
+            owner_huihui_id = avatar.owner_id
+            credentials = await self.boxim.get_credentials(owner_huihui_id)
+            if not credentials:
+                logger.warning(f"Cannot obtain IM credentials for owner: {owner_huihui_id}")
+                return False
+
+            reply = await self.generate_reply(auth.avatar_id, message.get("content", ""))
+            if not reply:
+                logger.warning("Avatar did not generate a reply")
+                return False
+
+            success = await self.boxim.send_p2p_message(
+                from_accid=credentials["accid"],
+                to_accid=message.get("from_accid", ""),
+                content=reply,
+            )
+            if success:
+                logger.info(f"Takeover reply sent successfully: {reply[:50]}...")
+            return success
+        except Exception as e:
+            logger.error(f"Takeover execution failed: {e}")
+            return False
+
+    def enqueue_delayed_message(self, auth: Authorization, message: dict):
+        """Write a message into the Redis delayed queue (TTL = delay + 10s buffer)."""
+        if not self.redis:
+            logger.warning("Redis not configured, degrading to immediate takeover")
+            return
+
+        avatar = self.db.query(Avatar).filter(Avatar.id == auth.avatar_id).first()
+        owner_huihui_id = avatar.owner_id if avatar else ""
+        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": owner_huihui_id,
+        })
+        self.redis.setex(key, auth.takeover_delay_seconds + 10, value)
+        logger.info(f"Message enqueued to delayed queue: {key}")
+
+    async def process_delayed_queue(self):
+        """Process expired messages from the delayed queue.
+
+        Scans Redis keys matching the takeover:delayed: pattern and dispatches
+        each to execute_takeover after resolving the Authorization.
+        """
+        if not self.redis:
+            return
+        try:
+            pattern = "takeover:delayed:*"
+            keys = self.redis.keys(pattern)
+            for key in keys:
+                raw = self.redis.get(key)
+                if not raw:
+                    continue
+                data = json.loads(raw)
+                auth = (
+                    self.db.query(Authorization)
+                    .filter(Authorization.target_id == key.split(":")[2])
+                    .first()
+                )
+                if auth:
+                    message = {
+                        "msg_id": key.split(":")[-1],
+                        "from_accid": data.get("from_accid", ""),
+                        "content": data.get("content", ""),
+                    }
+                    await self.execute_takeover(auth, message)
+                self.redis.delete(key)
+        except Exception as e:
+            logger.error(f"Failed to process delayed queue: {e}")
+
+    async def poll_and_process_messages(self):
+        """Periodic polling job: fetch unread messages and process each."""
+        try:
+            messages = await self.fetch_unread_messages()
+            for msg in messages:
+                await self.process_message(msg)
+        except Exception as e:
+            logger.error(f"poll_and_process_messages failed: {e}")
+
+    async def fetch_unread_messages(self) -> list:
+        """Fetch unread messages from Box IM. Stub — replace with real API call."""
+        logger.debug("fetch_unread_messages: no real API wired yet")
+        return []
+
+    async def process_message(self, message: dict):
+        """Process a single message: check takeover, dispatch immediate or delayed."""
+        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)

digital-avatar-app/backend/tests/conftest.py

@@ -0,0 +1,26 @@
+import pytest
+from database import init_db, SessionLocal
+from models import Authorization
+
+
+@pytest.fixture(scope="session", autouse=True)
+def setup_database():
+    """Initialize DB tables and seed an Authorization row."""
+    init_db()
+    db = SessionLocal()
+    try:
+        existing = db.query(Authorization).first()
+        if existing is None:
+            auth = Authorization(
+                id="test-auth-1",
+                avatar_id="test-avatar-1",
+                target_type="user",
+                target_id="test-user-1",
+                target_name="Test User",
+                permissions=["read", "write"],
+                status="active",
+            )
+            db.add(auth)
+            db.commit()
+    finally:
+        db.close()

digital-avatar-app/backend/tests/test_boxim_client.py

@@ -0,0 +1,134 @@
+"""Tests for the Box IM client (Netease Yunxin gateway wrapper)."""
+import pytest
+from unittest.mock import AsyncMock, MagicMock, patch
+
+
+@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",
+    }
+
+
+def _make_mock_response(json_data: dict):
+    """Create a properly configured mock for httpx.Response."""
+    mock_response = MagicMock()
+    mock_response.json.return_value = json_data
+    return mock_response
+
+
+def _patch_httpx_client(json_data: dict):
+    """Patch httpx.AsyncClient so that `async with httpx.AsyncClient() as c: await c.post(...)` returns json_data."""
+    mock_client = AsyncMock()
+    mock_client.post.return_value = _make_mock_response(json_data)
+
+    mock_cm = AsyncMock()
+    mock_cm.__aenter__.return_value = mock_client
+    mock_cm.__aexit__.return_value = None
+
+    return patch("httpx.AsyncClient", return_value=mock_cm)
+
+
+@pytest.mark.asyncio
+async def test_get_credentials(mock_config):
+    """get_credentials should return accid and token from the gateway response."""
+    with _patch_httpx_client({"code": 200, "data": {"accid": "user123", "token": "tok_xyz"}}):
+        from services.boxim_client import BoxIMClient
+
+        client = BoxIMClient(mock_config)
+        result = await client.get_credentials("user123")
+
+        assert result["accid"] == "user123"
+        assert result["token"] == "tok_xyz"
+
+
+@pytest.mark.asyncio
+async def test_send_p2p_message_success(mock_config):
+    """send_p2p_message should return True when the gateway responds with code 200."""
+    with _patch_httpx_client({"code": 200}):
+        from services.boxim_client import BoxIMClient
+
+        client = BoxIMClient(mock_config)
+        result = await client.send_p2p_message("owner_acc", "target_acc", "Hello")
+
+        assert result is True
+
+
+@pytest.mark.asyncio
+async def test_send_p2p_message_failure(mock_config):
+    """send_p2p_message should return False when the gateway responds with a non-200 code."""
+    with _patch_httpx_client({"code": 500, "message": "error"}):
+        from services.boxim_client import BoxIMClient
+
+        client = BoxIMClient(mock_config)
+        result = await client.send_p2p_message("owner_acc", "target_acc", "Hello")
+
+        assert result is False
+
+
+@pytest.mark.asyncio
+async def test_get_credentials_returns_none_on_error(mock_config):
+    """get_credentials should return None when the gateway responds with an error code."""
+    with _patch_httpx_client({"code": 500, "message": "user not found"}):
+        from services.boxim_client import BoxIMClient
+
+        client = BoxIMClient(mock_config)
+        result = await client.get_credentials("nonexistent")
+
+        assert result is None
+
+
+def test_build_sign_params_contains_required_fields(mock_config):
+    """_build_sign_params should produce appId, accessId, nonce, timestamp, signature, signType, signVersion."""
+    from services.boxim_client import BoxIMClient
+
+    client = BoxIMClient(mock_config)
+    params = client._build_sign_params({"userId": "u1"})
+
+    assert "appId" in params
+    assert "accessId" in params
+    assert "nonce" in params
+    assert "timestamp" in params
+    assert "signature" in params
+    assert params["signType"] == "MD5"
+    assert params["signVersion"] == "1.0"
+    assert len(params["nonce"]) == 12
+
+
+def test_build_sign_params_excludes_signature_and_accessSecret_from_signing_string(mock_config):
+    """signature and accessSecret must be excluded from the signing string to match news_service.py."""
+    from services.boxim_client import BoxIMClient
+
+    client = BoxIMClient(mock_config)
+
+    # Pass params that already contain a stale "signature" value
+    params_with_stale_sig = client._build_sign_params({
+        "userId": "u1",
+        "signature": "OLD_STALE_SIG",
+    })
+
+    # The returned signature must be freshly computed (32-char MD5 uppercase),
+    # NOT the stale value we passed in.
+    assert params_with_stale_sig["signature"] != "OLD_STALE_SIG"
+    assert len(params_with_stale_sig["signature"]) == 32
+
+    # Calling with the same extra params but no stale signature should also work.
+    params_clean = client._build_sign_params({"userId": "u1"})
+    assert len(params_clean["signature"]) == 32
+
+
+def test_build_sign_params_signature_is_deterministic(mock_config):
+    """Same inputs should produce valid MD5 signatures."""
+    from services.boxim_client import BoxIMClient
+
+    client = BoxIMClient(mock_config)
+
+    params1 = client._build_sign_params({"userId": "u1"})
+    params2 = client._build_sign_params({"userId": "u1"})
+
+    assert params1["signature"] is not None
+    assert params2["signature"] is not None
+    assert len(params1["signature"]) == 32  # MD5 hex length

digital-avatar-app/backend/tests/test_takeover_api.py

@@ -0,0 +1,105 @@
+"""Tests for PUT /api/avatar/{avatar_id}/authorizations/takeover endpoint."""
+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()
+    db.refresh(avatar)
+    auth = Authorization(avatar_id=avatar.id, target_id="user1", target_name="测试用户")
+    db.add(auth)
+    db.commit()
+    db.refresh(auth)
+    return db, auth.id
+
+
+def test_update_takeover_config():
+    db, auth_id = setup_test_db()
+    try:
+        client = TestClient(app)
+        response = client.put(
+            f"/api/avatar/test_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
+        assert data["data"]["takeoverEnabled"] is True
+        assert data["data"]["takeoverMode"] == "delayed"
+        assert data["data"]["takeoverDelaySeconds"] == 60
+        # 验证数据库已更新
+        auth = db.query(Authorization).filter(Authorization.id == auth_id).first()
+        assert auth.takeover_enabled is True
+        assert auth.takeover_mode == "delayed"
+        assert auth.takeover_delay_seconds == 60
+    finally:
+        db.close()
+
+
+def test_update_takeover_invalid_mode():
+    db, auth_id = setup_test_db()
+    try:
+        client = TestClient(app)
+        response = client.put(
+            f"/api/avatar/test/authorizations/takeover",
+            json={
+                "authorization_id": auth_id,
+                "takeover_mode": "invalid_mode",
+            },
+        )
+        assert response.status_code == 200
+        data = response.json()
+        assert data["code"] == 400
+    finally:
+        db.close()
+
+
+def test_update_takeover_invalid_delay():
+    db, auth_id = setup_test_db()
+    try:
+        client = TestClient(app)
+        response = client.put(
+            f"/api/avatar/test/authorizations/takeover",
+            json={
+                "authorization_id": auth_id,
+                "takeover_delay_seconds": 2,
+            },
+        )
+        assert response.status_code == 200
+        data = response.json()
+        assert data["code"] == 400
+    finally:
+        db.close()
+
+
+def test_update_takeover_missing_auth_id():
+    client = TestClient(app)
+    response = client.put(
+        f"/api/avatar/test/authorizations/takeover",
+        json={"takeover_enabled": True},
+    )
+    assert response.status_code == 200
+    data = response.json()
+    assert data["code"] == 400
+
+
+def test_update_takeover_not_found():
+    client = TestClient(app)
+    response = client.put(
+        f"/api/avatar/test/authorizations/takeover",
+        json={"authorization_id": "nonexistent"},
+    )
+    assert response.status_code == 200
+    data = response.json()
+    assert data["code"] == 404

digital-avatar-app/backend/tests/test_takeover_model.py

@@ -0,0 +1,30 @@
+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
+        # Check new fields exist and have default values
+        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 'takeoverEnabled' in d
+        assert 'takeoverMode' in d
+        assert 'takeoverDelaySeconds' in d
+    finally:
+        db.close()

digital-avatar-app/backend/tests/test_takeover_scheduler.py

@@ -0,0 +1,197 @@
+"""Tests for the scheduled takeover message polling."""
+import json
+import pytest
+from unittest.mock import MagicMock, patch, AsyncMock
+
+
+def test_app_has_startup_event():
+    """Verify the app has a startup event configured."""
+    from main import app
+    startup_handlers = [handler for handler in app.router.on_startup]
+    assert len(startup_handlers) > 0
+
+
+@patch("services.takeover_service.TakeoverService")
+@patch("services.boxim_client.BoxIMClient")
+@patch("main.redis_lib.from_url")
+@patch("main.BackgroundScheduler")
+def test_scheduler_initialized_with_redis(mock_scheduler_class, mock_redis_from_url, mock_boxim_cls, mock_takeover_cls):
+    """Verify scheduler is initialized when Redis is available."""
+    mock_redis = MagicMock()
+    mock_redis.ping.return_value = None
+    mock_redis_from_url.return_value = mock_redis
+
+    mock_boxim = MagicMock()
+    mock_boxim_cls.return_value = mock_boxim
+
+    mock_takeover = MagicMock()
+    mock_takeover_cls.return_value = mock_takeover
+
+    with patch("main.init_db"), patch("main.seed"), patch.dict("os.environ", {"REDIS_URL": "redis://localhost:6379"}):
+        from main import on_startup
+        on_startup()
+
+    mock_scheduler_class.return_value.add_job.assert_called_once()
+    call_kwargs = mock_scheduler_class.return_value.add_job.call_args[1]
+    assert call_kwargs["id"] == "takeover_message_poll"
+
+
+@patch("services.takeover_service.TakeoverService")
+@patch("services.boxim_client.BoxIMClient")
+@patch("main.BackgroundScheduler")
+def test_scheduler_starts_without_redis(mock_scheduler_class, mock_boxim_cls, mock_takeover_cls):
+    """App should start even when REDIS_URL is not set."""
+    mock_boxim = MagicMock()
+    mock_boxim_cls.return_value = mock_boxim
+
+    mock_takeover = MagicMock()
+    mock_takeover_cls.return_value = mock_takeover
+
+    with patch("main.init_db"), patch("main.seed"), patch.dict("os.environ", {"REDIS_URL": ""}, clear=False):
+        from main import on_startup
+        on_startup()
+
+    mock_scheduler_class.return_value.add_job.assert_called_once()
+
+
+@patch("services.takeover_service.TakeoverService")
+@patch("services.boxim_client.BoxIMClient")
+@patch("main.redis_lib.from_url")
+@patch("main.BackgroundScheduler")
+def test_scheduler_starts_when_redis_fails(mock_scheduler_class, mock_redis_from_url, mock_boxim_cls, mock_takeover_cls):
+    """App should start even when Redis ping fails."""
+    mock_redis_from_url.side_effect = ConnectionError("Connection refused")
+
+    mock_boxim = MagicMock()
+    mock_boxim_cls.return_value = mock_boxim
+
+    mock_takeover = MagicMock()
+    mock_takeover_cls.return_value = mock_takeover
+
+    with patch("main.init_db"), patch("main.seed"), patch.dict("os.environ", {"REDIS_URL": "redis://badhost:6379"}):
+        from main import on_startup
+        on_startup()
+
+    mock_scheduler_class.return_value.add_job.assert_called_once()
+
+
+@patch("main.BackgroundScheduler")
+def test_scheduler_fails_gracefully(mock_scheduler_class):
+    """If scheduler init raises, the app should still start (exception caught)."""
+    mock_scheduler_class.side_effect = RuntimeError("Scheduler crash")
+
+    with patch("main.init_db"), patch("main.seed"):
+        from main import on_startup
+        on_startup()
+
+    # No exception should propagate
+
+
+# --- poll_and_process_messages ---
+
+
+@pytest.fixture
+def mock_db():
+    return MagicMock()
+
+
+@pytest.fixture
+def mock_boxim():
+    return AsyncMock()
+
+
+@pytest.mark.asyncio
+async def test_poll_and_process_messages_calls_fetch_and_process(mock_db, mock_boxim):
+    """poll_and_process_messages should fetch messages and process each."""
+    from services.takeover_service import TakeoverService
+
+    service = TakeoverService(mock_db, mock_boxim)
+    service.fetch_unread_messages = AsyncMock(return_value=[
+        {"owner_huihui_id": "owner_1", "from_accid": "user_1", "content": "hi"},
+        {"owner_huihui_id": "owner_2", "from_accid": "user_2", "content": "hello"},
+    ])
+    service.process_message = AsyncMock()
+
+    await service.poll_and_process_messages()
+
+    service.fetch_unread_messages.assert_awaited_once()
+    assert service.process_message.await_count == 2
+
+
+@pytest.mark.asyncio
+async def test_poll_and_process_messages_handles_errors(mock_db, mock_boxim):
+    """poll_and_process_messages should not crash on fetch failure."""
+    from services.takeover_service import TakeoverService
+
+    service = TakeoverService(mock_db, mock_boxim)
+    service.fetch_unread_messages = AsyncMock(side_effect=ConnectionError("Box IM down"))
+
+    await service.poll_and_process_messages()
+    # No exception should propagate
+
+
+# --- process_message ---
+
+
+@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
+
+
+@pytest.mark.asyncio
+async def test_process_message_immediate_mode(mock_db, mock_boxim, mock_auth):
+    """When takeover_mode is 'immediate', execute_takeover should be called."""
+    from services.takeover_service import TakeoverService
+
+    service = TakeoverService(mock_db, mock_boxim)
+    service.check_takeover_enabled = MagicMock(return_value=mock_auth)
+    service.execute_takeover = AsyncMock(return_value=True)
+    service.enqueue_delayed_message = MagicMock()
+
+    message = {"owner_huihui_id": "owner_1", "from_accid": "user_1", "content": "hi"}
+    await service.process_message(message)
+
+    service.execute_takeover.assert_awaited_once_with(mock_auth, message)
+    service.enqueue_delayed_message.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_process_message_delayed_mode(mock_db, mock_boxim, mock_auth):
+    """When takeover_mode is not 'immediate', message should be enqueued."""
+    from services.takeover_service import TakeoverService
+
+    mock_auth.takeover_mode = "delayed"
+
+    service = TakeoverService(mock_db, mock_boxim)
+    service.check_takeover_enabled = MagicMock(return_value=mock_auth)
+    service.execute_takeover = AsyncMock()
+    service.enqueue_delayed_message = MagicMock()
+
+    message = {"owner_huihui_id": "owner_1", "from_accid": "user_1", "content": "hi"}
+    await service.process_message(message)
+
+    service.enqueue_delayed_message.assert_called_once_with(mock_auth, message)
+    service.execute_takeover.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_process_message_no_takeover(mock_db, mock_boxim):
+    """When takeover is not enabled, nothing should happen."""
+    from services.takeover_service import TakeoverService
+
+    service = TakeoverService(mock_db, mock_boxim)
+    service.check_takeover_enabled = MagicMock(return_value=None)
+    service.execute_takeover = AsyncMock()
+    service.enqueue_delayed_message = MagicMock()
+
+    message = {"owner_huihui_id": "owner_1", "from_accid": "user_1", "content": "hi"}
+    await service.process_message(message)
+
+    service.execute_takeover.assert_not_awaited()
+    service.enqueue_delayed_message.assert_not_called()

digital-avatar-app/backend/tests/test_takeover_service.py

@@ -0,0 +1,318 @@
+"""Tests for the TakeoverService — message listening, decision, reply execution."""
+import pytest
+from unittest.mock import AsyncMock, patch, MagicMock
+from services.takeover_service import TakeoverService
+from models import Authorization, Avatar
+
+
+@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(spec=Authorization)
+    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
+
+
+@pytest.fixture
+def mock_avatar():
+    avatar = MagicMock(spec=Avatar)
+    avatar.id = "avatar_123"
+    avatar.owner_id = "owner_huihui_123"
+    return avatar
+
+
+# --- check_takeover_enabled ---
+
+
+def test_check_takeover_enabled_returns_auth_when_enabled(mock_db, mock_auth, mock_boxim, mock_avatar):
+    avatar_filter = MagicMock()
+    avatar_filter.first.return_value = mock_avatar
+    avatar_query = MagicMock()
+    avatar_query.filter.return_value = avatar_filter
+
+    auth_filter = MagicMock()
+    auth_filter.filter.return_value = auth_filter
+    auth_filter.first.return_value = mock_auth
+
+    def query_side_effect(model):
+        if model == Avatar:
+            return avatar_query
+        return auth_filter
+
+    mock_db.query.side_effect = query_side_effect
+
+    service = TakeoverService(mock_db, mock_boxim)
+    result = service.check_takeover_enabled("owner_huihui_123", "target_user_123")
+    assert result == mock_auth
+
+
+def test_check_takeover_enabled_returns_none_when_no_avatar(mock_db, mock_boxim):
+    avatar_filter = MagicMock()
+    avatar_filter.first.return_value = None
+    avatar_query = MagicMock()
+    avatar_query.filter.return_value = avatar_filter
+    mock_db.query.return_value = avatar_query
+
+    service = TakeoverService(mock_db, mock_boxim)
+    result = service.check_takeover_enabled("owner_123", "target_123")
+    assert result is None
+
+
+def test_check_takeover_enabled_returns_none_when_disabled(mock_db, mock_boxim, mock_avatar):
+    avatar_filter = MagicMock()
+    avatar_filter.first.return_value = mock_avatar
+    avatar_query = MagicMock()
+    avatar_query.filter.return_value = avatar_filter
+
+    disabled_auth = MagicMock(spec=Authorization)
+    disabled_auth.takeover_enabled = False
+    auth_filter = MagicMock()
+    auth_filter.filter.return_value = auth_filter
+    auth_filter.first.return_value = disabled_auth
+
+    def query_side_effect(model):
+        if model == Avatar:
+            return avatar_query
+        return auth_filter
+
+    mock_db.query.side_effect = query_side_effect
+
+    service = TakeoverService(mock_db, mock_boxim)
+    result = service.check_takeover_enabled("owner_123", "target_123")
+    assert result is None
+
+
+def test_check_takeover_enabled_filters_by_owner_and_target(mock_db, mock_boxim, mock_avatar, mock_auth):
+    """Verify that queries use the correct filter arguments."""
+    avatar_filter = MagicMock()
+    avatar_filter.first.return_value = mock_avatar
+    avatar_query = MagicMock()
+    avatar_query.filter.return_value = avatar_filter
+
+    auth_filter = MagicMock()
+    auth_filter.filter.return_value = auth_filter
+    auth_filter.first.return_value = mock_auth
+
+    call_order = []
+
+    def query_side_effect(model):
+        if model == Avatar:
+            call_order.append("Avatar")
+            return avatar_query
+        call_order.append("Authorization")
+        return auth_filter
+
+    mock_db.query.side_effect = query_side_effect
+
+    service = TakeoverService(mock_db, mock_boxim)
+    service.check_takeover_enabled("owner_huihui_123", "target_user_123")
+
+    assert "Avatar" in call_order
+    assert "Authorization" in call_order
+
+
+# --- generate_reply ---
+
+
+@pytest.mark.asyncio
+async def test_generate_reply_returns_answer(mock_boxim):
+    mock_db = MagicMock()
+    with patch("services.takeover_service.httpx.AsyncClient") as mock_client_class:
+        mock_response = MagicMock()
+        mock_response.json.return_value = {"code": 200, "data": {"answer": "Hello back"}}
+        mock_client_class.return_value.__aenter__.return_value.post.return_value = mock_response
+
+        service = TakeoverService(mock_db, mock_boxim)
+        result = await service.generate_reply("avatar_123", "Hello")
+        assert result == "Hello back"
+
+
+@pytest.mark.asyncio
+async def test_generate_reply_handles_empty_answer(mock_boxim):
+    """generate_reply should return empty string when answer is missing."""
+    mock_db = MagicMock()
+    with patch("services.takeover_service.httpx.AsyncClient") as mock_client_class:
+        mock_response = MagicMock()
+        mock_response.json.return_value = {"code": 200, "data": {}}
+        mock_client_class.return_value.__aenter__.return_value.post.return_value = mock_response
+
+        service = TakeoverService(mock_db, mock_boxim)
+        result = await service.generate_reply("avatar_123", "Hello")
+        assert result == ""
+
+
+@pytest.mark.asyncio
+async def test_generate_reply_handles_error_code(mock_boxim):
+    """generate_reply should return empty string when API returns error code."""
+    mock_db = MagicMock()
+    with patch("services.takeover_service.httpx.AsyncClient") as mock_client_class:
+        mock_response = MagicMock()
+        mock_response.json.return_value = {"code": 500, "message": "Internal error"}
+        mock_client_class.return_value.__aenter__.return_value.post.return_value = mock_response
+
+        service = TakeoverService(mock_db, mock_boxim)
+        result = await service.generate_reply("avatar_123", "Hello")
+        assert result == ""
+
+
+# --- execute_takeover ---
+
+
+@pytest.mark.asyncio
+async def test_execute_takeover_success(mock_db, mock_boxim, mock_auth, mock_avatar):
+    avatar_filter = MagicMock()
+    avatar_filter.first.return_value = mock_avatar
+    avatar_query = MagicMock()
+    avatar_query.filter.return_value = avatar_filter
+    mock_db.query.return_value = avatar_query
+
+    with patch("services.takeover_service.httpx.AsyncClient") as mock_client_class:
+        mock_response = MagicMock()
+        mock_response.json.return_value = {"code": 200, "data": {"answer": "Hello back"}}
+        mock_client_class.return_value.__aenter__.return_value.post.return_value = mock_response
+
+        service = TakeoverService(mock_db, mock_boxim)
+        message = {"from_accid": "user_acc", "content": "Hello"}
+
+        result = await service.execute_takeover(mock_auth, message)
+
+        assert result is True
+        mock_boxim.get_credentials.assert_called_once_with("owner_huihui_123")
+        mock_boxim.send_p2p_message.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_execute_takeover_fails_when_avatar_not_found(mock_db, mock_boxim, mock_auth):
+    """execute_takeover should return False when Avatar is not found."""
+    avatar_filter = MagicMock()
+    avatar_filter.first.return_value = None
+    avatar_query = MagicMock()
+    avatar_query.filter.return_value = avatar_filter
+    mock_db.query.return_value = avatar_query
+
+    service = TakeoverService(mock_db, mock_boxim)
+    message = {"from_accid": "user_acc", "content": "Hello"}
+
+    result = await service.execute_takeover(mock_auth, message)
+
+    assert result is False
+    mock_boxim.get_credentials.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_execute_takeover_fails_when_no_credentials(mock_db, mock_boxim, mock_auth, mock_avatar):
+    """execute_takeover should return False when boxim.get_credentials returns None."""
+    avatar_filter = MagicMock()
+    avatar_filter.first.return_value = mock_avatar
+    avatar_query = MagicMock()
+    avatar_query.filter.return_value = avatar_filter
+    mock_db.query.return_value = avatar_query
+
+    mock_boxim.get_credentials.return_value = None
+    service = TakeoverService(mock_db, mock_boxim)
+    message = {"from_accid": "user_acc", "content": "Hello"}
+
+    result = await service.execute_takeover(mock_auth, message)
+
+    assert result is False
+
+
+# --- enqueue_delayed_message ---
+
+
+def test_enqueue_delayed_message_with_redis(mock_db, mock_boxim, mock_auth, mock_avatar):
+    mock_redis = MagicMock()
+    avatar_filter = MagicMock()
+    avatar_filter.first.return_value = mock_avatar
+    avatar_query = MagicMock()
+    avatar_query.filter.return_value = avatar_filter
+    mock_db.query.return_value = avatar_query
+
+    service = TakeoverService(mock_db, mock_boxim, mock_redis)
+    message = {"msg_id": "msg_1", "from_accid": "user_acc", "content": "Hello"}
+
+    service.enqueue_delayed_message(mock_auth, message)
+
+    mock_redis.setex.assert_called_once()
+    call_args = mock_redis.setex.call_args
+    value = call_args[0][1]
+    import json
+    payload = json.loads(call_args[0][2])
+    assert payload["owner_huihui_id"] == "owner_huihui_123"
+
+
+def test_enqueue_delayed_message_without_redis_logs_warning(mock_db, mock_boxim, mock_auth, mock_avatar):
+    """When Redis is not configured, enqueue_delayed_message should log a warning and not crash."""
+    avatar_filter = MagicMock()
+    avatar_filter.first.return_value = mock_avatar
+    avatar_query = MagicMock()
+    avatar_query.filter.return_value = avatar_filter
+    mock_db.query.return_value = avatar_query
+
+    service = TakeoverService(mock_db, mock_boxim)
+    message = {"msg_id": "msg_1", "from_accid": "user_acc", "content": "Hello"}
+
+    service.enqueue_delayed_message(mock_auth, message)
+
+
+# --- process_delayed_queue ---
+
+
+@pytest.mark.asyncio
+async def test_process_delayed_queue_no_redis(mock_db, mock_boxim):
+    """process_delayed_queue should return immediately without Redis."""
+    service = TakeoverService(mock_db, mock_boxim)
+    await service.process_delayed_queue()
+    mock_db.query.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_process_delayed_queue_processes_messages(mock_db, mock_boxim, mock_auth, mock_avatar):
+    """process_delayed_queue should read from Redis, resolve auth, and execute takeover."""
+    mock_redis = MagicMock()
+    mock_redis.keys.return_value = ["takeover:delayed:target_user_123:msg_1"]
+    mock_redis.get.return_value = '{"from_accid": "user_acc", "content": "Hello"}'
+
+    avatar_filter = MagicMock()
+    avatar_filter.first.return_value = mock_avatar
+    avatar_query = MagicMock()
+    avatar_query.filter.return_value = avatar_filter
+
+    auth_filter = MagicMock()
+    auth_filter.filter.return_value = auth_filter
+    auth_filter.first.return_value = mock_auth
+
+    def query_side_effect(model):
+        if model == Avatar:
+            return avatar_query
+        return auth_filter
+
+    mock_db.query.side_effect = query_side_effect
+
+    with patch("services.takeover_service.httpx.AsyncClient") as mock_client_class:
+        mock_response = MagicMock()
+        mock_response.json.return_value = {"code": 200, "data": {"answer": "Hello back"}}
+        mock_client_class.return_value.__aenter__.return_value.post.return_value = mock_response
+
+        service = TakeoverService(mock_db, mock_boxim, mock_redis)
+        await service.process_delayed_queue()
+
+        mock_boxim.send_p2p_message.assert_called_once()
+        mock_redis.delete.assert_called_once()

digital-avatar-app/src/api/index.ts

@@ -299,4 +299,13 @@ export const getCurrentUser = () =>
 export const logoutUser = () =>
   request.post('/huihui/logout')
 
+// 更新接管配置
+export const updateTakeoverConfig = (avatarId: string, data: {
+  authorizationId: string
+  takeoverEnabled: boolean
+  takeoverMode?: 'immediate' | 'delayed'
+  takeoverDelaySeconds?: number
+}) =>
+  request.put(`/avatar/${avatarId}/authorizations/takeover`, data)
+
 export default request

digital-avatar-app/src/views/AuthorizationManage.vue

@@ -30,6 +30,46 @@
             {{ auth.status === 'active' ? '撤销' : '授权' }}
           </button>
         </div>
+        <!-- 接管配置 -->
+        <div class="takeover-section">
+          <button class="takeover-toggle-btn" @click="toggleTakeoverEdit(auth)">
+            {{ editingTakeover[auth.id] ? '收起配置' : '配置接管' }}
+          </button>
+          <div class="takeover-form" v-if="editingTakeover[auth.id]">
+            <div class="takeover-row">
+              <label class="takeover-label">启用接管</label>
+              <label class="toggle-switch">
+                <input type="checkbox" v-model="auth.takeoverEnabled" />
+                <span class="toggle-slider"></span>
+              </label>
+            </div>
+            <div class="takeover-row">
+              <label class="takeover-label">接管模式</label>
+              <select class="takeover-select" v-model="auth.takeoverMode" :disabled="!auth.takeoverEnabled">
+                <option value="immediate">立即接管</option>
+                <option value="delayed">延迟接管</option>
+              </select>
+            </div>
+            <div class="takeover-row" v-if="auth.takeoverMode === 'delayed' && auth.takeoverEnabled">
+              <label class="takeover-label">延迟时间（秒）</label>
+              <input
+                type="number"
+                class="takeover-input"
+                v-model.number="auth.takeoverDelaySeconds"
+                min="5"
+                placeholder="30"
+              />
+            </div>
+            <div class="takeover-actions">
+              <button class="takeover-cancel-btn" @click="cancelTakeoverEdit(auth)">取消</button>
+              <button class="takeover-save-btn" @click="saveTakeoverConfig(auth)">保存</button>
+            </div>
+          </div>
+          <div class="takeover-summary" v-else-if="auth.takeoverEnabled">
+            <span class="takeover-badge enabled">已启用</span>
+            <span class="takeover-mode-text">{{ auth.takeoverMode === 'immediate' ? '立即接管' : '延迟 ' + (auth.takeoverDelaySeconds || 30) + 's' }}</span>
+          </div>
+        </div>
       </div>
     </section>
 
@@ -47,7 +87,7 @@
 import { ref, onMounted } from 'vue'
 import { useRouter } from 'vue-router'
 import { useAvatarStore } from '@/store/avatar'
-import { getAuthorizationList, updateAuthorization } from '@/api'
+import { getAuthorizationList, updateAuthorization, updateTakeoverConfig } from '@/api'
 import { pickAvatarId, unwrapListData } from '@/utils/avatar-page-data.js'
 
 const router = useRouter()
@@ -60,8 +100,14 @@ const authList = ref<Array<{
   targetName: string
   permissions: string[]
   status: 'active' | 'inactive'
+  takeoverEnabled?: boolean
+  takeoverMode?: 'immediate' | 'delayed'
+  takeoverDelaySeconds?: number
 }>>([])
 
+// 接管配置编辑状态（每行一个）
+const editingTakeover = ref<Record<string, boolean>>({})
+
 // 从后端加载授权列表
 const loadAuth = async () => {
   try {
@@ -137,6 +183,38 @@ const goBack = () => {
   router.back()
 }
 
+// 切换接管配置编辑状态
+const toggleTakeoverEdit = (auth: any) => {
+  editingTakeover.value[auth.id] = !editingTakeover.value[auth.id]
+}
+
+// 保存接管配置
+const saveTakeoverConfig = async (auth: any) => {
+  const avatarId = pickAvatarId(avatarStore.currentAvatarId, avatarStore.avatars)
+  if (!avatarId) return
+
+  const delay = auth.takeoverMode === 'delayed' ? (auth.takeoverDelaySeconds ?? 30) : undefined
+
+  try {
+    await updateTakeoverConfig(avatarId, {
+      authorizationId: auth.id,
+      takeoverEnabled: auth.takeoverEnabled ?? false,
+      takeoverMode: auth.takeoverMode || 'immediate',
+      takeoverDelaySeconds: delay,
+    })
+    editingTakeover.value[auth.id] = false
+  } catch (e: any) {
+    alert('保存接管配置失败：' + (e?.message || '请重试'))
+  }
+}
+
+// 取消编辑接管配置
+const cancelTakeoverEdit = (auth: any) => {
+  // 重新加载以恢复原始值
+  loadAuth()
+  editingTakeover.value[auth.id] = false
+}
+
 onMounted(() => {
   loadAuth()
 })
@@ -321,4 +399,172 @@ onMounted(() => {
   font-weight: 600;
   cursor: pointer;
 }
+
+/* 接管配置 */
+.takeover-section {
+  margin-top: 8px;
+  padding-top: 12px;
+  border-top: 1px solid #F3F4F6;
+  width: 100%;
+}
+
+.takeover-toggle-btn {
+  background: none;
+  border: none;
+  color: #6366F1;
+  font-size: 12px;
+  font-weight: 500;
+  cursor: pointer;
+  padding: 4px 0;
+}
+
+.takeover-form {
+  margin-top: 10px;
+  display: flex;
+  flex-direction: column;
+  gap: 10px;
+  background: #FAFBFC;
+  border-radius: 8px;
+  padding: 12px;
+}
+
+.takeover-row {
+  display: flex;
+  align-items: center;
+  gap: 10px;
+}
+
+.takeover-label {
+  font-size: 13px;
+  color: #374151;
+  font-weight: 500;
+  min-width: 72px;
+  flex-shrink: 0;
+}
+
+/* 切换开关 */
+.toggle-switch {
+  position: relative;
+  display: inline-block;
+  width: 40px;
+  height: 22px;
+  flex-shrink: 0;
+}
+
+.toggle-switch input {
+  opacity: 0;
+  width: 0;
+  height: 0;
+}
+
+.toggle-slider {
+  position: absolute;
+  cursor: pointer;
+  inset: 0;
+  background: #D1D5DB;
+  border-radius: 22px;
+  transition: background 0.2s;
+}
+
+.toggle-slider::before {
+  content: '';
+  position: absolute;
+  height: 16px;
+  width: 16px;
+  left: 3px;
+  bottom: 3px;
+  background: white;
+  border-radius: 50%;
+  transition: transform 0.2s;
+}
+
+.toggle-switch input:checked + .toggle-slider {
+  background: #6366F1;
+}
+
+.toggle-switch input:checked + .toggle-slider::before {
+  transform: translateX(18px);
+}
+
+.takeover-select {
+  flex: 1;
+  padding: 6px 10px;
+  border: 1px solid #E5E7EB;
+  border-radius: 6px;
+  font-size: 13px;
+  color: #374151;
+  background: white;
+  outline: none;
+}
+
+.takeover-select:disabled {
+  background: #F3F4F6;
+  color: #9CA3AF;
+}
+
+.takeover-input {
+  flex: 1;
+  padding: 6px 10px;
+  border: 1px solid #E5E7EB;
+  border-radius: 6px;
+  font-size: 13px;
+  color: #374151;
+  background: white;
+  outline: none;
+  width: 80px;
+}
+
+.takeover-actions {
+  display: flex;
+  gap: 8px;
+  justify-content: flex-end;
+  margin-top: 4px;
+}
+
+.takeover-cancel-btn {
+  padding: 6px 16px;
+  border-radius: 6px;
+  font-size: 12px;
+  font-weight: 500;
+  cursor: pointer;
+  border: 1px solid #E5E7EB;
+  background: white;
+  color: #6B7280;
+}
+
+.takeover-save-btn {
+  padding: 6px 16px;
+  border-radius: 6px;
+  font-size: 12px;
+  font-weight: 500;
+  cursor: pointer;
+  border: none;
+  background: #6366F1;
+  color: white;
+}
+
+/* 接管配置摘要 */
+.takeover-summary {
+  display: flex;
+  align-items: center;
+  gap: 8px;
+  margin-top: 6px;
+}
+
+.takeover-badge {
+  padding: 2px 8px;
+  border-radius: 4px;
+  font-size: 11px;
+  font-weight: 600;
+}
+
+.takeover-badge.enabled {
+  background: #EEF2FF;
+  color: #6366F1;
+}
+
+.takeover-mode-text {
+  font-size: 12px;
+  color: #6B7280;
+}
 </style>

docs/superpowers/specs/2026-08-07-avatar-takeover-chat-design.md

@@ -0,0 +1,284 @@
+# 数字分身接管聊天功能 — 设计文档
+
+## 1. 概述
+
+### 1.1 背景
+当前数字分身应用中，分身只能与主人一对一聊天。用户希望分身能在主人授权下，接管主人在会会平台上的聊天（单聊和群聊），替主人与他人对话。
+
+### 1.2 目标
+- 主人可在授权管理页面开启"分身接管聊天"功能
+- 支持两种接管模式：立即接管、延迟接管（N 秒后）
+- 分身回复优先使用知识库（QA 问答对 + 文档向量搜索）
+- 支持单聊和群聊两种场景
+- 以主人身份发送回复，对方无感知
+
+### 1.3 技术基础
+- 会会平台使用网易云信作为 IM 底层
+- 已有接口：`/api/im/netease`（获取云信凭证）、`/api/v2message/push/record/*`（消息推送）
+- 已有分身聊天接口：`POST /api/avatar/{id}/chat`
+
+---
+
+## 2. 架构设计
+
+### 2.1 整体架构
+
+```
+┌─────────────────────────────────────────────────────────┐
+│                    会会平台 (网易云信)                     │
+│  ┌─────────────┐  ┌─────────────┐  ┌─────────────────┐  │
+│  │  单聊 P2P   │  │  群聊 Group │  │  消息推送 Push  │  │
+│  └─────────────┘  └─────────────┘  └─────────────────┘  │
+└──────────────────────┬──────────────────────────────────┘
+                       │ HTTP API
+──────────────────────▼──────────────────────────────────┐
+│              分身接管服务 (Avatar Takeover Service)       │
+│  ┌──────────────┐  ┌──────────────┐  ┌───────────────┐ │
+│  │  消息监听器  │  │  接管决策器  │  │  回复执行器   │ │
+│  └──────────────┘  ──────────────┘  └───────────────┘ │
+│  ┌──────────────────────────────────────────────────┐  │
+│  │              Redis 延迟队列                       │  │
+│  └──────────────────────────────────────────────────┘  │
+└──────────────────────┬──────────────────────────────────┘
+                       │ 调用现有接口
+┌──────────────────────▼──────────────────────────────────┐
+│                   数字分身应用                            │
+│  ┌──────────────┐  ┌──────────────┐  ┌───────────────┐ │
+│  │  授权管理 UI │  │  分身聊天 API │  │  知识库       │ │
+│  └──────────────┘  └──────────────┘  └───────────────┘ │
+└─────────────────────────────────────────────────────────┘
+```
+
+### 2.2 模块划分
+
+| 模块 | 位置 | 职责 |
+|------|------|------|
+| 接管配置 | 授权管理页面 + authorizations 表 | 存储接管开关、模式、延迟秒数 |
+| 消息监听器 | 后端定时任务 | 轮询网易云信消息，发现新消息 |
+| 接管决策器 | 后端服务 | 判断是否接管、立即还是延迟 |
+| 回复执行器 | 后端服务 | 调用分身聊天接口生成回复，以主人身份发送 |
+| 延迟队列 | Redis | 存储待处理消息，实现延迟接管 |
+
+---
+
+## 3. 数据模型
+
+### 3.1 扩展 Authorization 表
+
+在现有 `authorizations` 表基础上新增字段：
+
+| 字段 | 类型 | 说明 |
+|------|------|------|
+| `takeover_enabled` | BOOLEAN | 是否开启分身接管聊天 |
+| `takeover_mode` | VARCHAR | 接管模式：`immediate`（立即）/ `delayed`（延迟） |
+| `takeover_delay_seconds` | INTEGER | 延迟秒数（仅 delayed 模式有效），默认 30 |
+
+**权限关联**：
+- `permissions` 数组中包含 `"takeover"` 时，表示该授权允许分身接管
+- `target_type = "user"` 且 `permissions` 含 `"takeover"` → 单聊接管
+- `target_type = "organization"` 或群聊相关 → 群聊接管
+
+### 3.2 Redis 数据结构
+
+```
+# 延迟消息队列
+takeover:delayed:{owner_accid}:{message_id} = {
+  "avatar_id": "...",
+  "from_accid": "...",
+  "content": "...",
+  "chat_type": "p2p|group",
+  "chat_id": "...",
+  "timestamp": 1234567890
+}
+TTL = takeover_delay_seconds + 10
+```
+
+---
+
+## 4. 核心流程
+
+### 4.1 消息监听流程
+
+```
+┌──────────┐     ┌──────────────┐     ┌──────────────┐
+│ 定时任务  │────►│ 轮询 Push API │────►│ 获取未读消息 │
+└──────────┘     └──────────────┘     └──────┬───────┘
+                                              │
+                                    ┌─────────▼─────────┐
+                                    │ 遍历每条新消息     │
+                                    └─────────┬─────────┘
+                                              │
+                                    ┌─────────▼─────────┐
+                                    │ 查询主人授权配置   │
+                                    └─────────┬─────────┘
+                                              │
+                          ┌───────────────────┼───────────────────┐
+                          │                   │                   │
+                    ┌─────▼─────┐      ┌──────▼──────    ┌──────▼──────┐
+                    │ 未开启接管 │      │ 立即接管模式 │    │ 延迟接管模式 │
+                    └───────────┘      └──────┬──────┘    └──────┬──────┘
+                                              │                  │
+                                    ┌─────────▼─────────┐ ┌────▼────
+                                    │ 立即调用分身回复   │ │ 写入延迟 │
+                                    ───────────────────┘ │ 队列    │
+                                                          └─────────┘
+```
+
+### 4.2 延迟接管流程
+
+```
+┌──────────────────────────────────────────────────────────┐
+│ 消息写入 Redis 延迟队列 (TTL = delay_seconds + 10)         │
+└────────────────────────┬─────────────────────────────────┘
+                         │
+                         │ 延迟期间
+                         │
+               ┌─────────▼─────────┐
+               │ 主人是否有回复？   │
+               └─────────┬─────────┘
+                    ────┴────┐
+                    │         │
+              ┌─────▼──┐ ┌───▼────
+              │ 有回复  │ │ 无回复  │
+              └──┬─────┘ └───┬────┘
+                 │           │
+           ┌─────▼─────┐ ───▼──────────────┐
+           │ 删除队列   │ │ TTL 到期触发      │
+           │ 不接管     │ │ 调用分身回复      │
+           ───────────┘ └──────────────────
+```
+
+### 4.3 回复生成流程
+
+```
+┌────────────────────┐
+│ 收到他人消息内容    │
+└─────────┬──────────┘
+          │
+┌─────────▼──────────┐
+│ 查找主人的分身      │
+│ (按 owner_id 匹配) │
+└─────────┬──────────
+          │
+┌─────────▼──────────┐
+│ 调用分身聊天接口    │
+│ POST /api/avatar/  │
+│   {id}/chat        │
+│ 优先：QA 问答对     │
+│ 其次：知识库文档    │
+│ 最后：Qwen 生成    │
+└─────────┬──────────┘
+          │
+┌─────────▼──────────
+│ 以主人身份发送回复  │
+│ 通过网易云信 API    │
+└────────────────────┘
+```
+
+---
+
+## 5. API 设计
+
+### 5.1 授权管理增强
+
+**PUT /api/avatar/{avatar_id}/authorizations/takeover**
+
+请求体：
+```json
+{
+  "authorization_id": "授权记录ID",
+  "takeover_enabled": true,
+  "takeover_mode": "delayed",
+  "takeover_delay_seconds": 30
+}
+```
+
+### 5.2 消息监听（内部接口）
+
+**POST /api/internal/takeover/process**
+
+由定时任务调用，无需认证。
+
+请求体：
+```json
+{
+  "owner_accid": "主人网易云信账号",
+  "messages": [
+    {
+      "msg_id": "消息ID",
+      "from_accid": "发送者账号",
+      "content": "消息内容",
+      "chat_type": "p2p",
+      "chat_id": "会话ID",
+      "timestamp": 1234567890
+    }
+  ]
+}
+```
+
+### 5.3 网易云信消息发送
+
+复用现有网易云信 API 封装，新增单聊消息发送方法：
+
+**POST /im/netease/message/send/p2p**
+
+请求体：
+```json
+{
+  "from_accid": "主人账号",
+  "to_accid": "对方账号",
+  "content": "回复内容",
+  "msg_type": "text"
+}
+```
+
+---
+
+## 6. 错误处理
+
+| 场景 | 处理方式 |
+|------|---------|
+| 分身聊天接口调用失败 | 记录日志，不回复，下次继续监听 |
+| 网易云信发送失败 | 重试 3 次，失败后记录日志 |
+| 主人有多个分身 | 取第一个 active 状态的分身 |
+| 知识库无匹配内容 | 使用 Qwen 兜底生成回复 |
+| Redis 连接失败 | 降级为立即接管模式 |
+
+---
+
+## 7. 配置项
+
+在 `SystemConfig` 表中新增配置：
+
+| 配置 key | 说明 | 默认值 |
+|---------|------|-------|
+| `takeover_poll_interval` | 消息轮询间隔（秒） | 10 |
+| `takeover_default_delay` | 默认延迟秒数 | 30 |
+| `takeover_retry_times` | 发送重试次数 | 3 |
+
+---
+
+## 8. 测试要点
+
+1. **功能测试**
+   - 立即接管：收到消息后立即由分身回复
+   - 延迟接管：主人 N 秒内回复则不分身接管，超时则分身回复
+   - 知识库优先：QA 问答对精确匹配优先返回
+
+2. **边界测试**
+   - 主人有多个分身时的选择逻辑
+   - 知识库为空时的 Qwen 兜底
+   - 网易云信 API 异常时的降级
+
+3. **性能测试**
+   - 轮询频率对服务器负载的影响
+   - Redis 延迟队列的内存占用
+
+---
+
+## 9. 后续扩展
+
+- Webhook 回调替代轮询（需要会会平台支持）
+- 多分身协同接管
+- 接管历史记录和统计
+- 接管效果评估和优化