feat(avatar): add BOXIM chat takeover

digital-avatar-app/backend/main.py

@@ -6,7 +6,6 @@ import logging
 
 from apscheduler.schedulers.asyncio import AsyncIOScheduler
 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
@@ -24,7 +23,6 @@ from responses import ok
 logger = logging.getLogger(__name__)
 
 takeover_scheduler = None
-takeover_db = None
 
 app = FastAPI(title="会会数字分身 API", version="1.0.0")
 
@@ -113,7 +111,7 @@ def seed():
 
 @app.on_event("startup")
 def on_startup():
-    global takeover_scheduler, takeover_db
+    global takeover_scheduler
 
     init_db()
     seed()
@@ -123,47 +121,43 @@ def on_startup():
 
     # --- Takeover scheduler ---
     try:
-        # Initialize Redis (optional)
+        # BOXIM production endpoints are intentionally separate from the login API.
-        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_PLATFORM_BASE_URL": os.getenv(
+                "HUIHUI_PLATFORM_BASE_URL", "https://open.99hui.com/api"
+            ),
+            "BOXIM_API_BASE_URL": os.getenv(
+                "BOXIM_API_BASE_URL", "https://im.99hui.com/api"
+            ),
             "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_TIMEOUT_SECONDS": os.getenv("BOXIM_TIMEOUT_SECONDS", "20"),
         }
         boxim_client = BoxIMClient(boxim_config)
 
-        # Initialize takeover service
         from services.takeover_service import TakeoverService
-        takeover_db = SessionLocal()
+        takeover_service = TakeoverService(SessionLocal, boxim_client)
-        takeover_service = TakeoverService(takeover_db, boxim_client, redis_client)
 
-        # AsyncIOScheduler awaits the service coroutine instead of dropping it.
+        poll_interval = max(0.5, float(os.getenv("BOXIM_POLL_INTERVAL_SECONDS", "1")))
         takeover_scheduler = AsyncIOScheduler()
         takeover_scheduler.add_job(
             takeover_service.poll_and_process_messages,
-            trigger=IntervalTrigger(seconds=10),
+            trigger=IntervalTrigger(seconds=poll_interval),
             id="takeover_message_poll",
+            max_instances=1,
+            coalesce=True,
         )
         takeover_scheduler.start()
-        logger.info("Takeover message polling scheduler started (interval=10s)")
+        logger.info("BOXIM takeover scheduler started (interval=%ss)", poll_interval)
     except Exception as e:
         stop_takeover_scheduler()
         logger.warning(f"Failed to initialize takeover scheduler, app will continue without it: {e}")
 
 
 def stop_takeover_scheduler():
-    global takeover_scheduler, takeover_db
+    global takeover_scheduler
 
     if takeover_scheduler is not None:
         try:
@@ -174,13 +168,6 @@ def stop_takeover_scheduler():
         finally:
             takeover_scheduler = None
 
-    if takeover_db is not None:
-        try:
-            takeover_db.close()
-        finally:
-            takeover_db = None
-
-
 @app.on_event("shutdown")
 def on_shutdown():
     stop_takeover_scheduler()

digital-avatar-app/backend/models.py

@@ -1,6 +1,17 @@
 import uuid
 
-from sqlalchemy import Column, String, Integer, Float, DateTime, Text, JSON, Boolean
+from sqlalchemy import (
+    Boolean,
+    Column,
+    DateTime,
+    Float,
+    Index,
+    Integer,
+    JSON,
+    String,
+    Text,
+    UniqueConstraint,
+)
 from sqlalchemy.sql import func
 
 from database import Base
@@ -74,6 +85,76 @@ class Authorization(Base):
         }
 
 
+class TakeoverCursor(Base):
+    """Durable BOXIM polling cursor for one avatar owner."""
+
+    __tablename__ = "takeover_cursors"
+    id = Column(String, primary_key=True, default=lambda: uuid.uuid4().hex)
+    avatar_id = Column(String, nullable=False, unique=True, index=True)
+    owner_id = Column(String, nullable=False, default="", index=True)
+    boxim_owner_id = Column(String, default="")
+    last_message_id = Column(String, default="0")
+    initialized = Column(Boolean, default=False)
+    last_polled_at = Column(DateTime)
+    last_error = Column(Text, default="")
+    created_at = Column(DateTime, server_default=func.now())
+    updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now())
+
+
+class TakeoverMessage(Base):
+    """BOXIM message receipt used for audit, deduplication, and chat context."""
+
+    __tablename__ = "takeover_messages"
+    __table_args__ = (
+        UniqueConstraint("owner_id", "boxim_message_id", name="uq_takeover_message_owner_boxim"),
+        Index("ix_takeover_message_conversation", "owner_id", "peer_id", "send_time"),
+    )
+
+    id = Column(String, primary_key=True, default=lambda: uuid.uuid4().hex)
+    avatar_id = Column(String, nullable=False, index=True)
+    owner_id = Column(String, nullable=False, index=True)
+    boxim_message_id = Column(String, nullable=False)
+    boxim_local_id = Column(String, nullable=True)
+    peer_id = Column(String, nullable=False, index=True)
+    direction = Column(String, nullable=False)  # incoming | outgoing
+    message_type = Column(Integer, default=0)
+    content = Column(Text, default="")
+    is_avatar = Column(Boolean, default=False)
+    send_time = Column(DateTime, nullable=False)
+    created_at = Column(DateTime, server_default=func.now())
+
+
+class TakeoverReplyTask(Base):
+    """Restart-safe three-second BOXIM reply task."""
+
+    __tablename__ = "takeover_reply_tasks"
+    __table_args__ = (
+        UniqueConstraint("owner_id", "trigger_message_id", name="uq_takeover_task_owner_trigger"),
+        Index("ix_takeover_task_due", "status", "scheduled_at"),
+        Index("ix_takeover_task_conversation", "owner_id", "peer_id", "status"),
+    )
+
+    id = Column(String, primary_key=True, default=lambda: uuid.uuid4().hex)
+    avatar_id = Column(String, nullable=False, index=True)
+    owner_id = Column(String, nullable=False, index=True)
+    peer_id = Column(String, nullable=False, index=True)
+    trigger_message_id = Column(String, nullable=False)
+    source_message_ids = Column(JSON, default=list)
+    prompt = Column(Text, default="")
+    response_text = Column(Text, default="")
+    status = Column(String, default="pending")
+    scheduled_at = Column(DateTime, nullable=False)
+    locked_at = Column(DateTime)
+    sent_at = Column(DateTime)
+    attempts = Column(Integer, default=0)
+    last_error = Column(Text, default="")
+    cancel_reason = Column(String, default="")
+    boxim_local_id = Column(String, nullable=False)
+    boxim_sent_message_id = Column(String, default="")
+    created_at = Column(DateTime, server_default=func.now())
+    updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now())
+
+
 class Organization(Base):
     __tablename__ = "organizations"
     id = Column(String, primary_key=True, default=lambda: uuid.uuid4().hex)

digital-avatar-app/backend/requirements.txt

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

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

@@ -11,7 +11,7 @@ router = APIRouter(tags=["授权"])
 TARGET_TYPES = {"user", "organization", "application"}
 PERMISSION_ORDER = ("friend", "chat", "publish", "browse", "interact", "takeover")
 ALLOWED_PERMISSIONS = set(PERMISSION_ORDER)
