fix(avatar): fail closed on BOXIM connection errors

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

@@ -1,7 +1,7 @@
 """End-to-end service tests for BOXIM takeover timing and human priority."""
 
 from datetime import datetime, timedelta, timezone
-from unittest.mock import patch
+from unittest.mock import AsyncMock, patch
 
 import pytest
 from sqlalchemy import create_engine
@@ -10,6 +10,7 @@ from sqlalchemy.pool import StaticPool
 
 from database import Base
 from models import Avatar, TakeoverCursor, TakeoverMessage, TakeoverReplyTask, User
+from services.boxim_client import BoxIMError
 from services.takeover_service import TakeoverService, _plain_text_reply
 
 
@@ -192,5 +193,28 @@ async def test_quick_successive_messages_are_coalesced_into_one_reply(service_co
         db.close()
 
 
+@pytest.mark.asyncio
+async def test_connection_failure_disables_takeover_and_stops_retrying(service_context):
+    session_factory, service, boxim, _ = service_context
+    boxim.exchange_access_token = AsyncMock(
+        side_effect=BoxIMError("无效的访问令牌", code=40101, auth_error=True)
+    )
+
+    await service.poll_and_process_messages()
+    await service.poll_and_process_messages()
+
+    db = session_factory()
+    try:
+        avatar = db.query(Avatar).one()
+        cursor = db.query(TakeoverCursor).one()
+        assert "takeover" not in avatar.config["authorizationPermissions"]
+        assert cursor.initialized is False
+        assert "重新登录" in cursor.last_error
+        assert db.query(TakeoverReplyTask).count() == 0
+    finally:
+        db.close()
+    boxim.exchange_access_token.assert_awaited_once_with("prod-huihui-token")
+
+
 def test_plain_text_reply_removes_markdown_and_empty_lines():
     assert _plain_text_reply("## 建议\n\n**不能自行用药**\n`必要时就医`") == "建议\n不能自行用药\n必要时就医"