diff --git a/digital-avatar-app/backend/services/takeover_service.py b/digital-avatar-app/backend/services/takeover_service.py
new file mode 100644
index 0000000..2316c61
--- /dev/null
+++ b/digital-avatar-app/backend/services/takeover_service.py
@@ -0,0 +1,111 @@
+"""Takeover service — message listening, decision, reply execution."""
+import json
+import logging
+import os
+from typing import Optional
+
+import httpx
+from sqlalchemy.orm import Session
+
+from models import Authorization
+from services.boxim_client import BoxIMClient
+
+logger = logging.getLogger(__name__)
+
+
+class TakeoverService:
+    """Service for handling avatar takeover — generating replies and sending them via IM."""
+
+    def __init__(
+        self,
+        db: Session,
+        boxim_client: BoxIMClient,
+        redis_client=None,
+    ):
+        self.db = db
+        self.boxim = boxim_client
+        self.redis = redis_client
+        self._chat_api_base = os.getenv(
+            "TAKEOVER_CHAT_API_BASE", "http://localhost:8000/api"
+        )
+
+    def check_takeover_enabled(
+        self, owner_huihui_id: str, from_user_id: str
+    ) -> Optional[Authorization]:
+        """Check whether takeover is enabled for the given target user."""
+        auth = (
+            self.db.query(Authorization)
+            .filter(Authorization.target_id == from_user_id)
+            .filter(Authorization.takeover_enabled == True)
+            .first()
+        )
+        if auth and auth.takeover_enabled:
+            return auth
+        return None
+
+    async def generate_reply(self, avatar_id: str, message: str) -> str:
+        """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:
+            owner_huihui_id = auth.owner_id if hasattr(auth, "owner_id") else ""
+            credentials = await self.boxim.get_credentials(owner_huihui_id)
+            if not credentials:
+                logger.warning(f"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
+        key = f"takeover:delayed:{auth.target_id}:{message.get('msg_id', '')}"
+        value = json.dumps({
+            "avatar_id": auth.avatar_id,
+            "from_accid": message.get("from_accid", ""),
+            "content": message.get("content", ""),
+            "owner_huihui_id": auth.owner_id if hasattr(auth, "owner_id") else "",
+        })
+        self.redis.setex(key, auth.takeover_delay_seconds + 10, value)
+        logger.info(f"Message enqueued to delayed queue: {key}")
+
+    async def process_delayed_queue(self):
+        """Process expired messages from the delayed queue.
+
+        Redis handles expiration automatically; this is a placeholder
+        for a future scheduled worker that scans and dispatches.
+        """
+        if not self.redis:
+            return
diff --git a/digital-avatar-app/backend/tests/test_takeover_service.py b/digital-avatar-app/backend/tests/test_takeover_service.py
new file mode 100644
index 0000000..49afc11
--- /dev/null
+++ b/digital-avatar-app/backend/tests/test_takeover_service.py
@@ -0,0 +1,136 @@
+"""Tests for the TakeoverService — message listening, decision, reply execution."""
+import pytest
+from unittest.mock import AsyncMock, patch, MagicMock
+from services.takeover_service import TakeoverService
+from models import Authorization
+
+
+@pytest.fixture
+def mock_db():
+    db = MagicMock()
+    return db
+
+
+@pytest.fixture
+def mock_boxim():
+    client = AsyncMock()
+    client.get_credentials.return_value = {"accid": "owner_acc", "token": "tok"}
+    client.send_p2p_message.return_value = True
+    return client
+
+
+@pytest.fixture
+def mock_auth():
+    auth = MagicMock(spec=Authorization)
+    auth.takeover_enabled = True
+    auth.takeover_mode = "immediate"
+    auth.takeover_delay_seconds = 30
+    auth.avatar_id = "avatar_123"
+    auth.target_id = "target_user_123"
+    auth.owner_id = "owner_huihui_123"
+    return auth
+
+
+def test_check_takeover_enabled_returns_auth_when_enabled(mock_db, mock_auth, mock_boxim):
+    mock_db.query.return_value.filter.return_value.filter.return_value.first.return_value = mock_auth
+    service = TakeoverService(mock_db, mock_boxim)
+    result = service.check_takeover_enabled("owner_123", "target_123")
+    assert result == mock_auth
+
+
+def test_check_takeover_enabled_returns_none_when_disabled(mock_db, mock_boxim):
+    disabled_auth = MagicMock(spec=Authorization)
+    disabled_auth.takeover_enabled = False
+    mock_db.query.return_value.filter.return_value.filter.return_value.first.return_value = disabled_auth
+    service = TakeoverService(mock_db, mock_boxim)
+    result = service.check_takeover_enabled("owner_123", "target_123")
+    assert result is None
+
+
+@pytest.mark.asyncio
+async def test_generate_reply_returns_answer(mock_boxim):
+    mock_db = MagicMock()
+    with patch("services.takeover_service.httpx.AsyncClient") as mock_client_class:
+        mock_response = MagicMock()
+        mock_response.json.return_value = {"code": 200, "data": {"answer": "Hello back"}}
+        mock_client_class.return_value.__aenter__.return_value.post.return_value = mock_response
+
+        service = TakeoverService(mock_db, mock_boxim)
+        result = await service.generate_reply("avatar_123", "Hello")
+        assert result == "Hello back"
+
+
+@pytest.mark.asyncio
+async def test_execute_takeover_success(mock_db, mock_boxim, mock_auth):
+    with patch("services.takeover_service.httpx.AsyncClient") as mock_client_class:
+        mock_response = MagicMock()
+        mock_response.json.return_value = {"code": 200, "data": {"answer": "Hello back"}}
+        mock_client_class.return_value.__aenter__.return_value.post.return_value = mock_response
+
+        service = TakeoverService(mock_db, mock_boxim)
+        message = {"from_accid": "user_acc", "content": "Hello"}
+
+        result = await service.execute_takeover(mock_auth, message)
+
+        assert result is True
+        mock_boxim.get_credentials.assert_called_once()
+        mock_boxim.send_p2p_message.assert_called_once()
+
+
+def test_enqueue_delayed_message_with_redis(mock_db, mock_boxim, mock_auth):
+    mock_redis = MagicMock()
+    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()
+
+
+def test_enqueue_delayed_message_without_redis_logs_warning(mock_db, mock_boxim, mock_auth):
+    """When Redis is not configured, enqueue_delayed_message should log a warning and not crash."""
+    service = TakeoverService(mock_db, mock_boxim)
+    message = {"msg_id": "msg_1", "from_accid": "user_acc", "content": "Hello"}
+
+    # Should not raise
+    service.enqueue_delayed_message(mock_auth, message)
+
+
+@pytest.mark.asyncio
+async def test_execute_takeover_fails_when_no_credentials(mock_db, mock_boxim, mock_auth):
+    """execute_takeover should return False when boxim.get_credentials returns None."""
+    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
+
+
+@pytest.mark.asyncio
+async def test_generate_reply_handles_empty_answer(mock_boxim):
+    """generate_reply should return empty string when answer is missing."""
+    mock_db = MagicMock()
+    with patch("services.takeover_service.httpx.AsyncClient") as mock_client_class:
+        mock_response = MagicMock()
+        mock_response.json.return_value = {"code": 200, "data": {}}
+        mock_client_class.return_value.__aenter__.return_value.post.return_value = mock_response
+
+        service = TakeoverService(mock_db, mock_boxim)
+        result = await service.generate_reply("avatar_123", "Hello")
+        assert result == ""
+
+
+@pytest.mark.asyncio
+async def test_generate_reply_handles_error_code(mock_boxim):
+    """generate_reply should return empty string when API returns error code."""
+    mock_db = MagicMock()
+    with patch("services.takeover_service.httpx.AsyncClient") as mock_client_class:
+        mock_response = MagicMock()
+        mock_response.json.return_value = {"code": 500, "message": "Internal error"}
+        mock_client_class.return_value.__aenter__.return_value.post.return_value = mock_response
+
+        service = TakeoverService(mock_db, mock_boxim)
+        result = await service.generate_reply("avatar_123", "Hello")
+        assert result == ""