-AVATAR_PERMISSION_ORDER = PERMISSION_ORDER[:-1]
+AVATAR_PERMISSION_ORDER = PERMISSION_ORDER
 AVATAR_PERMISSION_KEY = "authorizationPermissions"
 DEFAULT_AVATAR_PERMISSIONS = ["friend", "chat"]
 LEGACY_PERMISSION_MAP = {

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

@@ -1,9 +1,12 @@
-"""数字分身单聊接管配置 API。"""
+"""数字分身 BOXIM 单聊接管 API。"""
+
+from datetime import datetime, timedelta
 
 from fastapi import APIRouter, Body, Depends, Header
 from sqlalchemy.orm import Session
 
 from database import get_db
+from models import TakeoverCursor, TakeoverReplyTask, User
 from responses import fail, ok
 from routers.authorizations import _require_authorization
 from routers.avatars import _require_owned_avatar
@@ -11,6 +14,53 @@ from routers.avatars import _require_owned_avatar
 router = APIRouter(tags=["分身接管"])
 
 
+@router.get("/avatar/{avatar_id}/takeover/status")
+def get_takeover_status(
+    avatar_id: str,
+    authorization: str = Header(None),
+    db: Session = Depends(get_db),
+):
+    avatar = _require_owned_avatar(db, avatar_id, authorization)
+    permissions = (avatar.config or {}).get("authorizationPermissions", [])
+    enabled = isinstance(permissions, list) and "takeover" in permissions
+    user = db.query(User).filter(User.huihui_user_id == avatar.owner_id).first()
+    cursor = db.query(TakeoverCursor).filter(TakeoverCursor.avatar_id == avatar.id).first()
+    pending_count = (
+        db.query(TakeoverReplyTask)
+        .filter(
+            TakeoverReplyTask.avatar_id == avatar.id,
+            TakeoverReplyTask.status.in_(("pending", "generating", "ready", "sending")),
+        )
+        .count()
+    )
+
+    if not enabled:
+        status, message = "disabled", "主动接管未开启"
+    elif not user or not user.huihui_token:
+        status, message = "needs_login", "请重新登录会会生产账号以连接 BOXIM"
+    elif cursor and cursor.last_error:
+        status, message = "error", cursor.last_error
+    elif (
+        cursor
+        and cursor.initialized
+        and cursor.last_polled_at
+        and cursor.last_polled_at >= datetime.utcnow() - timedelta(seconds=15)
+    ):
+        status, message = "ready", "BOXIM 已连接，收到私聊消息 3 秒后自动回复"
+    else:
+        status, message = "connecting", "正在连接 BOXIM"
+
+    return ok(
+        {
+            "enabled": enabled,
+            "status": status,
+            "message": message,
+            "pendingCount": pending_count,
+            "lastPolledAt": cursor.last_polled_at.isoformat() if cursor and cursor.last_polled_at else None,
+        }
+    )
+
+
 def _has(payload: dict, camel_key: str, snake_key: str) -> bool:
     return camel_key in payload or snake_key in payload
 

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

@@ -1,76 +1,178 @@
-"""盒子 IM 客户端 — 封装网易云信 IM 接口调用"""
+"""Client for Huihui's self-hosted BOXIM production APIs."""
+
 import hashlib
 import random
+import secrets
 import string
-from datetime import datetime
+import time
-from typing import Optional
+from datetime import datetime, timedelta, timezone
+from typing import Any
 
 import httpx
 
 
+_CN_TZ = timezone(timedelta(hours=8))
+
+
+class BoxIMError(RuntimeError):
+    def __init__(self, message: str, *, code: Any = None, auth_error: bool = False):
+        super().__init__(message)
+        self.code = code
+        self.auth_error = auth_error
+
+
 class BoxIMClient:
-    """盒子 IM 客户端，通过会会平台网关调用网易云信 IM"""
+    """Exchange Huihui credentials and call BOXIM's private-message API."""
 
     def __init__(self, config: dict):
-        self.base_url = config.get("HUIHUI_IM_BASE_URL", "http://192.168.1.200:60040")
+        self.platform_base_url = config.get(
+            "HUIHUI_PLATFORM_BASE_URL", "https://open.99hui.com/api"
+        ).rstrip("/")
+        self.im_base_url = config.get(
+            "BOXIM_API_BASE_URL", "https://im.99hui.com/api"
+        ).rstrip("/")
         self.app_id = config.get("HUIHUI_APP_ID", "")
         self.access_id = config.get("HUIHUI_ACCESS_ID", "")
         self.access_secret = config.get("HUIHUI_ACCESS_SECRET", "")
+        self.timeout = float(config.get("BOXIM_TIMEOUT_SECONDS", 20))
 
-    def _build_sign_params(self, extra: dict) -> dict:
+    def _build_sign_params(self, extra: dict | None = None) -> dict:
-        """构建带签名的请求参数（复用 news_service 签名模式）"""
+        """Build the same signed form used by Huihui's current production app."""
-        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,
+            "nonce": "".join(random.choices(string.ascii_lowercase + string.digits, k=12)),
-            "timestamp": timestamp,
+            "timestamp": datetime.now(_CN_TZ).strftime("%Y%m%d%H%M%S"),
-            **extra,
+            "signType": "MD5",
+            "signVersion": "1.0",
+            **(extra or {}),
         }
-        # 计算签名 — 排序 key, 过滤空值, 拼接后加 accessSecret, MD5 大写
+        params.pop("accessSecret", None)
-        keys = sorted(params.keys())
+        params.pop("signature", None)
         sign_parts = []
-        for k in keys:
+        for key in sorted(params):
-            if k in ("signature", "accessSecret"):
+            value = params[key]
+            if value in (None, "", []):
                 continue
-            v = params.get(k)
+            if isinstance(value, list):
-            if v and v != "" and v != []:
+                continue
-                sign_parts.append(f"{k}={v}")
+            sign_parts.append(f"{key}={value}")
-        sign_str = "&".join(sign_parts) + f"&accessSecret={self.access_secret}"
+        sign_source = "&".join(sign_parts) + f"&accessSecret={self.access_secret}"
-        signature = hashlib.md5(sign_str.encode("utf-8")).hexdigest().upper()
+        params["signature"] = hashlib.md5(sign_source.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]:
+    @staticmethod
-        """获取用户的网易云信 IM 凭证 (accid, token)"""
+    def _response_payload(response: httpx.Response) -> dict:
-        params = self._build_sign_params({"userId": user_id})
+        try:
-        async with httpx.AsyncClient(timeout=10) as client:
+            payload = response.json()
-            r = await client.post(
+        except ValueError as exc:
-                f"{self.base_url}/box/netease",
+            raise BoxIMError("BOXIM 返回了无效响应") from exc
-                params=params,
+        if not isinstance(payload, dict):
-            )
+            raise BoxIMError("BOXIM 返回格式不正确")
-        data = r.json()
+        return payload
-        if data.get("code") in (0, 200):
-            return data.get("data", {})
-        return None
 
