diff --git a/app/application/messaging/interaction.py b/app/application/messaging/interaction.py
index 90f13ef82..0c2b1657a 100644
--- a/app/application/messaging/interaction.py
+++ b/app/application/messaging/interaction.py
@@ -3,13 +3,102 @@ import uuid
 from dataclasses import dataclass, field
 from datetime import datetime, timedelta
 from threading import Lock
-from typing import Any, Dict, List, Optional, Protocol, Sequence, Tuple, Union
+from typing import (
+    Any,
+    Dict,
+    Generic,
+    List,
+    Optional,
+    Protocol,
+    Sequence,
+    Tuple,
+    TypeVar,
+    Union,
+)
 
 from app.schemas.message import Message
 from app.schemas.notification import ChannelCapabilityManager
 from app.schemas.types import NotificationChannel
 
 
+class _UserInteractionEntry(Protocol):
+    """声明按用户维护的交互状态所需的最小字段。"""
+
+    request_id: str
+    user_id: str
+    created_at: datetime
+
+
+InteractionEntryT = TypeVar("InteractionEntryT", bound=_UserInteractionEntry)
+
+
+class _ExpiringUserInteractionStore(Generic[InteractionEntryT]):
+    """提供单用户单会话的过期状态存储和双向索引。"""
+
+    _ttl = timedelta(hours=24)
+
+    def __init__(self) -> None:
+        """初始化按请求 ID、用户 ID 查询的线程安全状态表。"""
+        self._by_id: Dict[str, InteractionEntryT] = {}
+        self._by_user: Dict[str, str] = {}
+        self._lock = Lock()
+
+    def _cleanup_locked(self) -> None:
+        """在持锁状态下移除超过统一保留时间的交互状态。"""
+        expire_before = datetime.now() - self._ttl
+        expired = [
+            request_id
+            for request_id, request in self._by_id.items()
+            if request.created_at < expire_before
+        ]
+        for request_id in expired:
+            request = self._by_id.pop(request_id, None)
+            if request:
+                self._by_user.pop(str(request.user_id), None)
+
+    def _replace_locked(self, request: InteractionEntryT) -> None:
+        """在持锁状态下替换用户旧状态并写入新的双向索引。"""
+        user_key = str(request.user_id)
+        old_request_id = self._by_user.get(user_key)
+        if old_request_id:
+            self._by_id.pop(old_request_id, None)
+        self._by_id[request.request_id] = request
+        self._by_user[user_key] = request.request_id
+
+    def get_by_user(self, user_id: Union[str, int]) -> Optional[InteractionEntryT]:
+        """按用户返回当前仍有效的交互状态。"""
+        with self._lock:
+            self._cleanup_locked()
+            request_id = self._by_user.get(str(user_id))
+            if not request_id:
+                return None
+            return self._by_id.get(request_id)
+
+    def get_by_id(
+            self, request_id: str, user_id: Union[str, int]
+    ) -> Optional[InteractionEntryT]:
+        """按请求 ID 查询状态，并校验请求属于当前用户。"""
+        with self._lock:
+            self._cleanup_locked()
+            request = self._by_id.get(request_id)
+            if not request or str(request.user_id) != str(user_id):
+                return None
+            return request
+
+    def remove(self, request_id: str) -> None:
+        """删除指定交互状态及其用户索引。"""
+        with self._lock:
+            request = self._by_id.pop(request_id, None)
+            if request:
+                self._by_user.pop(str(request.user_id), None)
+
+    def clear(self) -> None:
+        """清空全部交互状态，供生命周期结束和测试场景使用。"""
+        with self._lock:
+            self._by_id.clear()
+            self._by_user.clear()
+
+
 @dataclass
 class PendingSlashInteraction:
     """
@@ -27,32 +116,13 @@ class PendingSlashInteraction:
     created_at: datetime = field(default_factory=datetime.now)
 
 
-class SlashInteractionManager:
+class SlashInteractionManager(
+        _ExpiringUserInteractionStore[PendingSlashInteraction]
+):
     """
     管理单个 slash 命令的交互会话。
     """
 
-    _ttl = timedelta(hours=24)
-
-    def __init__(self) -> None:
-        """初始化按请求和用户索引的 slash 会话表。"""
-        self._by_id: Dict[str, PendingSlashInteraction] = {}
-        self._by_user: Dict[str, str] = {}
-        self._lock = Lock()
-
-    def _cleanup_locked(self) -> None:
-        """在持锁状态下移除过期 slash 会话。"""
-        expire_before = datetime.now() - self._ttl
-        expired = [
-            request_id
-            for request_id, request in self._by_id.items()
-            if request.created_at < expire_before
-        ]
-        for request_id in expired:
-            request = self._by_id.pop(request_id, None)
-            if request:
-                self._by_user.pop(str(request.user_id), None)
-
     def create_or_replace(
             self,
             user_id: Union[str, int],
@@ -65,9 +135,6 @@ class SlashInteractionManager:
         with self._lock:
             self._cleanup_locked()
             user_key = str(user_id)
-            old_request_id = self._by_user.get(user_key)
-            if old_request_id:
-                self._by_id.pop(old_request_id, None)
             request = PendingSlashInteraction(
                 request_id=uuid.uuid4().hex[:12],
                 user_id=user_key,
@@ -76,45 +143,9 @@ class SlashInteractionManager:
                 source=source,
                 username=username,
             )
-            self._by_id[request.request_id] = request
-            self._by_user[user_key] = request.request_id
+            self._replace_locked(request)
             return request
 
-    def get_by_user(
-            self, user_id: Union[str, int]
-    ) -> Optional[PendingSlashInteraction]:
-        """按用户返回仍有效的 slash 会话。"""
-        with self._lock:
-            self._cleanup_locked()
-            request_id = self._by_user.get(str(user_id))
-            if not request_id:
-                return None
-            return self._by_id.get(request_id)
-
-    def get_by_id(
-            self, request_id: str, user_id: Union[str, int]
-    ) -> Optional[PendingSlashInteraction]:
-        """按请求 ID 和用户联合校验 slash 会话。"""
-        with self._lock:
-            self._cleanup_locked()
-            request = self._by_id.get(request_id)
-            if not request or str(request.user_id) != str(user_id):
-                return None
-            return request
-
-    def remove(self, request_id: str) -> None:
-        """删除指定 slash 会话及其用户索引。"""
-        with self._lock:
-            request = self._by_id.pop(request_id, None)
-            if request:
-                self._by_user.pop(str(request.user_id), None)
-
-    def clear(self) -> None:
-        """清空全部 slash 会话。"""
-        with self._lock:
-            self._by_id.clear()
-            self._by_user.clear()
-
 
 @dataclass(frozen=True, slots=True)
 class InteractionContext:
diff --git a/app/application/messaging/media.py b/app/application/messaging/media.py
index 547719a97..8142ed635 100644
--- a/app/application/messaging/media.py
+++ b/app/application/messaging/media.py
@@ -1,9 +1,9 @@
 import uuid
 from dataclasses import dataclass, field
-from datetime import datetime, timedelta
-from threading import Lock
+from datetime import datetime
 from typing import Any, Dict, List, Optional, Union
 
+from app.application.messaging.interaction import _ExpiringUserInteractionStore
 from app.domain.context import MediaInfo
 from app.domain.meta.metabase import MetaBase
 from app.schemas.types import NotificationChannel
@@ -36,36 +36,15 @@ class PendingMediaInteraction:
     created_at: datetime = field(default_factory=datetime.now)
 
 
-class MediaInteractionManager:
+class MediaInteractionManager(
+        _ExpiringUserInteractionStore[PendingMediaInteraction]
+):
     """
     管理用户当前激活的媒体交互状态。
 
     每个用户只保留一个有效会话，避免旧按钮与新一轮搜索混用。
     """
 
-    _ttl = timedelta(hours=24)
-
-    def __init__(self):
-        """初始化按请求和用户索引的媒体会话表。"""
-        self._by_id: Dict[str, PendingMediaInteraction] = {}
-        self._by_user: Dict[str, str] = {}
-        self._lock = Lock()
-
-    def _cleanup_locked(self) -> None:
-        """
-        清理超时会话，避免内存中残留旧交互状态。
-        """
-        expire_before = datetime.now() - self._ttl
-        expired = [
-            request_id
-            for request_id, request in self._by_id.items()
-            if request.created_at < expire_before
-        ]
-        for request_id in expired:
-            request = self._by_id.pop(request_id, None)
-            if request:
-                self._by_user.pop(str(request.user_id), None)
-
     def create_or_replace(
             self,
             user_id: Union[str, int],
@@ -84,10 +63,6 @@ class MediaInteractionManager:
         with self._lock:
             self._cleanup_locked()
             user_key = str(user_id)
-            old_request_id = self._by_user.get(user_key)
-            if old_request_id:
-                self._by_id.pop(old_request_id, None)
-
             request = PendingMediaInteraction(
                 request_id=uuid.uuid4().hex[:12],
                 user_id=user_key,
@@ -100,52 +75,8 @@ class MediaInteractionManager:
                 meta=meta,
                 items=list(items or []),
             )
-            self._by_id[request.request_id] = request
-            self._by_user[user_key] = request.request_id
+            self._replace_locked(request)
             return request
 
-    def get_by_user(
-            self, user_id: Union[str, int]
-    ) -> Optional[PendingMediaInteraction]:
-        """
-        按用户读取当前会话，供文本回复和旧按钮兼容使用。
-        """
-        with self._lock:
-            self._cleanup_locked()
-            request_id = self._by_user.get(str(user_id))
-            if not request_id:
-                return None
-            return self._by_id.get(request_id)
-
-    def get_by_id(
-            self, request_id: str, user_id: Union[str, int]
-    ) -> Optional[PendingMediaInteraction]:
-        """
-        按请求 ID 读取会话，并校验用户归属。
-        """
-        with self._lock:
-            self._cleanup_locked()
-            request = self._by_id.get(request_id)
-            if not request or str(request.user_id) != str(user_id):
-                return None
-            return request
-
-    def remove(self, request_id: str) -> None:
-        """
-        主动结束一条会话。
-        """
-        with self._lock:
-            request = self._by_id.pop(request_id, None)
-            if request:
-                self._by_user.pop(str(request.user_id), None)
-
-    def clear(self) -> None:
-        """
-        清空所有交互状态，主要用于测试。
-        """
-        with self._lock:
-            self._by_id.clear()
-            self._by_user.clear()
-
 
 media_interaction_manager = MediaInteractionManager()
diff --git a/app/application/messaging/skill.py b/app/application/messaging/skill.py
index b65b977bd..fd0e642bc 100644
--- a/app/application/messaging/skill.py
+++ b/app/application/messaging/skill.py
@@ -1,11 +1,11 @@
 import re
 import uuid
 from dataclasses import dataclass, field
-from datetime import datetime, timedelta
-from threading import Lock
+from datetime import datetime
 from typing import Any, Callable, Dict, List, Optional, Protocol, Tuple, Union
 
 from app.application.messaging.interaction import (
+    _ExpiringUserInteractionStore,
     MessageGateway,
     build_navigation_buttons,
     page_items,
@@ -92,36 +92,15 @@ class PendingSkillInteraction:
     created_at: datetime = field(default_factory=datetime.now)
 
 
-class SkillInteractionManager:
+class SkillInteractionManager(
+        _ExpiringUserInteractionStore[PendingSkillInteraction]
+):
     """
     管理用户当前的技能交互状态。
 
     每个用户同一时间只保留一个有效会话，避免旧按钮继续生效。
     """
 
-    _ttl = timedelta(hours=24)
-
-    def __init__(self):
-        """初始化按请求和用户索引的技能交互会话表。"""
-        self._by_id: Dict[str, PendingSkillInteraction] = {}
-        self._by_user: Dict[str, str] = {}
-        self._lock = Lock()
-
-    def _cleanup_locked(self):
-        """
-        清理超时会话，避免按钮回调无限积累。
-        """
-        expire_before = datetime.now() - self._ttl
-        expired = [
-            request_id
-            for request_id, request in self._by_id.items()
-            if request.created_at < expire_before
-        ]
-        for request_id in expired:
-            request = self._by_id.pop(request_id, None)
-            if request:
-                self._by_user.pop(str(request.user_id), None)
-
     def create_or_replace(
             self,
             user_id: Union[str, int],
@@ -135,9 +114,6 @@ class SkillInteractionManager:
         with self._lock:
             self._cleanup_locked()
             user_key = str(user_id)
-            old_request_id = self._by_user.get(user_key)
-            if old_request_id:
-                self._by_id.pop(old_request_id, None)
             request_id = uuid.uuid4().hex[:12]
             request = PendingSkillInteraction(
                 request_id=request_id,
@@ -146,53 +122,9 @@ class SkillInteractionManager:
                 source=source,
                 username=username,
             )
-            self._by_id[request_id] = request
-            self._by_user[user_key] = request_id
+            self._replace_locked(request)
             return request
 
-    def get_by_user(
-            self, user_id: Union[str, int]
-    ) -> Optional[PendingSkillInteraction]:
-        """
-        按用户获取当前有效会话，供纯文本回复路由使用。
-        """
-        with self._lock:
-            self._cleanup_locked()
-            request_id = self._by_user.get(str(user_id))
-            if not request_id:
-                return None
-            return self._by_id.get(request_id)
-
-    def get_by_id(
-            self, request_id: str, user_id: Union[str, int]
-    ) -> Optional[PendingSkillInteraction]:
-        """
-        按请求 ID 获取会话，并校验会话归属用户。
-        """
-        with self._lock:
-            self._cleanup_locked()
-            request = self._by_id.get(request_id)
-            if not request or str(request.user_id) != str(user_id):
-                return None
-            return request
-
-    def remove(self, request_id: str) -> None:
-        """
-        主动结束会话，释放用户和请求 ID 的双向索引。
-        """
-        with self._lock:
-            request = self._by_id.pop(request_id, None)
-            if request:
-                self._by_user.pop(str(request.user_id), None)
-
-    def clear(self):
-        """
-        清空所有会话，主要用于测试场景。
-        """
-        with self._lock:
-            self._by_id.clear()
-            self._by_user.clear()
-
 
 skill_interaction_manager = SkillInteractionManager()