-    async def send_p2p_message(
+    async def exchange_access_token(self, huihui_token: str) -> dict:
-        self, from_accid: str, to_accid: str, content: str
+        """Exchange a production Huihui token for a BOXIM access token."""
-    ) -> bool:
+        if not huihui_token:
-        """发送单聊消息（文本）"""
+            raise BoxIMError("缺少会会登录凭证", auth_error=True)
-        params = self._build_sign_params({
+        if not (self.app_id and self.access_id and self.access_secret):
-            "from": from_accid,
+            raise BoxIMError("会会开放平台凭证未配置", auth_error=True)
-            "to": to_accid,
+
-            "msgType": "text",
+        headers = {
-            "content": content,
+            "Authorization": f"Bearer {huihui_token}",
-        })
+            "appId": self.app_id,
-        async with httpx.AsyncClient(timeout=10) as client:
+            "windowAppId": self.app_id,
-            r = await client.post(
+        }
-                f"{self.base_url}/box/message/send/p2p",
+        async with httpx.AsyncClient(timeout=self.timeout, follow_redirects=True) as client:
-                params=params,
+            response = await client.post(
+                f"{self.platform_base_url}/im/box/netease",
+                headers=headers,
+                data=self._build_sign_params(),
             )
-        data = r.json()
+        payload = self._response_payload(response)
-        return data.get("code") in (0, 200)
+        data = payload.get("data") or {}
+        code = payload.get("code")
+        if response.status_code >= 400 or code not in (0, 200, "0", "200"):
+            raise BoxIMError(
+                payload.get("message") or "BOXIM 授权失败",
+                code=code or response.status_code,
+                auth_error=response.status_code in (400, 401, 403)
+                or code in (400, 401, 40100, 403, "400", "401", "40100", "403"),
+            )
+        if not data.get("accessToken"):
+            raise BoxIMError("会会未返回 BOXIM 访问凭证", auth_error=True)
+        return data
+
+    async def _request(
+        self,
+        method: str,
+        path: str,
+        access_token: str,
+        *,
+        params: dict | None = None,
+        json: dict | None = None,
+    ) -> Any:
+        headers = {"accessToken": access_token}
+        async with httpx.AsyncClient(timeout=self.timeout) as client:
+            response = await client.request(
+                method,
+                f"{self.im_base_url}{path}",
+                headers=headers,
+                params=params,
+                json=json,
+            )
+        payload = self._response_payload(response)
+        code = payload.get("code")
+        if response.status_code >= 400 or code not in (200, "200"):
+            raise BoxIMError(
+                payload.get("message") or "BOXIM 请求失败",
+                code=code or response.status_code,
+                auth_error=response.status_code in (400, 401, 403)
+                or code in (400, 401, 403, "400", "401", "403"),
+            )
+        return payload.get("data")
+
+    async def get_self(self, access_token: str) -> dict:
+        data = await self._request("GET", "/user/self", access_token)
+        if not isinstance(data, dict) or data.get("id") is None:
+            raise BoxIMError("BOXIM 未返回当前用户信息")
+        return data
+
+    async def fetch_private_messages(self, access_token: str, min_id: str = "0") -> list[dict]:
+        data = await self._request(
+            "GET",
+            "/message/private/loadOfflineMessage",
+            access_token,
+            params={"minId": str(min_id or "0")},
+        )
+        if data is None:
+            return []
+        if not isinstance(data, list):
+            raise BoxIMError("BOXIM 私聊消息格式不正确")
+        return [item for item in data if isinstance(item, dict)]
+
+    async def send_private_message(
+        self,
+        access_token: str,
+        peer_id: str,
+        content: str,
+        *,
+        local_id: int | str | None = None,
+    ) -> dict:
+        local_id = int(local_id or (int(time.time() * 1000) * 1000 + secrets.randbelow(1000)))
+        data = await self._request(
+            "POST",
+            "/message/private/send",
+            access_token,
+            json={
+                "localId": local_id,
+                "recvId": int(peer_id) if str(peer_id).isdigit() else peer_id,
+                "content": content,
+                "type": 0,
+                "receipt": False,
+                "atUserIds": [],
+            },
+        )
+        if not isinstance(data, dict):
+            raise BoxIMError("BOXIM 未返回发送结果")
+        return data

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

@@ -1,184 +1,565 @@
-"""Takeover service — message listening, decision, reply execution."""
+"""Restart-safe automatic replies over Huihui's self-hosted BOXIM."""
-import json
+
+import asyncio
+import hashlib
 import logging
-import os
+import re
-from typing import Optional
+import secrets
+import time
+from datetime import datetime, timedelta
+from typing import Callable
 
-import httpx
 from sqlalchemy.orm import Session
 
-from models import Avatar, Authorization
+from models import (
-from services.boxim_client import BoxIMClient
+    Avatar,
+    TakeoverCursor,
+    TakeoverMessage,
+    TakeoverReplyTask,
+    User,
+)
+from services.boxim_client import BoxIMClient, BoxIMError
 
 logger = logging.getLogger(__name__)
 
+ACTIVE_TASK_STATUSES = ("pending", "generating", "ready", "sending")
+GENERATABLE_TASK_STATUSES = ("pending",)
+MAX_PROMPT_LENGTH = 4000
+MAX_STALE_SECONDS = 120
+STUCK_LOCK_SECONDS = 90
+TAKEOVER_PERMISSION = "takeover"
+
+
+def _utcnow() -> datetime:
+    return datetime.utcnow()
+
+
+def _takeover_enabled(avatar: Avatar | None) -> bool:
+    if not avatar or avatar.status != "active":
+        return False
+    permissions = (avatar.config or {}).get("authorizationPermissions", [])
+    return isinstance(permissions, list) and TAKEOVER_PERMISSION in permissions
+
+
+def _boxim_time(value, fallback: datetime) -> datetime:
+    try:
+        timestamp = float(value)
+        if timestamp > 10_000_000_000:
+            timestamp /= 1000
+        return datetime.utcfromtimestamp(timestamp)
+    except (TypeError, ValueError, OSError, OverflowError):
+        return fallback
+
+
+def _numeric_id(value) -> int:
+    try:
+        return int(value)
+    except (TypeError, ValueError):
+        return 0
+
+
+def _plain_text_reply(value: str) -> str:
+    """BOXIM is plain text, so remove Markdown markers without damaging paragraphs."""
+    text = (value or "").replace("\r\n", "\n").replace("\r", "\n")
+    text = re.sub(r"```(?:\w+)?\n?(.*?)```", r"\1", text, flags=re.S)
+    text = re.sub(r"\*\*(.*?)\*\*|__(.*?)__", lambda m: m.group(1) or m.group(2), text)
+    text = re.sub(r"(?<!\*)\*([^*\n]+)\*(?!\*)", r"\1", text)
+    text = re.sub(r"`([^`]+)`", r"\1", text)
+    text = re.sub(r"^\s{0,3}#{1,6}\s*", "", text, flags=re.M)
+    lines = [line.strip() for line in text.split("\n")]
+    return "\n".join(line for line in lines if line).strip()
+
 
 class TakeoverService:
-    """Service for handling avatar takeover — generating replies and sending them via IM."""
+    """Poll BOXIM, prepare replies during the grace period, then send at +3s."""
 
     def __init__(
         self,
-        db: Session,
+        session_factory: Callable[[], Session],
         boxim_client: BoxIMClient,
-        redis_client=None,
+        *,
+        reply_delay_seconds: int = 3,
+        now: Callable[[], datetime] = _utcnow,
     ):
-        self.db = db
+        self.session_factory = session_factory
         self.boxim = boxim_client
-        self.redis = redis_client
+        self.reply_delay_seconds = reply_delay_seconds
-        self._chat_api_base = os.getenv(
+        self.now = now
-            "TAKEOVER_CHAT_API_BASE", "http://localhost:8000/api"
+        self._sessions: dict[str, dict] = {}
-        )
+        self._run_lock = asyncio.Lock()
-
-    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, Avatar.status == "active")
-            .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.target_type == "user")
-            .filter(Authorization.status == "active")
-            .filter(Authorization.takeover_enabled == True)
-            .first()
-        )
-        permissions = set(auth.permissions or []) if auth else set()
-        return auth if auth and "takeover" in permissions 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."""
+        """Run one complete cycle; polling always happens before reply dispatch."""
+        if self._run_lock.locked():
+            return
+        async with self._run_lock:
+            self._recover_stuck_tasks()
+            avatar_ids = self._enabled_avatar_ids()
+            self._cancel_disabled_tasks(set(avatar_ids))
+            for avatar_id in avatar_ids:
+                await self._sync_avatar(avatar_id)
+
+            generated = await self._prepare_replies()
+            if generated:
+                # Catch a human reply sent while the model was preparing its answer.
+                for avatar_id in avatar_ids:
+                    await self._sync_avatar(avatar_id)
+            await self._dispatch_ready_replies()
+
+    def _enabled_avatar_ids(self) -> list[str]:
+        db = self.session_factory()
         try:
-            messages = await self.fetch_unread_messages()
+            return [
-            for msg in messages:
+                avatar.id
-                await self.process_message(msg)
+                for avatar in db.query(Avatar).filter(Avatar.status == "active").all()
-        except Exception as e:
+                if _takeover_enabled(avatar)
-            logger.error(f"poll_and_process_messages failed: {e}")
+            ]
+        finally:
+            db.close()
 
-    async def fetch_unread_messages(self) -> list:
+    def _cancel_disabled_tasks(self, enabled_avatar_ids: set[str]):
-        """Fetch unread messages from Box IM. Stub — replace with real API call."""
+        db = self.session_factory()
-        logger.debug("fetch_unread_messages: no real API wired yet")
+        try:
-        return []
+            tasks = (
+                db.query(TakeoverReplyTask)
+                .filter(TakeoverReplyTask.status.in_(ACTIVE_TASK_STATUSES))
+                .all()
+            )
+            changed = False
+            for task in tasks:
+                if task.avatar_id not in enabled_avatar_ids:
+                    task.status = "cancelled"
+                    task.cancel_reason = "takeover_disabled"
+                    task.locked_at = None
+                    changed = True
+            if changed:
+                db.commit()
+        finally:
+            db.close()
 
-    async def process_message(self, message: dict):
+    def _recover_stuck_tasks(self):
-        """Process a single message: check takeover, dispatch immediate or delayed."""
+        db = self.session_factory()
-        owner_id = message.get("owner_huihui_id", "")
+        try:
-        from_id = message.get("from_accid", "")
+            threshold = self.now() - timedelta(seconds=STUCK_LOCK_SECONDS)
+            tasks = (
+                db.query(TakeoverReplyTask)
+                .filter(
+                    TakeoverReplyTask.status.in_(("generating", "sending")),
+                    TakeoverReplyTask.locked_at.isnot(None),
+                    TakeoverReplyTask.locked_at < threshold,
+                )
+                .all()
+            )
+            for task in tasks:
+                task.status = "pending" if task.status == "generating" else "ready"
+                task.locked_at = None
+                task.last_error = "上次处理意外中断，已自动恢复"
+            if tasks:
+                db.commit()
+        finally:
+            db.close()
 
-        auth = self.check_takeover_enabled(owner_id, from_id)
+    async def _boxim_session(self, user: User) -> dict:
-        if not auth:
+        token_fingerprint = hashlib.sha256((user.huihui_token or "").encode()).hexdigest()
+        cached = self._sessions.get(user.id)
+        if (
+            cached
+            and cached["expires_at"] > time.monotonic()
+            and cached["token_fingerprint"] == token_fingerprint
+        ):
+            return cached
+
+        token_data = await self.boxim.exchange_access_token(user.huihui_token)
+        access_token = token_data["accessToken"]
+        profile = await self.boxim.get_self(access_token)
+        try:
+            expires_in = int(token_data.get("accessTokenExpiresIn") or 3600)
+        except (TypeError, ValueError):
+            expires_in = 3600
+        if expires_in > 86_400:
+            expires_in //= 1000
+        cache_for = max(60, min(expires_in - 60, 3600))
+        cached = {
+            "access_token": access_token,
+            "boxim_owner_id": str(profile["id"]),
+            "expires_at": time.monotonic() + cache_for,
+            "token_fingerprint": token_fingerprint,
+        }
+        self._sessions[user.id] = cached
+        return cached
+
+    def _forget_boxim_session(self, user_id: str):
+        self._sessions.pop(user_id, None)
+
+    async def _sync_avatar(self, avatar_id: str) -> bool:
+        db = self.session_factory()
+        try:
+            avatar = db.query(Avatar).filter(Avatar.id == avatar_id).first()
+            if not _takeover_enabled(avatar):
+                return False
+            user = db.query(User).filter(User.huihui_user_id == avatar.owner_id).first()
+            cursor = db.query(TakeoverCursor).filter(TakeoverCursor.avatar_id == avatar.id).first()
+            if not cursor:
+                cursor = TakeoverCursor(avatar_id=avatar.id, owner_id=avatar.owner_id)
+                db.add(cursor)
+                db.flush()
+            if not user or not user.huihui_token:
+                cursor.last_error = "请重新登录会会生产账号后再开启主动接管"
+                cursor.last_polled_at = self.now()
+                db.commit()
+                return False
+
+            try:
+                session = await self._boxim_session(user)
+                owner_boxim_id = session["boxim_owner_id"]
+                if cursor.boxim_owner_id and cursor.boxim_owner_id != owner_boxim_id:
+                    cursor.initialized = False
+                    cursor.last_message_id = "0"
+                cursor.boxim_owner_id = owner_boxim_id
+                messages = await self.boxim.fetch_private_messages(
+                    session["access_token"], cursor.last_message_id or "0"
+                )
+            except Exception as exc:
+                if isinstance(exc, BoxIMError) and exc.auth_error:
+                    self._forget_boxim_session(user.id)
+                    cursor.last_error = "BOXIM 授权已失效，请重新登录会会生产账号"
+                else:
+                    cursor.last_error = f"BOXIM 暂时连接失败：{str(exc)[:160]}"
+                cursor.last_polled_at = self.now()
+                db.commit()
+                logger.warning("BOXIM sync failed for avatar %s: %s", avatar.id, exc)
+                return False
+
+            messages.sort(key=lambda item: (_numeric_id(item.get("id")), item.get("sendTime") or 0))
+            priming = not bool(cursor.initialized)
+            max_message_id = _numeric_id(cursor.last_message_id)
+            for message in messages:
+                self._record_message(
+                    db,
+                    avatar,
+                    cursor.boxim_owner_id,
+                    message,
+                    schedule_reply=not priming,
+                )
+                max_message_id = max(max_message_id, _numeric_id(message.get("id")))
+
+            cursor.last_message_id = str(max_message_id)
+            cursor.initialized = True
+            cursor.last_polled_at = self.now()
+            cursor.last_error = ""
+            db.commit()
+            return True
+        except Exception:
+            db.rollback()
+            logger.exception("Failed to persist BOXIM messages for avatar %s", avatar_id)
+            return False
+        finally:
+            db.close()
+
+    def _record_message(
+        self,
+        db: Session,
+        avatar: Avatar,
+        boxim_owner_id: str,
+        message: dict,
+        *,
+        schedule_reply: bool,
+    ):
+        message_id = str(message.get("id") or "").strip()
+        if not message_id:
+            return
+        local_id = str(message.get("localId") or "").strip() or None
+        if (
+            db.query(TakeoverMessage)
+            .filter(
+                TakeoverMessage.owner_id == avatar.owner_id,
+                TakeoverMessage.boxim_message_id == message_id,
+            )
+            .first()
+        ):
             return
 
-        if auth.takeover_mode == "immediate":
+        send_id = str(message.get("sendId") or "")
-            await self.execute_takeover(auth, message)
+        recv_id = str(message.get("recvId") or "")
-        elif not self.redis:
+        if send_id == boxim_owner_id:
-            logger.warning("Redis not configured, executing delayed takeover immediately")
+            direction, peer_id = "outgoing", recv_id
-            await self.execute_takeover(auth, message)
+        elif recv_id == boxim_owner_id:
+            direction, peer_id = "incoming", send_id
         else:
-            self.enqueue_delayed_message(auth, message)
+            return
+        if not peer_id:
+            return
+
+        now = self.now()
+        send_time = _boxim_time(message.get("sendTime"), now)
+        is_avatar = False
+        if direction == "outgoing" and local_id:
+            is_avatar = bool(
+                db.query(TakeoverReplyTask)
+                .filter(
+                    TakeoverReplyTask.owner_id == avatar.owner_id,
+                    TakeoverReplyTask.boxim_local_id == local_id,
+                    TakeoverReplyTask.status == "sent",
+                )
+                .first()
+            )
+
+        event = TakeoverMessage(
+            avatar_id=avatar.id,
+            owner_id=avatar.owner_id,
+            boxim_message_id=message_id,
+            boxim_local_id=local_id,
+            peer_id=peer_id,
+            direction=direction,
+            message_type=int(message.get("type") or 0),
+            content=str(message.get("content") or ""),
+            is_avatar=is_avatar,
+            send_time=send_time,
+        )
+        db.add(event)
+        db.flush()
+
+        if direction == "outgoing":
+            if not is_avatar:
+                self._cancel_conversation(db, avatar.owner_id, peer_id, "owner_replied")
+            return
+        if not schedule_reply or event.message_type != 0 or not event.content.strip():
+            return
+        if (now - send_time).total_seconds() > MAX_STALE_SECONDS:
+            return
+        self._schedule_reply(db, avatar, event)
+
+    @staticmethod
+    def _cancel_conversation(db: Session, owner_id: str, peer_id: str, reason: str):
+        tasks = (
+            db.query(TakeoverReplyTask)
+            .filter(
+                TakeoverReplyTask.owner_id == owner_id,
+                TakeoverReplyTask.peer_id == peer_id,
+                TakeoverReplyTask.status.in_(ACTIVE_TASK_STATUSES),
+            )
+            .all()
+        )
+        for task in tasks:
+            task.status = "cancelled"
+            task.cancel_reason = reason
+            task.locked_at = None
+
+    def _schedule_reply(self, db: Session, avatar: Avatar, event: TakeoverMessage):
+        active_tasks = (
+            db.query(TakeoverReplyTask)
+            .filter(
+                TakeoverReplyTask.owner_id == avatar.owner_id,
+                TakeoverReplyTask.peer_id == event.peer_id,
+                TakeoverReplyTask.status.in_(("pending", "generating", "ready")),
+            )
+            .order_by(TakeoverReplyTask.created_at.desc())
+            .all()
+        )
+        prompt_parts = []
+        source_ids = []
+        if active_tasks:
+            latest = active_tasks[0]
+            prompt_parts.append(latest.prompt)
+            source_ids.extend(latest.source_message_ids or [])
+            for task in active_tasks:
+                task.status = "cancelled"
+                task.cancel_reason = "newer_incoming_message"
+                task.locked_at = None
+        prompt_parts.append(event.content.strip())
+        source_ids.append(event.boxim_message_id)
+        prompt = "\n".join(part for part in prompt_parts if part).strip()[-MAX_PROMPT_LENGTH:]
+        due_at = event.send_time + timedelta(seconds=self.reply_delay_seconds)
+        task_id = secrets.token_hex(16)
+        local_id = int(time.time() * 1000) * 1000 + secrets.randbelow(1000)
+        db.add(
+            TakeoverReplyTask(
+                id=task_id,
+                avatar_id=avatar.id,
+                owner_id=avatar.owner_id,
+                peer_id=event.peer_id,
+                trigger_message_id=event.boxim_message_id,
+                source_message_ids=source_ids,
+                prompt=prompt,
+                status="pending",
+                scheduled_at=due_at,
+                boxim_local_id=str(local_id),
+            )
+        )
+
+    async def _prepare_replies(self) -> int:
+        db = self.session_factory()
+        try:
+            task_ids = [
+                row[0]
+                for row in (
+                    db.query(TakeoverReplyTask.id)
+                    .filter(
+                        TakeoverReplyTask.status.in_(GENERATABLE_TASK_STATUSES),
+                        TakeoverReplyTask.response_text == "",
+                    )
+                    .order_by(TakeoverReplyTask.created_at.asc())
+                    .limit(10)
+                    .all()
+                )
+            ]
+        finally:
+            db.close()
+
+        generated = 0
+        for task_id in task_ids:
+            if await asyncio.to_thread(self._generate_reply, task_id):
+                generated += 1
+        return generated
+
+    def _generate_reply(self, task_id: str) -> bool:
+        db = self.session_factory()
+        try:
+            task = db.query(TakeoverReplyTask).filter(TakeoverReplyTask.id == task_id).first()
+            if not task or task.status != "pending":
+                return False
+            avatar = db.query(Avatar).filter(Avatar.id == task.avatar_id).first()
+            if not _takeover_enabled(avatar):
+                task.status = "cancelled"
+                task.cancel_reason = "takeover_disabled"
+                db.commit()
+                return False
+
+            task.status = "generating"
+            task.locked_at = self.now()
+            db.commit()
+
+            excluded_ids = set(task.source_message_ids or [])
+            events = (
+                db.query(TakeoverMessage)
+                .filter(
+                    TakeoverMessage.owner_id == task.owner_id,
+                    TakeoverMessage.peer_id == task.peer_id,
+                )
+                .order_by(TakeoverMessage.send_time.desc())
+                .limit(30)
+                .all()
+            )
+            history = []
+            for event in reversed(events):
+                if event.boxim_message_id in excluded_ids or not event.content.strip():
+                    continue
+                history.append(
+                    {
+                        "role": "user" if event.direction == "incoming" else "assistant",
+                        "content": event.content.strip(),
+                    }
+                )
+            history = history[-10:]
+
+            from routers.chat import _resolve_reply
+
+            result = _resolve_reply(db, avatar, task.prompt, history)
+            answer = _plain_text_reply(result.get("answer", ""))
+            db.refresh(task)
+            if task.status != "generating":
+                return False
+            if not answer:
+                raise RuntimeError("分身没有生成有效回复")
+            task.response_text = answer
+            task.status = "ready"
+            task.locked_at = None
+            task.last_error = ""
+            db.commit()
+            return True
+        except Exception as exc:
+            db.rollback()
+            task = db.query(TakeoverReplyTask).filter(TakeoverReplyTask.id == task_id).first()
+            if task and task.status in ("pending", "generating"):
+                task.attempts = (task.attempts or 0) + 1
+                task.status = "pending" if task.attempts < 3 else "failed"
+                task.locked_at = None
+                task.last_error = str(exc)[:300]
+                db.commit()
+            logger.warning("Failed to prepare takeover reply %s: %s", task_id, exc)
+            return False
+        finally:
+            db.close()
+
+    async def _dispatch_ready_replies(self):
+        db = self.session_factory()
+        try:
+            task_ids = [
+                row[0]
+                for row in (
+                    db.query(TakeoverReplyTask.id)
+                    .filter(
+                        TakeoverReplyTask.status == "ready",
+                        TakeoverReplyTask.scheduled_at <= self.now(),
+                    )
+                    .order_by(TakeoverReplyTask.scheduled_at.asc())
+                    .limit(10)
+                    .all()
+                )
+            ]
+        finally:
+            db.close()
+
+        for task_id in task_ids:
+            await self._send_task(task_id)
+
+    async def _send_task(self, task_id: str) -> bool:
+        db = self.session_factory()
+        user = None
+        try:
+            task = db.query(TakeoverReplyTask).filter(TakeoverReplyTask.id == task_id).first()
+            if not task or task.status != "ready":
+                return False
+            avatar = db.query(Avatar).filter(Avatar.id == task.avatar_id).first()
+            if not _takeover_enabled(avatar):
+                task.status = "cancelled"
+                task.cancel_reason = "takeover_disabled"
+                db.commit()
+                return False
+            if (self.now() - task.scheduled_at).total_seconds() > MAX_STALE_SECONDS:
+                task.status = "cancelled"
+                task.cancel_reason = "stale_reply"
+                db.commit()
+                return False
+            user = db.query(User).filter(User.huihui_user_id == task.owner_id).first()
+            if not user or not user.huihui_token:
+                raise BoxIMError("缺少会会登录凭证", auth_error=True)
+
+            task.status = "sending"
+            task.locked_at = self.now()
+            db.commit()
+            session = await self._boxim_session(user)
+            result = await self.boxim.send_private_message(
+                session["access_token"],
+                task.peer_id,
+                task.response_text,
+                local_id=task.boxim_local_id,
+            )
+            db.refresh(task)
+            if task.status != "sending":
+                return False
+            task.status = "sent"
+            task.sent_at = self.now()
+            task.locked_at = None
+            task.last_error = ""
+            task.boxim_sent_message_id = str(result.get("id") or "")
+            db.commit()
+            logger.info("BOXIM takeover reply sent for task %s", task.id)
+            return True
+        except Exception as exc:
+            db.rollback()
+            if user and isinstance(exc, BoxIMError) and exc.auth_error:
+                self._forget_boxim_session(user.id)
+            task = db.query(TakeoverReplyTask).filter(TakeoverReplyTask.id == task_id).first()
+            if task and task.status in ("ready", "sending"):
+                task.attempts = (task.attempts or 0) + 1
+                task.status = "ready" if task.attempts < 3 else "failed"
+                task.locked_at = None
+                task.last_error = str(exc)[:300]
+                if task.status == "ready":
+                    task.scheduled_at = self.now() + timedelta(seconds=2 ** task.attempts)
+                db.commit()
+            logger.warning("Failed to send takeover reply %s: %s", task_id, exc)
+            return False
+        finally:
+            db.close()

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

@@ -2,7 +2,14 @@ import uuid
 
 import pytest
 from database import init_db, SessionLocal
-from models import Authorization, Avatar, User
+from models import (
+    Authorization,
+    Avatar,
+    TakeoverCursor,
+    TakeoverMessage,
+    TakeoverReplyTask,
+    User,
+)
 
 
 @pytest.fixture(scope="session", autouse=True)
@@ -84,10 +91,20 @@ def authorization_context():
         }
     finally:
         db.rollback()
-        db.query(Authorization).filter(
+        avatar_ids = [avatar.id, other_avatar.id]
-            Authorization.avatar_id.in_([avatar.id, other_avatar.id])
+        db.query(TakeoverReplyTask).filter(
+            TakeoverReplyTask.avatar_id.in_(avatar_ids)
         ).delete(synchronize_session=False)
-        db.query(Avatar).filter(Avatar.id.in_([avatar.id, other_avatar.id])).delete(
+        db.query(TakeoverMessage).filter(
+            TakeoverMessage.avatar_id.in_(avatar_ids)
+        ).delete(synchronize_session=False)
+        db.query(TakeoverCursor).filter(
+            TakeoverCursor.avatar_id.in_(avatar_ids)
+        ).delete(synchronize_session=False)
+        db.query(Authorization).filter(
+            Authorization.avatar_id.in_(avatar_ids)
+        ).delete(synchronize_session=False)
+        db.query(Avatar).filter(Avatar.id.in_(avatar_ids)).delete(
             synchronize_session=False
         )
         db.query(User).filter(User.id.in_([owner.id, other.id])).delete(

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

@@ -108,13 +108,13 @@ def test_avatar_permission_settings_default_and_persist(authorization_context):
     updated = client.put(
         endpoint,
         headers=context["owner_headers"],
-        json={"permissions": ["interact", "publish", "friend", "friend"]},
+        json={"permissions": ["interact", "takeover", "publish", "friend", "friend"]},
     ).json()
     assert updated["code"] == 200
-    assert updated["data"]["permissions"] == ["friend", "publish", "interact"]
+    assert updated["data"]["permissions"] == ["friend", "publish", "interact", "takeover"]
 
     reloaded = client.get(endpoint, headers=context["owner_headers"]).json()
-    assert reloaded["data"]["permissions"] == ["friend", "publish", "interact"]
+    assert reloaded["data"]["permissions"] == ["friend", "publish", "interact", "takeover"]
 
 
 def test_avatar_permission_settings_allow_all_disabled(authorization_context):
@@ -137,7 +137,7 @@ def test_avatar_permission_settings_validate_owner_and_permissions(authorization
     invalid = client.put(
         endpoint,
         headers=context["owner_headers"],
-        json={"permissions": ["takeover"]},
+        json={"permissions": ["admin"]},
     ).json()
     assert invalid["code"] == 400
 

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

@@ -1,134 +1,118 @@
-"""Tests for the Box IM client (Netease Yunxin gateway wrapper)."""
+"""Contract tests for the self-hosted BOXIM client."""
-import pytest
+
 from unittest.mock import AsyncMock, MagicMock, patch
 
+import pytest
+
+from services.boxim_client import BoxIMClient, BoxIMError
+
 
 @pytest.fixture
-def mock_config():
+def config():
     return {
-        "HUIHUI_IM_BASE_URL": "http://192.168.1.200:60040",
+        "HUIHUI_PLATFORM_BASE_URL": "https://open.example/api",
+        "BOXIM_API_BASE_URL": "https://im.example/api",
         "HUIHUI_APP_ID": "test_app",
         "HUIHUI_ACCESS_ID": "test_access",
         "HUIHUI_ACCESS_SECRET": "test_secret",
     }
 
 
-def _make_mock_response(json_data: dict):
+def _response(payload: dict, status_code: int = 200):
-    """Create a properly configured mock for httpx.Response."""
+    response = MagicMock()
-    mock_response = MagicMock()
+    response.status_code = status_code
-    mock_response.json.return_value = json_data
+    response.json.return_value = payload
-    return mock_response
+    return response
 
 
-def _patch_httpx_client(json_data: dict):
+def _client_patch(*, post_payload=None, request_payload=None, status_code=200):
-    """Patch httpx.AsyncClient so that `async with httpx.AsyncClient() as c: await c.post(...)` returns json_data."""
+    client = AsyncMock()
-    mock_client = AsyncMock()
+    if post_payload is not None:
-    mock_client.post.return_value = _make_mock_response(json_data)
+        client.post.return_value = _response(post_payload, status_code)
-
+    if request_payload is not None:
-    mock_cm = AsyncMock()
+        client.request.return_value = _response(request_payload, status_code)
-    mock_cm.__aenter__.return_value = mock_client
+    context = AsyncMock()
-    mock_cm.__aexit__.return_value = None
+    context.__aenter__.return_value = client
-
+    context.__aexit__.return_value = None
-    return patch("httpx.AsyncClient", return_value=mock_cm)
+    return patch("services.boxim_client.httpx.AsyncClient", return_value=context), client
 
 
 @pytest.mark.asyncio
-async def test_get_credentials(mock_config):
+async def test_exchange_access_token_uses_huihui_bearer_and_signed_form(config):
-    """get_credentials should return accid and token from the gateway response."""
+    mocked, client = _client_patch(
-    with _patch_httpx_client({"code": 200, "data": {"accid": "user123", "token": "tok_xyz"}}):
+        post_payload={"code": 0, "data": {"accessToken": "box-token", "accessTokenExpiresIn": 3600}}
-        from services.boxim_client import BoxIMClient
+    )
+    with mocked:
+        result = await BoxIMClient(config).exchange_access_token("huihui-token")
 
-        client = BoxIMClient(mock_config)
+    assert result["accessToken"] == "box-token"
-        result = await client.get_credentials("user123")
+    call = client.post.await_args
-
+    assert call.args[0] == "https://open.example/api/im/box/netease"
-        assert result["accid"] == "user123"
+    assert call.kwargs["headers"]["Authorization"] == "Bearer huihui-token"
-        assert result["token"] == "tok_xyz"
+    assert call.kwargs["data"]["appId"] == "test_app"
+    assert len(call.kwargs["data"]["signature"]) == 32
 
 
 @pytest.mark.asyncio
-async def test_send_p2p_message_success(mock_config):
+async def test_get_self_and_incremental_private_messages_use_boxim_header(config):
-    """send_p2p_message should return True when the gateway responds with code 200."""
+    client_instance = BoxIMClient(config)
-    with _patch_httpx_client({"code": 200}):
+    mocked, client = _client_patch(
-        from services.boxim_client import BoxIMClient
+        request_payload={"code": 200, "data": {"id": 42, "nickName": "Owner"}}
+    )
+    with mocked:
+        profile = await client_instance.get_self("box-token")
+    assert profile["id"] == 42
+    assert client.request.await_args.kwargs["headers"] == {"accessToken": "box-token"}
 
-        client = BoxIMClient(mock_config)
+    mocked, client = _client_patch(
-        result = await client.send_p2p_message("owner_acc", "target_acc", "Hello")
+        request_payload={"code": 200, "data": [{"id": 101, "sendId": 7, "recvId": 42}]}
-
+    )
-        assert result is True
+    with mocked:
+        messages = await client_instance.fetch_private_messages("box-token", "100")
+    assert messages[0]["id"] == 101
+    assert client.request.await_args.kwargs["params"] == {"minId": "100"}
 
 
 @pytest.mark.asyncio
-async def test_send_p2p_message_failure(mock_config):
+async def test_send_private_message_matches_boxim_payload(config):
-    """send_p2p_message should return False when the gateway responds with a non-200 code."""
+    mocked, client = _client_patch(
-    with _patch_httpx_client({"code": 500, "message": "error"}):
+        request_payload={"code": 200, "data": {"id": 88, "localId": 12345}}
-        from services.boxim_client import BoxIMClient
+    )
+    with mocked:
+        result = await BoxIMClient(config).send_private_message(
+            "box-token", "77", "你好", local_id="12345"
+        )
 
-        client = BoxIMClient(mock_config)
+    assert result["id"] == 88
-        result = await client.send_p2p_message("owner_acc", "target_acc", "Hello")
+    call = client.request.await_args
-
+    assert call.args[:2] == ("POST", "https://im.example/api/message/private/send")
-        assert result is False
+    assert call.kwargs["json"] == {
+        "localId": 12345,
+        "recvId": 77,
+        "content": "你好",
+        "type": 0,
+        "receipt": False,
+        "atUserIds": [],
+    }
 
 
 @pytest.mark.asyncio
-async def test_get_credentials_returns_none_on_error(mock_config):
+async def test_boxim_auth_error_is_explicit(config):
-    """get_credentials should return None when the gateway responds with an error code."""
+    mocked, _ = _client_patch(
-    with _patch_httpx_client({"code": 500, "message": "user not found"}):
+        request_payload={"code": 400, "message": "未登录"}, status_code=200
-        from services.boxim_client import BoxIMClient
+    )
-
+    with mocked, pytest.raises(BoxIMError) as exc_info:
-        client = BoxIMClient(mock_config)
+        await BoxIMClient(config).get_self("expired")
-        result = await client.get_credentials("nonexistent")
+    assert exc_info.value.auth_error is True
-
-        assert result is None
 
 
-def test_build_sign_params_contains_required_fields(mock_config):
+def test_sign_params_include_production_required_fields(config):
-    """_build_sign_params should produce appId, accessId, nonce, timestamp, signature, signType, signVersion."""
+    params = BoxIMClient(config)._build_sign_params()
-    from services.boxim_client import BoxIMClient
+    assert params["appId"] == "test_app"
-
+    assert params["accessId"] == "test_access"
-    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
-
+    assert len(params["timestamp"]) == 14
-
+    assert len(params["signature"]) == 32
-def test_build_sign_params_excludes_signature_and_accessSecret_from_signing_string(mock_config):
+    assert "accessSecret" not in params
-    """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

@@ -1,10 +1,12 @@
-"""Tests for the authorization takeover configuration endpoint."""
+"""Tests for takeover configuration and BOXIM connection status."""
+
+from datetime import datetime
 
 from fastapi.testclient import TestClient
 
 from database import SessionLocal
 from main import app
-from models import Authorization
+from models import Authorization, TakeoverCursor, TakeoverReplyTask, User
 
 
 client = TestClient(app)
@@ -123,3 +125,81 @@ def test_takeover_is_limited_to_active_user_authorizations(authorization_context
     ).json()
     assert response["code"] == 400
     assert "单聊接管" in response["message"]
+
+
+def test_takeover_status_reports_disabled_and_requires_owner_login(authorization_context):
+    context = authorization_context
+    endpoint = f"/api/avatar/{context['avatar'].id}/takeover/status"
+
+    disabled = client.get(endpoint, headers=context["owner_headers"])
+    assert disabled.status_code == 200
+    assert disabled.json()["data"]["status"] == "disabled"
+
+    client.put(
+        f"/api/avatar/{context['avatar'].id}/permission-settings",
+        headers=context["owner_headers"],
+        json={"permissions": ["chat", "takeover"]},
+    )
+    needs_login = client.get(endpoint, headers=context["owner_headers"]).json()["data"]
+    assert needs_login["enabled"] is True
+    assert needs_login["status"] == "needs_login"
+    assert "BOXIM" in needs_login["message"]
+
+    assert client.get(endpoint).status_code == 401
+    assert client.get(endpoint, headers=context["other_headers"]).status_code == 403
+
+
+def test_takeover_status_reports_ready_pending_count_and_errors(authorization_context):
+    context = authorization_context
+    avatar_id = context["avatar"].id
+    endpoint = f"/api/avatar/{avatar_id}/takeover/status"
+    client.put(
+        f"/api/avatar/{avatar_id}/permission-settings",
+        headers=context["owner_headers"],
+        json={"permissions": ["chat", "takeover"]},
+    )
+
+    db = SessionLocal()
+    try:
+        owner = db.query(User).filter(User.id == context["owner"].id).one()
+        owner.huihui_token = "production-login-token"
+        cursor = TakeoverCursor(
+            avatar_id=avatar_id,
+            owner_id=owner.huihui_user_id,
+            boxim_owner_id="100",
+            last_message_id="10",
+            initialized=True,
+            last_polled_at=datetime.utcnow(),
+        )
+        task = TakeoverReplyTask(
+            avatar_id=avatar_id,
+            owner_id=owner.huihui_user_id,
+            peer_id="200",
+            trigger_message_id="11",
+            source_message_ids=["11"],
+            prompt="你好",
+            status="pending",
+            scheduled_at=datetime.utcnow(),
+            boxim_local_id="123",
+        )
+        db.add_all([cursor, task])
+        db.commit()
+    finally:
+        db.close()
+
+    ready = client.get(endpoint, headers=context["owner_headers"]).json()["data"]
+    assert ready["status"] == "ready"
+    assert ready["pendingCount"] == 1
+    assert ready["lastPolledAt"]
+
+    db = SessionLocal()
+    try:
+        cursor = db.query(TakeoverCursor).filter(TakeoverCursor.avatar_id == avatar_id).one()
+        cursor.last_error = "BOXIM 暂时不可用"
+        db.commit()
+    finally:
+        db.close()
+
+    failed = client.get(endpoint, headers=context["owner_headers"]).json()["data"]
+    assert failed["status"] == "error"
+    assert failed["message"] == "BOXIM 暂时不可用"

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

@@ -1,239 +1,83 @@
-"""Tests for the scheduled takeover message polling."""
+"""Tests for the BOXIM takeover scheduler lifecycle."""
-import json
+
-import pytest
+from unittest.mock import AsyncMock, MagicMock, patch
-from unittest.mock import MagicMock, patch, AsyncMock
 
 
-def test_app_has_startup_event():
+def test_app_has_startup_and_shutdown_events():
-    """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
 
-
+    assert app.router.on_startup
-@patch("services.takeover_service.TakeoverService")
+    assert app.router.on_shutdown
-@patch("services.boxim_client.BoxIMClient")
-@patch("main.redis_lib.from_url")
-@patch("main.AsyncIOScheduler")
-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.poll_and_process_messages = AsyncMock()
-    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()
-    scheduled_callable = mock_scheduler_class.return_value.add_job.call_args.args[0]
-    call_kwargs = mock_scheduler_class.return_value.add_job.call_args[1]
-    assert scheduled_callable is mock_takeover.poll_and_process_messages
-    assert call_kwargs["id"] == "takeover_message_poll"
-    mock_scheduler_class.return_value.start.assert_called_once_with()
 
 
 @patch("services.takeover_service.TakeoverService")
 @patch("services.boxim_client.BoxIMClient")
 @patch("main.AsyncIOScheduler")
-def test_scheduler_starts_without_redis(mock_scheduler_class, mock_boxim_cls, mock_takeover_cls):
+def test_scheduler_uses_boxim_and_restart_safe_service(
-    """App should start even when REDIS_URL is not set."""
+    mock_scheduler_class,
-    mock_boxim = MagicMock()
+    mock_boxim_class,
-    mock_boxim_cls.return_value = mock_boxim
+    mock_takeover_class,
-
+):
-    mock_takeover = MagicMock()
-    mock_takeover.poll_and_process_messages = AsyncMock()
-    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.AsyncIOScheduler")
-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.poll_and_process_messages = AsyncMock()
-    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.AsyncIOScheduler")
-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
-
-
-def test_scheduler_shutdown_releases_resources():
-    """Shutdown should stop polling and close its dedicated database session."""
     import main
 
-    mock_scheduler = MagicMock()
+    scheduler = MagicMock()
-    mock_scheduler.running = True
+    mock_scheduler_class.return_value = scheduler
-    mock_db = MagicMock()
+    boxim = MagicMock()
-    main.takeover_scheduler = mock_scheduler
+    mock_boxim_class.return_value = boxim
-    main.takeover_db = mock_db
+    takeover = MagicMock()
+    takeover.poll_and_process_messages = AsyncMock()
+    mock_takeover_class.return_value = takeover
+
+    environment = {
+        "HUIHUI_PLATFORM_BASE_URL": "https://open.example/api",
+        "BOXIM_API_BASE_URL": "https://im.example/api",
+        "HUIHUI_APP_ID": "app-id",
+        "HUIHUI_ACCESS_ID": "access-id",
+        "HUIHUI_ACCESS_SECRET": "secret",
+        "BOXIM_POLL_INTERVAL_SECONDS": "1",
+    }
+    with patch("main.init_db"), patch("main.seed"), patch.dict(
+        "os.environ", environment, clear=False
+    ):
+        main.on_startup()
+
+    config = mock_boxim_class.call_args.args[0]
+    assert config["HUIHUI_PLATFORM_BASE_URL"] == "https://open.example/api"
+    assert config["BOXIM_API_BASE_URL"] == "https://im.example/api"
+    mock_takeover_class.assert_called_once_with(main.SessionLocal, boxim)
+
+    scheduler.add_job.assert_called_once()
+    scheduled_callable = scheduler.add_job.call_args.args[0]
+    job_options = scheduler.add_job.call_args.kwargs
+    assert scheduled_callable is takeover.poll_and_process_messages
+    assert job_options["id"] == "takeover_message_poll"
+    assert job_options["trigger"].interval.total_seconds() == 1
+    assert job_options["max_instances"] == 1
+    assert job_options["coalesce"] is True
+    scheduler.start.assert_called_once_with()
+
+    main.takeover_scheduler = None
+
+
+@patch("main.AsyncIOScheduler")
+def test_scheduler_failure_does_not_stop_the_api(mock_scheduler_class):
+    import main
+
+    mock_scheduler_class.side_effect = RuntimeError("scheduler crash")
+    with patch("main.init_db"), patch("main.seed"):
+        main.on_startup()
+
+    assert main.takeover_scheduler is None
+
+
+def test_shutdown_stops_only_the_scheduler():
+    import main
+
+    scheduler = MagicMock()
+    scheduler.running = True
+    main.takeover_scheduler = scheduler
 
     main.on_shutdown()
 
-    mock_scheduler.shutdown.assert_called_once_with(wait=False)
+    scheduler.shutdown.assert_called_once_with(wait=False)
-    mock_db.close.assert_called_once_with()
     assert main.takeover_scheduler is None
-    assert main.takeover_db is None
-
-
-# --- 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, MagicMock())
-    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_delayed_mode_without_redis_falls_back_immediately(mock_db, mock_boxim, mock_auth):
-    """A missing Redis connection must not silently drop delayed replies."""
-    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(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_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

@@ -1,322 +1,196 @@
-"""Tests for the TakeoverService — message listening, decision, reply execution."""
+"""End-to-end service tests for BOXIM takeover timing and human priority."""
+
+from datetime import datetime, timedelta, timezone
+from unittest.mock import patch
+
 import pytest
-from unittest.mock import AsyncMock, patch, MagicMock
+from sqlalchemy import create_engine
-from services.takeover_service import TakeoverService
+from sqlalchemy.orm import sessionmaker
-from models import Authorization, Avatar
+from sqlalchemy.pool import StaticPool
+
+from database import Base
+from models import Avatar, TakeoverCursor, TakeoverMessage, TakeoverReplyTask, User
+from services.takeover_service import TakeoverService, _plain_text_reply
+
+
+class Clock:
+    def __init__(self):
+        self.value = datetime(2026, 8, 19, 10, 0, 0)
+
+    def now(self):
+        return self.value
+
+    def advance(self, seconds: int):
+        self.value += timedelta(seconds=seconds)
+
+    def millis(self):
+        return int(self.value.replace(tzinfo=timezone.utc).timestamp() * 1000)
+
+
+class FakeBoxIM:
+    def __init__(self):
+        self.messages = []
+        self.sent = []
+
+    async def exchange_access_token(self, huihui_token):
+        assert huihui_token == "prod-huihui-token"
+        return {"accessToken": "box-token", "accessTokenExpiresIn": 3600}
+
+    async def get_self(self, access_token):
+        assert access_token == "box-token"
+        return {"id": 100}
+
+    async def fetch_private_messages(self, access_token, min_id="0"):
+        assert access_token == "box-token"
+        return [item.copy() for item in self.messages if int(item["id"]) > int(min_id)]
+
+    async def send_private_message(self, access_token, peer_id, content, *, local_id=None):
+        self.sent.append({"peerId": str(peer_id), "content": content, "localId": str(local_id)})
+        return {"id": 900 + len(self.sent), "localId": int(local_id)}
 
 
 @pytest.fixture
-def mock_db():
+def service_context():
-    db = MagicMock()
+    engine = create_engine(
-    return db
+        "sqlite://",
+        connect_args={"check_same_thread": False},
+        poolclass=StaticPool,
+    )
+    session_factory = sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)
+    Base.metadata.create_all(engine)
+    db = session_factory()
+    user = User(
+        id="owner-local",
+        huihui_user_id="owner-huihui",
+        huihui_token="prod-huihui-token",
+        app_token="app-token",
+    )
+    avatar = Avatar(
+        id="avatar-1",
+        owner_id=user.huihui_user_id,
+        name="分身",
+        status="active",
+        config={"authorizationPermissions": ["chat", "takeover"]},
+    )
+    db.add_all([user, avatar])
+    db.commit()
+    db.close()
 
-
+    clock = Clock()
-@pytest.fixture
+    boxim = FakeBoxIM()
-def mock_boxim():
+    service = TakeoverService(session_factory, boxim, now=clock.now)
-    client = AsyncMock()
+    return session_factory, service, boxim, clock
-    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"
-    auth.target_type = "user"
-    auth.status = "active"
-    auth.permissions = ["chat", "takeover"]
-    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
-    disabled_auth.permissions = []
-    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):
+async def test_first_sync_primes_cursor_without_replying_to_history(service_context):
-    mock_db = MagicMock()
+    session_factory, service, boxim, clock = service_context
-    with patch("services.takeover_service.httpx.AsyncClient") as mock_client_class:
+    boxim.messages = [
-        mock_response = MagicMock()
+        {"id": 10, "localId": 1, "sendId": 200, "recvId": 100, "sendTime": clock.millis(), "type": 0, "content": "旧消息"}
-        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)
+    with patch("routers.chat._resolve_reply", return_value={"answer": "不应发送"}):
-        result = await service.generate_reply("avatar_123", "Hello")
+        await service.poll_and_process_messages()
-        assert result == "Hello back"
+
+    db = session_factory()
+    try:
+        cursor = db.query(TakeoverCursor).one()
+        assert cursor.initialized is True
+        assert cursor.last_message_id == "10"
+        assert db.query(TakeoverMessage).count() == 1
+        assert db.query(TakeoverReplyTask).count() == 0
+        assert boxim.sent == []
+    finally:
+        db.close()
 
 
 @pytest.mark.asyncio
-async def test_generate_reply_handles_empty_answer(mock_boxim):
+async def test_incoming_message_is_prepared_then_sent_at_three_seconds(service_context):
-    """generate_reply should return empty string when answer is missing."""
+    session_factory, service, boxim, clock = service_context
-    mock_db = MagicMock()
+    await service.poll_and_process_messages()
-    with patch("services.takeover_service.httpx.AsyncClient") as mock_client_class:
+    boxim.messages.append(
-        mock_response = MagicMock()
+        {"id": 11, "localId": 2, "sendId": 200, "recvId": 100, "sendTime": clock.millis(), "type": 0, "content": "你好"}
-        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)
+    with patch("routers.chat._resolve_reply", return_value={"answer": "**你好**\n\n很高兴见到你"}):
-        result = await service.generate_reply("avatar_123", "Hello")
+        await service.poll_and_process_messages()
-        assert result == ""
+    assert boxim.sent == []
+
+    clock.advance(2)
+    await service.poll_and_process_messages()
+    assert boxim.sent == []
+
+    clock.advance(1)
+    await service.poll_and_process_messages()
+    assert boxim.sent == [{"peerId": "200", "content": "你好\n很高兴见到你", "localId": boxim.sent[0]["localId"]}]
+
+    db = session_factory()
+    try:
+        task = db.query(TakeoverReplyTask).one()
+        assert task.status == "sent"
+        assert task.sent_at == clock.now()
+    finally:
+        db.close()
 
 
 @pytest.mark.asyncio
-async def test_generate_reply_handles_error_code(mock_boxim):
+async def test_owner_message_cancels_pending_reply(service_context):
-    """generate_reply should return empty string when API returns error code."""
+    session_factory, service, boxim, clock = service_context
-    mock_db = MagicMock()
+    await service.poll_and_process_messages()
-    with patch("services.takeover_service.httpx.AsyncClient") as mock_client_class:
+    boxim.messages.append(
-        mock_response = MagicMock()
+        {"id": 21, "localId": 3, "sendId": 200, "recvId": 100, "sendTime": clock.millis(), "type": 0, "content": "在吗"}
-        mock_response.json.return_value = {"code": 500, "message": "Internal error"}
+    )
-        mock_client_class.return_value.__aenter__.return_value.post.return_value = mock_response
+    with patch("routers.chat._resolve_reply", return_value={"answer": "在的"}):
+        await service.poll_and_process_messages()
 
-        service = TakeoverService(mock_db, mock_boxim)
+    clock.advance(2)
-        result = await service.generate_reply("avatar_123", "Hello")
+    boxim.messages.append(
-        assert result == ""
+        {"id": 22, "localId": 4, "sendId": 100, "recvId": 200, "sendTime": clock.millis(), "type": 0, "content": "我来回复"}
+    )
+    await service.poll_and_process_messages()
+    clock.advance(2)
+    await service.poll_and_process_messages()
 
-
+    db = session_factory()
-# --- execute_takeover ---
+    try:
+        task = db.query(TakeoverReplyTask).filter(TakeoverReplyTask.trigger_message_id == "21").one()
+        assert task.status == "cancelled"
+        assert task.cancel_reason == "owner_replied"
+        assert boxim.sent == []
+    finally:
+        db.close()
 
 
 @pytest.mark.asyncio
-async def test_execute_takeover_success(mock_db, mock_boxim, mock_auth, mock_avatar):
+async def test_quick_successive_messages_are_coalesced_into_one_reply(service_context):
-    avatar_filter = MagicMock()
+    session_factory, service, boxim, clock = service_context
-    avatar_filter.first.return_value = mock_avatar
+    await service.poll_and_process_messages()
-    avatar_query = MagicMock()
+    boxim.messages.append(
-    avatar_query.filter.return_value = avatar_filter
+        {"id": 31, "localId": 5, "sendId": 200, "recvId": 100, "sendTime": clock.millis(), "type": 0, "content": "第一句"}
-    mock_db.query.return_value = avatar_query
+    )
+    with patch("routers.chat._resolve_reply", return_value={"answer": "第一版"}):
+        await service.poll_and_process_messages()
 
-    with patch("services.takeover_service.httpx.AsyncClient") as mock_client_class:
+    clock.advance(1)
-        mock_response = MagicMock()
+    boxim.messages.append(
-        mock_response.json.return_value = {"code": 200, "data": {"answer": "Hello back"}}
+        {"id": 32, "localId": 6, "sendId": 200, "recvId": 100, "sendTime": clock.millis(), "type": 0, "content": "第二句"}
-        mock_client_class.return_value.__aenter__.return_value.post.return_value = mock_response
+    )
+    with patch("routers.chat._resolve_reply", return_value={"answer": "合并回复"}) as resolver:
+        await service.poll_and_process_messages()
+    assert resolver.call_args.args[2] == "第一句\n第二句"
 
-        service = TakeoverService(mock_db, mock_boxim)
+    clock.advance(3)
-        message = {"from_accid": "user_acc", "content": "Hello"}
+    await service.poll_and_process_messages()
+    assert [item["content"] for item in boxim.sent] == ["合并回复"]
 
-        result = await service.execute_takeover(mock_auth, message)
+    db = session_factory()
-
+    try:
-        assert result is True
+        tasks = db.query(TakeoverReplyTask).order_by(TakeoverReplyTask.created_at).all()
-        mock_boxim.get_credentials.assert_called_once_with("owner_huihui_123")
+        assert [task.status for task in tasks] == ["cancelled", "sent"]
-        mock_boxim.send_p2p_message.assert_called_once()
+        assert tasks[0].cancel_reason == "newer_incoming_message"
+    finally:
+        db.close()
 
 
-@pytest.mark.asyncio
+def test_plain_text_reply_removes_markdown_and_empty_lines():
-async def test_execute_takeover_fails_when_avatar_not_found(mock_db, mock_boxim, mock_auth):
+    assert _plain_text_reply("## 建议\n\n**不能自行用药**\n`必要时就医`") == "建议\n不能自行用药\n必要时就医"
-    """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

@@ -145,7 +145,7 @@ export const chargeToken = (planId: string) =>
 
 // ==================== 授权管理 API ====================
 
-export type AvatarPermission = 'friend' | 'chat' | 'publish' | 'browse' | 'interact'
+export type AvatarPermission = 'friend' | 'chat' | 'publish' | 'browse' | 'interact' | 'takeover'
 
 export interface AvatarPermissionSettings {
   avatarId: string
@@ -158,6 +158,17 @@ export const getAvatarPermissionSettings = (avatarId: string) =>
 export const updateAvatarPermissionSettings = (avatarId: string, permissions: AvatarPermission[]) =>
   request.put<AvatarPermissionSettings>(`/avatar/${avatarId}/permission-settings`, { permissions })
 
+export interface TakeoverStatus {
+  enabled: boolean
+  status: 'disabled' | 'connecting' | 'ready' | 'needs_login' | 'error'
+  message: string
+  pendingCount: number
+  lastPolledAt: string | null
+}
+
+export const getTakeoverStatus = (avatarId: string) =>
+  request.get<TakeoverStatus>(`/avatar/${avatarId}/takeover/status`)
+
 export interface Authorization {
   id: string
   avatarId: string

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

@@ -27,6 +27,7 @@
             v-for="item in permissionItems"
             :key="item.key"
             class="permission-row"
+            :class="{ takeover: item.key === 'takeover' }"
             type="button"
             role="switch"
             :aria-checked="permissionState[item.key]"
@@ -51,14 +52,18 @@
                 <rect x="4" y="14" width="6" height="6" rx="1" />
                 <rect x="14" y="14" width="6" height="6" rx="1" />
               </svg>
-              <svg v-else viewBox="0 0 24 24">
+              <svg v-else-if="item.key === 'interact'" viewBox="0 0 24 24">
                 <path d="M7 10v10H4a2 2 0 0 1-2-2v-6a2 2 0 0 1 2-2h3Zm0 10h9.5a2 2 0 0 0 1.9-1.4l2.3-7A2 2 0 0 0 18.8 9H14l.8-3.2A2.3 2.3 0 0 0 12.6 3L7 10v10Z" />
               </svg>
+              <svg v-else viewBox="0 0 24 24">
+                <path d="M20 13.5a4.5 4.5 0 0 1-4.5 4.5H9l-5 3 1.4-4.5A6.5 6.5 0 0 1 3 11.4V8.5A4.5 4.5 0 0 1 7.5 4H14" />
+                <path d="M18 3v5m-2.5-2.5h5" />
+              </svg>
             </span>
 
             <span class="permission-copy">
               <strong>{{ item.title }}</strong>
-              <small>{{ item.description }}</small>
+              <small>{{ item.key === 'takeover' ? takeoverDescription : item.description }}</small>
             </span>
 
             <span class="permission-switch" :class="{ active: permissionState[item.key] }" aria-hidden="true">
@@ -96,12 +101,14 @@
 </template>
 
 <script setup lang="ts">
-import { onMounted, reactive, ref } from 'vue'
+import { computed, onMounted, reactive, ref } from 'vue'
 import { useRouter } from 'vue-router'
 import {
   getAvatarPermissionSettings,
+  getTakeoverStatus,
   updateAvatarPermissionSettings,
   type AvatarPermission,
+  type TakeoverStatus,
 } from '@/api'
 import { useAvatarStore } from '@/store/avatar'
 import { pickAvatarId } from '@/utils/avatar-page-data.js'
@@ -147,6 +154,12 @@ const permissionItems: Array<{
     description: '点赞、收藏、评论、回复等操作',
     tone: 'pink',
   },
+  {
+    key: 'takeover',
+    title: '分身主动接管聊天回复',
+    description: '收到私聊消息 3 秒后回复，主人发言时暂停',
+    tone: 'cyan',
+  },
 ]
 
 const permissionState = reactive<PermissionState>({
@@ -155,14 +168,21 @@ const permissionState = reactive<PermissionState>({
   publish: false,
   browse: false,
   interact: false,
+  takeover: false,
 })
 const activeAvatarId = ref('')
 const loading = ref(true)
 const saving = ref(false)
 const errorMessage = ref('')
 const toastMessage = ref('')
+const takeoverStatus = ref<TakeoverStatus | null>(null)
 let toastTimer: number | undefined
 
+const takeoverDescription = computed(() => {
+  if (!permissionState.takeover) return '收到私聊消息 3 秒后回复，主人发言时暂停'
+  return takeoverStatus.value?.message || '正在连接 BOXIM'
+})
+
 const setPermissions = (permissions: AvatarPermission[]) => {
   const enabled = new Set(permissions)
   for (const item of permissionItems) permissionState[item.key] = enabled.has(item.key)
@@ -196,6 +216,7 @@ const loadSettings = async () => {
   try {
     const settings = await getAvatarPermissionSettings(activeAvatarId.value)
     setPermissions(settings.permissions || [])
+    await loadTakeoverStatus()
   } catch (error: any) {
     errorMessage.value = readableError(error, '授权设置加载失败，请稍后重试')
   } finally {
@@ -203,6 +224,15 @@ const loadSettings = async () => {
   }
 }
 
+const loadTakeoverStatus = async () => {
+  if (!activeAvatarId.value) return
+  try {
+    takeoverStatus.value = await getTakeoverStatus(activeAvatarId.value)
+  } catch {
+    takeoverStatus.value = null
+  }
+}
+
 const saveSettings = async () => {
   if (!activeAvatarId.value || loading.value || saving.value) return
   saving.value = true
@@ -210,6 +240,8 @@ const saveSettings = async () => {
   try {
     const settings = await updateAvatarPermissionSettings(activeAvatarId.value, selectedPermissions())
     setPermissions(settings.permissions || [])
+    await loadTakeoverStatus()
+    if (permissionState.takeover) window.setTimeout(loadTakeoverStatus, 1400)
     showToast('授权设置已保存')
   } catch (error: any) {
     errorMessage.value = readableError(error, '授权设置保存失败，请重试')
@@ -363,6 +395,10 @@ svg {
   background: #fcfcfd;
 }
 
+.permission-row.takeover {
+  min-height: 76px;
+}
+
 .permission-icon {
   width: 34px;
   height: 34px;
@@ -400,6 +436,11 @@ svg {
   background: #fff3f7;
 }
 
+.permission-icon.cyan {
+  color: #35a6a2;
+  background: #effaf9;
+}
+
 .permission-copy {
   min-width: 0;
 }
@@ -427,6 +468,15 @@ svg {
   white-space: nowrap;
 }
 
+.permission-row.takeover .permission-copy small {
+  display: -webkit-box;
+  overflow: hidden;
+  text-overflow: ellipsis;
+  white-space: normal;
+  -webkit-box-orient: vertical;
+  -webkit-line-clamp: 2;
+}
+
 .permission-switch {
   position: relative;
   width: 43px;