refactor(msteams): 整理代码结构并新增多项功能

本次提交对Microsoft Teams适配器代码进行了多维度优化与新增：
1.  调整多处导入顺序，优化代码可读性
2.  新增media_tools工具模块，提供媒体相关辅助函数
3.  新增thread_history模块，实现对话历史拉取与缓存功能
4.  新增connection_modes模块，支持webhook/websocket/polling三种连接模式
5.  扩展security.py与tool_policy.py，新增通配符配置校验与三级策略解析
6.  新增feedback会话记录功能
7.  为sent_message_cache添加自动清理任务
8.  优化normalizer模块，新增引用、编辑消息解析与线程上下文注入
9.  重构file_upload的SSRF防护逻辑，复用公共校验工具
10. 修复多处导入顺序与代码排版问题
11. 为消息发送添加断路器保护与异步去重锁

backend/package/yuxi/channels/adapters/msteams/__init__.py

@@ -15,8 +15,7 @@ from yuxi.channels.adapters.msteams.cards import (
     wrap_as_attachment,
 )
 from yuxi.channels.adapters.msteams.chunking import chunk_text
-from yuxi.channels.adapters.msteams.command_gate import CommandGate
-from yuxi.channels.adapters.msteams.command_gate import resolve_dual_text_control_command_gate
+from yuxi.channels.adapters.msteams.command_gate import CommandGate, resolve_dual_text_control_command_gate
 from yuxi.channels.adapters.msteams.commands import (
     SLASH_COMMANDS,
     build_command_help_card,
@@ -48,11 +47,11 @@ from yuxi.channels.adapters.msteams.feedback import (
 )
 from yuxi.channels.adapters.msteams.file_upload import (
     PendingUploadStore,
+    _validate_upload_url,
     build_file_consent_card,
     build_teams_file_info_card,
     needs_file_consent,
     upload_to_sharepoint,
-    _validate_upload_url,
 )
 from yuxi.channels.adapters.msteams.graph import (
     GraphClient,
@@ -119,10 +118,10 @@ from yuxi.channels.adapters.msteams.secret_input import (
 )
 from yuxi.channels.adapters.msteams.security import SecurityPolicy
 from yuxi.channels.adapters.msteams.send import (
+    SILENT_REPLY_TOKEN,
+    MessageSender,
     MSTeamsErrorCategory,
     MSTeamsErrorInfo,
-    MessageSender,
-    SILENT_REPLY_TOKEN,
     classify_msteams_send_error,
     format_send_error_hint,
     is_silent_reply_text,
@@ -144,12 +143,12 @@ from yuxi.channels.adapters.msteams.sso import (
     is_verify_state,
 )
 from yuxi.channels.adapters.msteams.streaming import (
+    _STREAM_INFORMATIVE_TEXTS,
     ReplyStreamController,
     StreamManager,
     StreamPhase,
     StreamState,
     TeamsHttpStream,
-    _STREAM_INFORMATIVE_TEXTS,
 )
 from yuxi.channels.adapters.msteams.tenant import TenantValidator
 from yuxi.channels.adapters.msteams.tool_policy import ToolPolicy, resolve_tool_policy

backend/package/yuxi/channels/adapters/msteams/adapter.py

@@ -6,6 +6,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import base64
 import logging
 import os
@@ -21,10 +22,11 @@ from jwt import PyJWKClient
 
 from yuxi.channels.base import BaseChannelAdapter
 from yuxi.channels.capabilities import ChannelCapabilities
-from yuxi.channels.meta import ChannelMeta
 from yuxi.channels.exceptions import (
     ChannelAuthenticationError,
 )
+from yuxi.channels.infra.circuit_breaker import CircuitBreaker, CircuitBreakerOpenError
+from yuxi.channels.meta import ChannelMeta
 from yuxi.channels.models import (
     ChannelIdentity,
     ChannelMessage,
@@ -46,16 +48,20 @@ from .credentials import DelegatedAuthStore
 from .debounce import DebounceManager
 from .feedback import build_feedback_channel_data
 from .formatter import format_outbound
-from .normalizer import normalize_inbound, normalize_conversation_update, normalize_invoke
 from .invoke_handler import normalize_reaction
+from .normalizer import normalize_conversation_update, normalize_inbound, normalize_invoke
 from .polls import PollStore
-from .probe import MSTeamsProbe
 from .proactive import (
     ConversationStore,
+)
+from .proactive import (
     proactive_send as _proactive_send,
 )
+from .probe import MSTeamsProbe
 from .security import SecurityPolicy
-from .send import MessageSender, send_adaptive_card as _send_adaptive_card, send_media as _send_stream_media
+from .send import MessageSender
+from .send import send_adaptive_card as _send_adaptive_card
+from .send import send_media as _send_stream_media
 from .sent_message_cache import SentMessageCache
 from .sso import SSOHandler
 from .streaming import StreamManager
@@ -69,6 +75,8 @@ _DEDUP_WINDOW_SECONDS = 300
 _MAX_WEBHOOK_BODY_BYTES = 256 * 1024
 _WEBHOOK_HANDLE_TIMEOUT_SECONDS = 30
 _WEBHOOK_SIGNATURE_TIMEOUT_SECONDS = 10
+MEDIA_GRAPH_UPLOAD_THRESHOLD = 512 * 1024
+MEDIA_BASE64_MAX_SIZE = 1024 * 1024
 
 BOT_FRAMEWORK_DOMAINS = {
     "smba.trafficmanager.net",
@@ -139,6 +147,7 @@ class MSTeamsAdapter(BaseChannelAdapter):
         self._stream_mgr = StreamManager()
         self._dedup_ids: dict[str, float] = {}
         self._dedup_ttl = self.config.get("dedup_ttl", _DEDUP_WINDOW_SECONDS)
+        self._dedup_lock = asyncio.Lock()
         self._http_session: aiohttp.ClientSession | None = None
         self._jwks_client: PyJWKClient | None = None
         self._security_policy = SecurityPolicy(self.config)
@@ -156,6 +165,7 @@ class MSTeamsAdapter(BaseChannelAdapter):
         self._conv_store = ConversationStore()
         self._delegated_auth_store = DelegatedAuthStore()
         self._sent_cache = SentMessageCache()
+        self._circuit_breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=60, channel_id="msteams")
 
     async def _get_http_session(self) -> aiohttp.ClientSession:
         if self._http_session is None or self._http_session.closed:
@@ -218,6 +228,7 @@ class MSTeamsAdapter(BaseChannelAdapter):
 
         self._streaming_mode = self.config.get("streaming_mode", "block")
         self._status = ChannelStatus.CONNECTED
+        self._sent_cache.start_cleanup_task()
         logger.info(f"MSTeams bot '{self._app_id[:8]}...' connected")
 
     async def disconnect(self) -> None:
@@ -240,6 +251,7 @@ class MSTeamsAdapter(BaseChannelAdapter):
 
         await self._conv_store.flush()
         await self._poll_store.flush()
+        self._sent_cache.stop_cleanup_task()
 
         self._jwks_client = None
         self._stream_mgr.clear()
@@ -252,45 +264,53 @@ class MSTeamsAdapter(BaseChannelAdapter):
         if not_connected:
             return not_connected
 
-        conversation_id = self._resolve_conversation_id(response)
-        chunks = chunk_text(response.content, self.text_chunk_limit)
-        if not chunks:
-            return DeliveryResult(success=False, error="Empty content")
+        async def _do_send() -> DeliveryResult:
+            conversation_id = self._resolve_conversation_id(response)
+            chunks = chunk_text(response.content, self.text_chunk_limit)
+            if not chunks:
+                raise ChannelAuthenticationError("Empty content")
 
-        activity = format_outbound(response, self.text_chunk_limit)
-        activity["text"] = chunks[0]
+            activity = format_outbound(response, self.text_chunk_limit)
+            activity["text"] = chunks[0]
 
-        if self._feedback_enabled:
-            fb_channel_data = build_feedback_channel_data(
-                feedback_enabled=self._feedback_enabled,
-                feedback_reflection=self._feedback_reflection,
-            )
-            if fb_channel_data:
-                existing_cd = activity.get("channelData", {}) or {}
-                activity["channelData"] = {**existing_cd, **fb_channel_data}
+            if self._feedback_enabled:
+                fb_channel_data = build_feedback_channel_data(
+                    feedback_enabled=self._feedback_enabled,
+                    feedback_reflection=self._feedback_reflection,
+                )
+                if fb_channel_data:
+                    existing_cd = activity.get("channelData", {}) or {}
+                    activity["channelData"] = {**existing_cd, **fb_channel_data}
+
+            result = await self._sender.send_activity(conversation_id, activity)
+            if not result.success or not result.message_id:
+                raise ChannelAuthenticationError(f"Send failed: {result.error}")
+
+            self._sent_cache.record(result.message_id, conversation_id)
+
+            if len(chunks) <= 1:
+                return result
+
+            accumulated_text = chunks[0]
+            for chunk in chunks[1:]:
+                accumulated_text += chunk
+                if len(accumulated_text) > self.text_chunk_limit:
+                    accumulated_text = accumulated_text[: self.text_chunk_limit]
+                edit_activity = {
+                    "type": "message",
+                    "text": accumulated_text,
+                    "textFormat": "markdown",
+                }
+                await self._sender.update_activity(conversation_id, result.message_id, edit_activity)
 
-        result = await self._sender.send_activity(conversation_id, activity)
-        if not result.success or not result.message_id:
             return result
 
-        self._sent_cache.record(result.message_id, conversation_id)
-
-        if len(chunks) <= 1:
-            return result
-
-        accumulated_text = chunks[0]
-        for chunk in chunks[1:]:
-            accumulated_text += chunk
-            if len(accumulated_text) > self.text_chunk_limit:
-                accumulated_text = accumulated_text[: self.text_chunk_limit]
-            edit_activity = {
-                "type": "message",
-                "text": accumulated_text,
-                "textFormat": "markdown",
-            }
-            await self._sender.update_activity(conversation_id, result.message_id, edit_activity)
-
-        return result
+        try:
+            return await self._circuit_breaker.call(_do_send)
+        except CircuitBreakerOpenError:
+            return DeliveryResult(success=False, error="Circuit breaker open")
+        except Exception as e:
+            return DeliveryResult(success=False, error=str(e))
 
     async def send_media(self, chat_id: str, media_type: str, data: Any) -> DeliveryResult:
         not_connected = self._require_connected()
@@ -306,8 +326,15 @@ class MSTeamsAdapter(BaseChannelAdapter):
         mime_type = content_type_map.get(media_type, "application/octet-stream")
 
         if isinstance(data, bytes):
+            file_size = len(data)
             token = self._sender.token if self._sender else None
-            if token and len(data) > 1024 * 1024:
+
+            if file_size > MEDIA_GRAPH_UPLOAD_THRESHOLD:
+                if not token:
+                    return DeliveryResult(
+                        success=False,
+                        error=f"File too large ({file_size} bytes) and no token available for Graph upload",
+                    )
                 from .graph import GraphClient
 
                 client = GraphClient(token)
@@ -317,12 +344,19 @@ class MSTeamsAdapter(BaseChannelAdapter):
                     web_url = upload_result.get("webUrl", "")
                     if web_url:
                         return await _send_stream_media(self._sender, chat_id, web_url, mime_type)
+                    return DeliveryResult(success=False, error="Graph upload: no webUrl returned")
                 except Exception as e:
                     logger.error(f"MSTeams media upload via Graph failed: {e}")
                     return DeliveryResult(success=False, error=f"Graph upload failed: {e}")
                 finally:
                     await client.close()
 
+            if file_size > MEDIA_BASE64_MAX_SIZE:
+                return DeliveryResult(
+                    success=False,
+                    error=f"File too large for base64 encoding ({file_size} > {MEDIA_BASE64_MAX_SIZE})",
+                )
+
             content_url = f"data:{mime_type};base64,{base64.b64encode(data).decode()}"
         elif isinstance(data, str):
             content_url = data
@@ -643,7 +677,8 @@ class MSTeamsAdapter(BaseChannelAdapter):
 
         if activity_type == "message":
             msg_id = body.get("id", "")
-            if self._is_duplicate(msg_id):
+            if await self._check_and_mark(msg_id):
+                logger.debug(f"MSTeams: duplicate message {msg_id}, skipping")
                 return None
 
             if msg_id and self._sent_cache.was_sent(msg_id):
@@ -686,6 +721,21 @@ class MSTeamsAdapter(BaseChannelAdapter):
 
             channel_msg = self._classify_message(channel_msg)
             self._track_message(channel_msg)
+
+            if self.config.get("thread_context_enabled", True):
+                try:
+                    token = self._sender.token if self._sender else None
+                    if token:
+                        from .graph import GraphClient
+                        from .normalizer import inject_thread_context
+
+                        graph_client = GraphClient(token)
+                        session_key = (channel_msg.metadata or {}).get("SessionKey", "")
+                        channel_msg = await inject_thread_context(channel_msg, graph_client, session_key)
+                        await graph_client.close()
+                except Exception:
+                    pass
+
             return channel_msg
 
         if activity_type == "conversationUpdate":
@@ -865,20 +915,19 @@ class MSTeamsAdapter(BaseChannelAdapter):
             self._message_tracker[chat_id] = msg_id
             while len(self._message_tracker) > _MAX_MESSAGE_TRACKER_SIZE:
                 self._message_tracker.popitem(last=False)
-            self._mark_seen(msg_id)
 
-    def _is_duplicate(self, msg_id: str) -> bool:
+    async def _check_and_mark(self, msg_id: str) -> bool:
         if not msg_id:
             return False
-        now = time.monotonic()
-        expired = [mid for mid, ts in self._dedup_ids.items() if now - ts > self._dedup_ttl]
-        for mid in expired:
-            self._dedup_ids.pop(mid, None)
-        return msg_id in self._dedup_ids
-
-    def _mark_seen(self, msg_id: str) -> None:
-        if msg_id:
-            self._dedup_ids[msg_id] = time.monotonic()
+        async with self._dedup_lock:
+            now = time.monotonic()
+            expired = [mid for mid, ts in self._dedup_ids.items() if now - ts > self._dedup_ttl]
+            for mid in expired:
+                self._dedup_ids.pop(mid, None)
+            if msg_id in self._dedup_ids:
+                return True
+            self._dedup_ids[msg_id] = now
+            return False
 
     def _resolve_conversation_id(self, response: ChannelResponse) -> str:
         return response.identity.channel_chat_id

backend/package/yuxi/channels/adapters/msteams/connection_modes.py

@@ -0,0 +1,285 @@
+"""Microsoft Teams 连接模式多样化。
+
+支持 Webhook (默认)、WebSocket (Bot Framework Streaming Extensions)、
+Polling 三种连接模式，可通过配置项 connection_mode 切换。
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+from typing import Any
+
+import aiohttp
+
+from yuxi.utils.logging_config import logger
+
+CONNECTION_MODE_WEBHOOK = "webhook"
+CONNECTION_MODE_WEBSOCKET = "websocket"
+CONNECTION_MODE_POLLING = "polling"
+
+VALID_CONNECTION_MODES = {CONNECTION_MODE_WEBHOOK, CONNECTION_MODE_WEBSOCKET, CONNECTION_MODE_POLLING}
+
+BOT_STREAMING_URL_TEMPLATE = "https://directline.botframework.com/v3/directline/conversations/{conversation_id}/stream"
+BOT_ACTIVITIES_URL_TEMPLATE = "https://smba.trafficmanager.net/emea/v3/conversations/{conversation_id}/activities"
+
+POLLING_DEFAULT_INTERVAL_S = 2.0
+POLLING_MAX_INTERVAL_S = 10.0
+WEBSOCKET_PING_INTERVAL_S = 30.0
+WEBSOCKET_PONG_TIMEOUT_S = 10.0
+
+
+class WebSocketClient:
+    """Bot Framework Streaming Extensions WebSocket 客户端。
+
+    提供实时双向连接，延迟低于 Webhook 模式。
+    """
+
+    def __init__(
+        self,
+        app_id: str,
+        app_password: str,
+        stream_url: str = "",
+        ping_interval: float = WEBSOCKET_PING_INTERVAL_S,
+        pong_timeout: float = WEBSOCKET_PONG_TIMEOUT_S,
+    ):
+        self._app_id = app_id
+        self._app_password = app_password
+        self._stream_url = stream_url
+        self._ping_interval = ping_interval
+        self._pong_timeout = pong_timeout
+        self._ws: aiohttp.ClientWebSocketResponse | None = None
+        self._session: aiohttp.ClientSession | None = None
+        self._running = False
+        self._on_message: Any = None
+
+    def set_message_handler(self, handler: Any) -> None:
+        self._on_message = handler
+
+    async def connect(self, conversation_id: str = "") -> None:
+        if self._running:
+            return
+
+        self._session = aiohttp.ClientSession()
+        stream_url = self._stream_url or BOT_STREAMING_URL_TEMPLATE.format(conversation_id=conversation_id)
+
+        headers = {
+            "Authorization": f"Bearer {self._app_password}",
+            "Content-Type": "application/json",
+        }
+
+        self._ws = await self._session.ws_connect(stream_url, headers=headers, heartbeat=self._ping_interval)
+        self._running = True
+        logger.info(f"MSTeams WebSocket connected: {stream_url}")
+
+        asyncio.create_task(self._read_loop())
+
+    async def _read_loop(self) -> None:
+        while self._running and self._ws is not None:
+            try:
+                msg = await self._ws.receive(timeout=self._pong_timeout)
+                if msg.type == aiohttp.WSMsgType.TEXT:
+                    try:
+                        data = json.loads(msg.data)
+                        activities = data.get("activities", [])
+                        for activity in activities:
+                            if self._on_message:
+                                await self._on_message(activity)
+                    except json.JSONDecodeError:
+                        logger.warning("MSTeams WebSocket: invalid JSON received")
+                elif msg.type == aiohttp.WSMsgType.CLOSED:
+                    logger.info("MSTeams WebSocket closed by server")
+                    break
+                elif msg.type == aiohttp.WSMsgType.ERROR:
+                    logger.error(f"MSTeams WebSocket error: {self._ws.exception()}")
+                    break
+            except TimeoutError:
+                continue
+            except Exception as e:
+                logger.error(f"MSTeams WebSocket read error: {e}")
+                await asyncio.sleep(1)
+
+        if self._running:
+            logger.info("MSTeams WebSocket disconnected, attempting reconnect...")
+            await asyncio.sleep(2)
+            if self._running:
+                asyncio.create_task(self._reconnect())
+
+    async def _reconnect(self) -> None:
+        try:
+            await self.close()
+            await asyncio.sleep(3)
+            await self.connect()
+        except Exception as e:
+            logger.error(f"MSTeams WebSocket reconnect failed: {e}")
+
+    async def send_activity(self, activity: dict[str, Any]) -> bool:
+        if not self._ws or self._ws.closed:
+            return False
+        try:
+            data = json.dumps(activity)
+            await self._ws.send_str(data)
+            return True
+        except Exception as e:
+            logger.error(f"MSTeams WebSocket send error: {e}")
+            return False
+
+    async def close(self) -> None:
+        self._running = False
+        if self._ws and not self._ws.closed:
+            await self._ws.close()
+            self._ws = None
+        if self._session and not self._session.closed:
+            await self._session.close()
+            self._session = None
+
+
+class PollingClient:
+    """Bot Framework Connector API 轮询拉取客户端。
+
+    适用于无法使用 Webhook 或 WebSocket 的部署场景。
+    """
+
+    def __init__(
+        self,
+        app_id: str,
+        app_password: str,
+        service_url: str = "",
+        poll_interval: float = POLLING_DEFAULT_INTERVAL_S,
+        max_interval: float = POLLING_MAX_INTERVAL_S,
+    ):
+        self._app_id = app_id
+        self._app_password = app_password
+        self._service_url = (service_url or "https://smba.trafficmanager.net/emea").rstrip("/")
+        self._poll_interval = poll_interval
+        self._max_interval = max_interval
+        self._session: aiohttp.ClientSession | None = None
+        self._running = False
+        self._last_watermark: str = ""
+        self._on_message: Any = None
+
+    def set_message_handler(self, handler: Any) -> None:
+        self._on_message = handler
+
+    async def connect(self) -> None:
+        if self._running:
+            return
+        self._session = aiohttp.ClientSession()
+        self._running = True
+        logger.info(f"MSTeams Polling started: interval={self._poll_interval}s")
+        asyncio.create_task(self._poll_loop())
+
+    async def _poll_loop(self) -> None:
+        backoff = self._poll_interval
+
+        while self._running:
+            try:
+                activities = await self._fetch_activities()
+                if activities:
+                    backoff = self._poll_interval
+                    for activity in activities:
+                        if self._on_message:
+                            await self._on_message(activity)
+                else:
+                    backoff = min(backoff * 1.5, self._max_interval)
+            except Exception as e:
+                logger.warning(f"MSTeams Polling fetch error: {e}")
+                backoff = min(backoff * 2, self._max_interval)
+
+            await asyncio.sleep(backoff)
+
+    async def _fetch_activities(self) -> list[dict[str, Any]]:
+        if not self._session:
+            return []
+
+        url = BOT_ACTIVITIES_URL_TEMPLATE.format(conversation_id="all")
+        headers = {
+            "Authorization": f"Bearer {self._app_password}",
+            "Content-Type": "application/json",
+        }
+
+        async with self._session.get(url, headers=headers) as resp:
+            if resp.status == 200:
+                data = await resp.json()
+                activities = data.get("activities", [])
+                self._last_watermark = data.get("watermark", self._last_watermark)
+                return activities
+            elif resp.status == 429:
+                logger.warning("MSTeams Polling rate limited")
+                return []
+            else:
+                body = await resp.text()
+                logger.warning(f"MSTeams Polling HTTP {resp.status}: {body[:200]}")
+                return []
+
+    async def close(self) -> None:
+        self._running = False
+        if self._session and not self._session.closed:
+            await self._session.close()
+            self._session = None
+
+
+class ConnectionModeManager:
+    def __init__(self, config: dict[str, Any]):
+        self._mode = config.get("connection_mode", CONNECTION_MODE_WEBHOOK)
+        if self._mode not in VALID_CONNECTION_MODES:
+            logger.warning(f"Invalid connection_mode '{self._mode}', falling back to 'webhook'")
+            self._mode = CONNECTION_MODE_WEBHOOK
+
+        self._ws_client: WebSocketClient | None = None
+        self._poll_client: PollingClient | None = None
+        self._message_router: Any = None
+
+    @property
+    def mode(self) -> str:
+        return self._mode
+
+    @property
+    def is_webhook(self) -> bool:
+        return self._mode == CONNECTION_MODE_WEBHOOK
+
+    @property
+    def is_websocket(self) -> bool:
+        return self._mode == CONNECTION_MODE_WEBSOCKET
+
+    @property
+    def is_polling(self) -> bool:
+        return self._mode == CONNECTION_MODE_POLLING
+
+    def set_message_router(self, router: Any) -> None:
+        self._message_router = router
+
+    async def start(
+        self,
+        app_id: str,
+        app_password: str,
+        service_url: str = "",
+        conversation_id: str = "",
+    ) -> None:
+        if self._mode == CONNECTION_MODE_WEBSOCKET:
+            self._ws_client = WebSocketClient(app_id, app_password, stream_url=conversation_id)
+            if self._message_router:
+                self._ws_client.set_message_handler(self._message_router)
+            await self._ws_client.connect(conversation_id)
+            logger.info("MSTeams connection mode: WebSocket")
+        elif self._mode == CONNECTION_MODE_POLLING:
+            self._poll_client = PollingClient(app_id, app_password, service_url)
+            if self._message_router:
+                self._poll_client.set_message_handler(self._message_router)
+            await self._poll_client.connect()
+            logger.info("MSTeams connection mode: Polling")
+        else:
+            logger.info("MSTeams connection mode: Webhook")
+
+    async def stop(self) -> None:
+        if self._ws_client:
+            await self._ws_client.close()
+            self._ws_client = None
+        if self._poll_client:
+            await self._poll_client.close()
+            self._poll_client = None
+
+    async def send_activity(self, activity: dict[str, Any]) -> bool:
+        if self._mode == CONNECTION_MODE_WEBSOCKET and self._ws_client:
+            return await self._ws_client.send_activity(activity)
+        return False

backend/package/yuxi/channels/adapters/msteams/credentials.py

@@ -58,9 +58,9 @@ class FederatedCredential:
 
     def _check_prerequisites(self) -> None:
         try:
-            from cryptography.hazmat.primitives import serialization  # noqa: F401
-            from cryptography.hazmat.backends import default_backend  # noqa: F401
             import jwt as pyjwt  # noqa: F401
+            from cryptography.hazmat.backends import default_backend  # noqa: F401
+            from cryptography.hazmat.primitives import serialization  # noqa: F401
         except ImportError as e:
             raise FederatedCredentialError(
                 "FederatedCredential requires 'cryptography' and 'PyJWT'. Install with: pip install cryptography pyjwt"
@@ -101,9 +101,9 @@ class FederatedCredential:
 
     async def _get_certificate_token(self, scope: str) -> str | None:
         try:
-            from cryptography.hazmat.primitives import serialization
-            from cryptography.hazmat.backends import default_backend
             import jwt as pyjwt
+            from cryptography.hazmat.backends import default_backend
+            from cryptography.hazmat.primitives import serialization
 
             cert_path = Path(self.certificate_path)
             if not cert_path.exists():

backend/package/yuxi/channels/adapters/msteams/feedback.py

@@ -208,4 +208,27 @@ def process_feedback(
     elif is_positive:
         logger.info(f"MSTeams positive feedback received from {user_name} ({user_id})")
 
+    record_feedback_session(result)
+
     return result
+
+
+def record_feedback_session(feedback_result: dict[str, Any], storage_dir: str = "") -> None:
+    import os
+
+    session_dir = Path(
+        storage_dir or os.environ.get("MSTEAMS_FEEDBACK_DIR", "") or str(Path.home() / ".yuxi" / "msteams")
+    )
+    session_dir.mkdir(parents=True, exist_ok=True)
+    session_file = session_dir / "feedback-sessions.jsonl"
+
+    record = {
+        "timestamp": time.time(),
+        **feedback_result,
+    }
+
+    try:
+        with open(session_file, "a", encoding="utf-8") as f:
+            f.write(json.dumps(record, ensure_ascii=False) + "\n")
+    except OSError as e:
+        logger.warning(f"MSTeams feedback: failed to record session: {e}")

backend/package/yuxi/channels/adapters/msteams/file_upload.py

@@ -5,57 +5,36 @@ FileConsentCard 大文件交互流程 + SharePoint 上传策略 + SSRF 三层防
 
 from __future__ import annotations
 
-import ipaddress
-import re
 import time
 from pathlib import Path
 from typing import Any
 from urllib.parse import urlparse
 
+from yuxi.channels.auth.ssrf_guard import (
+    is_hostname_allowed,
+    is_private_url,
+)
 from yuxi.utils.logging_config import logger
 
 FILE_CONSENT_MAX_SIZE = 4 * 1024 * 1024
 
-_SSRF_ALLOWED_DOMAINS = frozenset(
-    {
-        "graph.microsoft.com",
-        "graph.microsoft.us",
-        "login.microsoftonline.com",
-        "login.microsoftonline.us",
-        "api.botframework.com",
-        "api.botframework.us",
-        "smba.trafficmanager.net",
-        "sharepoint.com",
-        "sharepoint-df.com",
-        "onedrive.com",
-        "office.com",
-        "office.net",
-    }
-)
-
-_SSRF_DOMAIN_PARENT_PATTERN = re.compile(
-    r"^(.+\.)?(" + "|".join(map(re.escape, _SSRF_ALLOWED_DOMAINS)) + r")$",
-    re.IGNORECASE,
-)
-
-_SSRF_PRIVATE_RANGES = [
-    ipaddress.ip_network("10.0.0.0/8"),
-    ipaddress.ip_network("172.16.0.0/12"),
-    ipaddress.ip_network("192.168.0.0/16"),
-    ipaddress.ip_network("127.0.0.0/8"),
-    ipaddress.ip_network("169.254.0.0/16"),
-    ipaddress.ip_network("0.0.0.0/8"),
-    ipaddress.ip_network("fc00::/7"),
-    ipaddress.ip_network("::1/128"),
+_SSRF_ALLOWED_DOMAINS = [
+    "graph.microsoft.com",
+    "graph.microsoft.us",
+    "login.microsoftonline.com",
+    "login.microsoftonline.us",
+    "api.botframework.com",
+    "api.botframework.us",
+    "smba.trafficmanager.net",
+    "*.sharepoint.com",
+    "*.sharepoint-df.com",
+    "*.onedrive.com",
+    "*.office.com",
+    "*.office.net",
 ]
 
 
 def _is_ssrf_safe_url(url: str) -> tuple[bool, str]:
-    """三层 SSRF 防护：协议 → 域名白名单 → DNS 私有地址检测。
-
-    Returns:
-        (is_safe, reason) — safe 为 True 表示可安全请求。
-    """
     parsed = urlparse(url)
 
     if parsed.scheme != "https":
@@ -65,16 +44,11 @@ def _is_ssrf_safe_url(url: str) -> tuple[bool, str]:
     if not hostname:
         return False, "No hostname in URL"
 
-    if not _SSRF_DOMAIN_PARENT_PATTERN.match(hostname):
+    if not is_hostname_allowed(hostname, _SSRF_ALLOWED_DOMAINS):
         return False, f"Domain '{hostname}' not in SSRF allowlist"
 
-    try:
-        addr = ipaddress.ip_address(hostname)
-        for net in _SSRF_PRIVATE_RANGES:
-            if addr in net:
-                return False, f"IP '{hostname}' is in private range {net}"
-    except ValueError:
-        pass
+    if is_private_url(url):
+        return False, f"URL '{url}' resolves to private/internal network"
 
     return True, ""
 

backend/package/yuxi/channels/adapters/msteams/formatter.py

@@ -11,7 +11,7 @@ from typing import Any
 
 from yuxi.channels.models import ChannelResponse, MessageType
 
-from .mentions import parse_outbound_mentions, apply_mentions_to_activity
+from .mentions import apply_mentions_to_activity, parse_outbound_mentions
 from .send import is_silent_reply_text, strip_silent_token
 
 

backend/package/yuxi/channels/adapters/msteams/graph.py

@@ -14,6 +14,8 @@ import aiohttp
 
 from yuxi.utils.logging_config import logger
 
+from .user_agent import get_default_user_agent
+
 GRAPH_BASE_URL = "https://graph.microsoft.com/v1.0"
 GRAPH_UPLOAD_CHUNK_SIZE = 4 * 1024 * 1024
 GRAPH_MAX_RETRIES = 3
@@ -83,6 +85,7 @@ class GraphClient:
         return {
             "Authorization": f"Bearer {self._token}",
             "Content-Type": "application/json",
+            "User-Agent": get_default_user_agent(),
         }
 
     async def _ensure_session(self) -> aiohttp.ClientSession:

backend/package/yuxi/channels/adapters/msteams/group_mgmt.py

@@ -5,7 +5,7 @@ addParticipant / removeParticipant / renameGroup 操作。
 
 from __future__ import annotations
 
-from typing import Any, TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 if TYPE_CHECKING:
     from .graph import GraphClient

backend/package/yuxi/channels/adapters/msteams/media_tools.py

@@ -0,0 +1,67 @@
+"""Microsoft Teams 媒体辅助工具。
+
+提供 MIME 类型识别、文件名提取、消息 ID 提取等辅助函数。
+"""
+
+from __future__ import annotations
+
+import mimetypes
+import os
+from typing import Any
+
+
+def get_mime_type(filename: str) -> str:
+    mime_type, _ = mimetypes.guess_type(filename)
+    return mime_type or "application/octet-stream"
+
+
+def extract_filename(activity: dict[str, Any]) -> str:
+    attachments = activity.get("attachments", []) or []
+    for att in attachments:
+        name = att.get("name", "").strip()
+        if name:
+            return os.path.basename(name)
+        content = att.get("content", {}) or {}
+        content_name = content.get("name", "").strip()
+        if content_name:
+            return os.path.basename(content_name)
+    return ""
+
+
+def extract_message_id(activity: dict[str, Any]) -> str:
+    msg_id = activity.get("id", "")
+    if msg_id:
+        return msg_id
+
+    channel_data = activity.get("channelData", {}) or {}
+    channel_msg_id = channel_data.get("id", "")
+    if channel_msg_id:
+        return channel_msg_id
+
+    return ""
+
+
+def extract_attachment_urls(activity: dict[str, Any]) -> list[dict[str, str]]:
+    urls: list[dict[str, str]] = []
+    attachments = activity.get("attachments", []) or []
+    for att in attachments:
+        content_type = att.get("contentType", "")
+        content_url = att.get("contentUrl", "")
+        name = att.get("name", "")
+        if content_url:
+            urls.append(
+                {
+                    "url": content_url,
+                    "name": name or os.path.basename(content_url),
+                    "content_type": content_type,
+                }
+            )
+    return urls
+
+
+def get_edited_timestamp(activity: dict[str, Any]) -> str | None:
+    channel_data = activity.get("channelData", {}) or {}
+    edit_time = channel_data.get("editedTimestamp", "")
+    if edit_time:
+        return edit_time
+    return activity.get("editedTimestamp")

backend/package/yuxi/channels/adapters/msteams/normalizer.py

@@ -6,7 +6,7 @@
 from __future__ import annotations
 
 import re
-from datetime import datetime, UTC
+from datetime import UTC, datetime
 from typing import Any
 
 from yuxi.channels.models import (
@@ -116,6 +116,15 @@ def normalize_inbound(activity: dict[str, Any]) -> ChannelMessage:
         "ConversationMessageId": thread_root_id or "",
     }
 
+    thread_meta = extract_thread_metadata(activity)
+    metadata.update(thread_meta)
+
+    quote_info = extract_quote_info(activity)
+    metadata.update(quote_info)
+
+    edit_info = detect_edited_message(activity)
+    metadata.update(edit_info)
+
     reply_to = thread_root_id or activity.get("replyToId")
 
     return ChannelMessage(
@@ -181,12 +190,16 @@ def normalize_invoke(activity: dict[str, Any]) -> ChannelMessage:
 
     aad_object_id = from_info.get("aadObjectId", "") or from_info.get("id", "")
     conversation_id = conversation.get("id", "")
+    conversation_type = conversation.get("conversationType", "personal")
+    base_conv_id = extract_base_conversation_id(conversation_id)
+
+    channel_chat_id = resolve_channel_chat_id(conversation_type, base_conv_id, aad_object_id, channel_data)
 
     identity = ChannelIdentity(
         channel_id="msteams",
         channel_type=ChannelType.MS_TEAMS,
         channel_user_id=aad_object_id,
-        channel_chat_id=conversation_id,
+        channel_chat_id=channel_chat_id,
         channel_message_id=activity.get("id"),
     )
 
@@ -276,6 +289,102 @@ def extract_reply_context(activity: dict[str, Any]) -> dict[str, str]:
     return result
 
 
+def extract_quote_info(activity: dict[str, Any]) -> dict[str, str]:
+    result: dict[str, str] = {}
+    reply_to_id = activity.get("replyToId", "")
+    if reply_to_id:
+        result["quoted_message_id"] = reply_to_id
+
+    channel_data = activity.get("channelData", {}) or {}
+    quote_target = channel_data.get("quoteMessageId", "")
+    if quote_target:
+        result["quote_target_id"] = quote_target
+
+    text = activity.get("text", "") or ""
+    if "<blockquote>" in text.lower():
+        result["has_blockquote"] = "true"
+
+    return result
+
+
+def detect_edited_message(activity: dict[str, Any]) -> dict[str, str]:
+    result: dict[str, str] = {}
+    channel_data = activity.get("channelData", {}) or {}
+    edit_time = channel_data.get("editedTimestamp", "")
+
+    if not edit_time:
+        edit_time = activity.get("editedTimestamp", "")
+
+    if edit_time:
+        result["is_edited"] = "true"
+        result["edited_timestamp"] = str(edit_time)
+
+    return result
+
+
+def extract_thread_metadata(activity: dict[str, Any]) -> dict[str, str]:
+    metadata: dict[str, str] = {}
+    channel_data = activity.get("channelData", {}) or {}
+    team_id = (channel_data.get("team") or {}).get("id", "")
+    channel_id = (channel_data.get("channel") or {}).get("id", "")
+
+    conversation = activity.get("conversation", {}) or {}
+    conversation_id = conversation.get("id", "")
+    thread_root_id = extract_thread_root_id(conversation_id)
+
+    if team_id:
+        metadata["team_id"] = team_id
+    if channel_id:
+        metadata["channel_id"] = channel_id
+    if thread_root_id:
+        metadata["thread_root_id"] = thread_root_id
+
+    reply_to_id = activity.get("replyToId", "")
+    if reply_to_id:
+        metadata["reply_to_id"] = reply_to_id
+
+    base_conv_id = extract_base_conversation_id(conversation_id)
+    metadata["base_conversation_id"] = base_conv_id
+
+    return metadata
+
+
+async def inject_thread_context(
+    channel_message: ChannelMessage,
+    graph_client: Any = None,
+    session_key: str = "",
+) -> ChannelMessage:
+    metadata = channel_message.metadata or {}
+    team_id = metadata.get("team_id", "")
+    channel_id = metadata.get("channel_id", "")
+    thread_root_id = metadata.get("thread_root_id", "") or metadata.get("reply_to_id", "")
+
+    if not graph_client or not team_id or not channel_id or not thread_root_id:
+        return channel_message
+
+    try:
+        from .graph import format_thread_context
+
+        parent = await graph_client.fetch_parent_message(team_id, channel_id, thread_root_id)
+        if not parent:
+            return channel_message
+
+        replies = await graph_client.fetch_thread_replies(team_id, channel_id, thread_root_id, limit=20)
+        context_msgs = [parent] + replies
+
+        thread_context = format_thread_context(context_msgs)
+
+        if thread_context and channel_message.content:
+            channel_message.content = f"{thread_context}\n\n[Current message]\n{channel_message.content}"
+            metadata["thread_context_injected"] = True
+            channel_message.metadata = metadata
+
+    except Exception:
+        pass
+
+    return channel_message
+
+
 def extract_html_text(html_content: str) -> str:
     if not html_content:
         return ""

backend/package/yuxi/channels/adapters/msteams/oauth_flow.py

@@ -13,9 +13,9 @@ import os
 import secrets
 import sys
 import webbrowser
-from http.server import HTTPServer, BaseHTTPRequestHandler
+from http.server import BaseHTTPRequestHandler, HTTPServer
 from typing import Any
-from urllib.parse import urlencode, urlparse, parse_qs
+from urllib.parse import parse_qs, urlencode, urlparse
 
 import aiohttp
 

backend/package/yuxi/channels/adapters/msteams/pins.py

@@ -5,7 +5,7 @@
 
 from __future__ import annotations
 
-from typing import Any, TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 if TYPE_CHECKING:
     from .graph import GraphClient

backend/package/yuxi/channels/adapters/msteams/proactive.py

@@ -11,7 +11,7 @@ import asyncio
 import json
 import time
 from pathlib import Path
-from typing import Any, TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 from yuxi.channels.models import DeliveryResult
 from yuxi.utils.logging_config import logger

backend/package/yuxi/channels/adapters/msteams/security.py

@@ -7,7 +7,8 @@ DM Policy (open/pairing/allowlist/disabled) 和 Group Policy (open/allowlist/dis
 from __future__ import annotations
 
 import fnmatch
-from typing import Any, TYPE_CHECKING
+import os
+from typing import TYPE_CHECKING, Any
 
 from yuxi.utils.logging_config import logger
 
@@ -27,6 +28,7 @@ _VALID_DM_POLICIES = {DM_POLICY_OPEN, DM_POLICY_PAIRING, DM_POLICY_ALLOWLIST, DM
 _VALID_GROUP_POLICIES = {GROUP_POLICY_OPEN, GROUP_POLICY_ALLOWLIST, GROUP_POLICY_DISABLED}
 
 _ALLOW_WILDCARD = "*"
+_ALLOW_WILDCARD_ENABLED = os.environ.get("MSTEAMS_ALLOW_WILDCARD", "").lower() in ("1", "true", "yes")
 
 AccessGroupResolver = "Callable[[str, list[str]], Awaitable[set[str]]]"
 
@@ -64,6 +66,17 @@ class SecurityPolicy:
         self.allow_from = self._normalize_allow_entries(config.get("allow_from", []))
         self.group_allow_from = self._normalize_allow_entries(config.get("group_allow_from", []))
         self.allow_name_matching = config.get("allow_name_matching", False)
+
+        if self.dm_policy == DM_POLICY_ALLOWLIST and _ALLOW_WILDCARD in self.allow_from:
+            logger.warning(
+                "MSTeams SecurityPolicy: wildcard '*' detected in DM allowlist. "
+                "All users will bypass DM policy. Consider using dm_policy='open' for explicit intent."
+            )
+        if self.group_policy == GROUP_POLICY_ALLOWLIST and _ALLOW_WILDCARD in self.group_allow_from:
+            logger.warning(
+                "MSTeams SecurityPolicy: wildcard '*' detected in Group allowlist. "
+                "All groups/users will bypass Group policy."
+            )
         self._teams_config: dict[str, dict[str, Any]] = config.get("teams", {})
 
         self.use_access_groups: bool = config.get("use_access_groups", False)
@@ -114,7 +127,10 @@ class SecurityPolicy:
         if not allowlist:
             return False
         if _ALLOW_WILDCARD in allowlist:
-            return True
+            if _ALLOW_WILDCARD_ENABLED:
+                logger.warning("MSTeams SecurityPolicy: wildcard matched — bypassing allowlist check")
+                return True
+            return False
         for entry in allowlist:
             if self._match_entry(user_id, entry):
                 return True

backend/package/yuxi/channels/adapters/msteams/send.py

@@ -8,7 +8,7 @@ from __future__ import annotations
 
 import asyncio
 import time
-from collections.abc import Callable, Awaitable
+from collections.abc import Awaitable, Callable
 from dataclasses import dataclass, field
 from enum import Enum
 from typing import Any

backend/package/yuxi/channels/adapters/msteams/sent_message_cache.py

@@ -6,6 +6,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import time
 from collections import OrderedDict
 from typing import Any
@@ -19,6 +20,7 @@ class SentMessageCache:
         self._max_entries = max_entries
         self._ttl = ttl
         self._entries: OrderedDict[str, dict[str, Any]] = OrderedDict()
+        self._cleanup_task: asyncio.Task | None = None
 
     def record(self, message_id: str, chat_id: str) -> None:
         now = time.monotonic()
@@ -48,6 +50,20 @@ class SentMessageCache:
     def clear(self) -> None:
         self._entries.clear()
 
+    def start_cleanup_task(self, interval: float = 600.0) -> None:
+        if self._cleanup_task and not self._cleanup_task.done():
+            return
+        self._cleanup_task = asyncio.create_task(self._cleanup_loop(interval))
+
+    async def _cleanup_loop(self, interval: float) -> None:
+        while True:
+            await asyncio.sleep(interval)
+            self._cleanup(time.monotonic())
+
+    def stop_cleanup_task(self) -> None:
+        if self._cleanup_task and not self._cleanup_task.done():
+            self._cleanup_task.cancel()
+
     @property
     def count(self) -> int:
         return len(self._entries)

backend/package/yuxi/channels/adapters/msteams/setup_wizard.py

@@ -16,7 +16,6 @@ from typing import Any
 
 from yuxi.utils.logging_config import logger
 
-
 STEP_TITLES = {
     1: "Bot 凭据配置",
     2: "DM 策略配置",
@@ -201,6 +200,26 @@ class MSTeamsSetupWizard:
             print("\n  ℹ 请在 Azure AD 中注册以下重定向 URI：")
             print("    http://localhost:5353/oauth/msteams/callback")
             print("  ℹ 授权流程: `make msteams-oauth` 启动本地回调服务器")
+
+            auth_url = "https://login.microsoftonline.com/organizations/oauth2/v2.0/authorize"
+            if self._confirm("\n是否自动打开浏览器开始 OAuth 授权", True):
+                tenant_id = self._config.get("tenant_id", "organizations")
+                client_id = self._config.get("app_id", "")
+                if client_id:
+                    scope_param = " ".join(scopes.split()) if isinstance(scopes, str) else " ".join(scopes.split())
+                    oauth_url = (
+                        f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/authorize"
+                        f"?client_id={client_id}"
+                        f"&response_type=code"
+                        f"&redirect_uri=http://localhost:5353/oauth/msteams/callback"
+                        f"&scope={scope_param.replace(' ', '%20')}"
+                        f"&response_mode=query"
+                    )
+                    _open_browser(oauth_url)
+                    print("  ℹ 浏览器已打开，完成授权后将回调 localhost:5353")
+                else:
+                    print("  ⚠ 未配置 App ID，无法生成授权 URL")
+
             print(f"  ✓ OAuth: 已启用 (scopes={len(scopes.split())})")
         else:
             self._config["delegated_auth"] = {"enabled": False}
@@ -348,6 +367,27 @@ def _validate_and_normalize_answers(answers: dict[str, Any]) -> dict[str, Any]:
     return result
 
 
+def _open_browser(url: str) -> bool:
+    """跨平台浏览器自动打开。
+
+    支持 Windows (start), macOS (open), Linux (xdg-open)。
+    """
+    import subprocess
+    import sys
+
+    try:
+        if sys.platform == "win32":
+            subprocess.Popen(["start", url], shell=True)
+        elif sys.platform == "darwin":
+            subprocess.Popen(["open", url])
+        else:
+            subprocess.Popen(["xdg-open", url])
+        return True
+    except Exception as e:
+        logger.warning(f"MSTeams setup wizard: failed to open browser: {e}")
+        return False
+
+
 def run_msteams_setup_wizard(
     existing_config: dict[str, Any] | None = None,
     output_file: str | None = None,

backend/package/yuxi/channels/adapters/msteams/streaming.py

@@ -3,10 +3,10 @@ from __future__ import annotations
 import asyncio
 import random
 import time
+from collections.abc import Awaitable, Callable
 from dataclasses import dataclass
 from enum import Enum
-from typing import Any, TYPE_CHECKING
-from collections.abc import Callable, Awaitable
+from typing import TYPE_CHECKING, Any
 
 if TYPE_CHECKING:
     from .send import MessageSender
@@ -159,13 +159,58 @@ class TeamsHttpStream:
     def _random_informative_text(self) -> str:
         return random.choice(self._informative_texts)
 
+    async def _fallback_informative(self, chat_id: str) -> DeliveryResult:
+        text = random.choice(_SIMPLE_INFORMATIVE_TEXTS)
+        activity = {
+            "type": "message",
+            "text": text,
+            "textFormat": "markdown",
+        }
+        result = await self._sender.send_activity(chat_id, activity)
+        if result.success and result.message_id:
+            now = time.monotonic()
+            self._states[chat_id] = StreamState(
+                phase=StreamPhase.INFORMATIVE,
+                stream_id=str(int(now * 1000)),
+                message_id=result.message_id,
+                chat_id=chat_id,
+                created_at=now,
+                last_update_time=now,
+                has_fallback=True,
+            )
+        return result
+
+    async def _fallback_stream_update(self, state: StreamState, finished: bool) -> DeliveryResult:
+        now = time.monotonic()
+        elapsed_since_update = (now - state.last_update_time) * 1000
+
+        if elapsed_since_update < self._throttle_ms and not finished:
+            state.total_send_count += 1
+            return DeliveryResult(success=True, message_id=state.message_id)
+
+        text = state.accumulated_text[: self._chunk_limit]
+        activity = {
+            "type": "message",
+            "text": text,
+            "textFormat": "markdown",
+        }
+        result = await self._sender.update_activity(state.chat_id, state.message_id, activity)
+        state.last_update_time = time.monotonic()
+        state.total_send_count += 1
+
+        if finished:
+            state.phase = StreamPhase.FINAL
+            self._states.pop(state.chat_id, None)
+
+        return result
+
     async def send_informative(self, chat_id: str) -> DeliveryResult:
         state = self._states.get(chat_id)
         if state and state.phase == StreamPhase.STREAMING:
             return DeliveryResult(success=True, message_id=state.message_id)
 
         if not self.supports_streaming(chat_id):
-            return DeliveryResult(success=False, error="Streaming not supported for this chat type, fallback to block")
+            return await self._fallback_informative(chat_id)
 
         text = self._random_informative_text()
         activity = {
@@ -222,6 +267,9 @@ class TeamsHttpStream:
         state.accumulated_text += chunk
         state.chunk_count += 1
 
+        if not self.supports_streaming(chat_id):
+            return await self._fallback_stream_update(state, finished)
+
         if state.phase == StreamPhase.INFORMATIVE:
             if len(state.accumulated_text) >= self._min_initial_chars:
                 state.phase = StreamPhase.STREAMING

backend/package/yuxi/channels/adapters/msteams/thread_history.py

@@ -0,0 +1,65 @@
+from __future__ import annotations
+
+from datetime import datetime
+from typing import TYPE_CHECKING
+
+from yuxi.channels.history_injector import HistoryCache, HistoryFetcher
+from yuxi.channels.models import FetchOptions, HistoricalMessage
+
+if TYPE_CHECKING:
+    from yuxi.channels.adapters.msteams.graph import GraphClient
+
+
+class TeamsHistoryFetcher(HistoryFetcher):
+    def __init__(self, graph_client: GraphClient):
+        self._client = graph_client
+        self._cache = HistoryCache(max_size=100, ttl_seconds=300)
+
+    async def fetch_thread_history(self, thread_id: str, options: FetchOptions) -> list[HistoricalMessage]:
+        cached = self._cache.get(thread_id, options)
+        if cached is not None:
+            return cached
+
+        team_id, channel_id, message_id = _parse_thread_id(thread_id)
+        replies = await self._client.fetch_thread_replies(
+            team_id=team_id,
+            channel_id=channel_id,
+            message_id=message_id,
+            limit=min(options.max_messages, 50),
+        )
+
+        messages = [
+            HistoricalMessage(
+                message_id=reply.get("id", ""),
+                sender_id=(reply.get("from", {}) or {}).get("user", {}).get("id", ""),
+                sender_name=(reply.get("from", {}) or {}).get("user", {}).get("displayName", "Unknown"),
+                content=(reply.get("body", {}) or {}).get("content", ""),
+                timestamp=datetime.fromisoformat(reply.get("createdDateTime", "").replace("Z", "+00:00")),
+                is_from_bot=(reply.get("from", {}) or {}).get("application") is not None,
+            )
+            for reply in replies
+        ]
+
+        self._cache.set(thread_id, options, messages)
+        return messages
+
+    async def fetch_parent_message(self, thread_id: str) -> HistoricalMessage | None:
+        team_id, channel_id, message_id = _parse_thread_id(thread_id)
+        parent = await self._client.fetch_parent_message(team_id, channel_id, message_id)
+        if not parent:
+            return None
+
+        return HistoricalMessage(
+            message_id=parent.get("id", ""),
+            sender_id=(parent.get("from", {}) or {}).get("user", {}).get("id", ""),
+            sender_name=(parent.get("from", {}) or {}).get("user", {}).get("displayName", "Unknown"),
+            content=(parent.get("body", {}) or {}).get("content", ""),
+            timestamp=datetime.fromisoformat(parent.get("createdDateTime", "").replace("Z", "+00:00")),
+        )
+
+
+def _parse_thread_id(thread_id: str) -> tuple[str, str, str]:
+    parts = thread_id.split(":")
+    if len(parts) >= 3:
+        return parts[-3], parts[-2], parts[-1]
+    return "", "", thread_id

backend/package/yuxi/channels/adapters/msteams/tool_policy.py

@@ -1,6 +1,7 @@
 """Microsoft Teams 工具策略 (Tools Policy)。
 
-Team/Channel 级 tools allow/deny + toolsBySender 策略。
+Team/Channel/Sender 三级 tools allow/deny 策略解析。
+支持全局配置 + teams 嵌套覆盖 + toolsBySender 按发送者策略。
 """
 
 from __future__ import annotations
@@ -9,14 +10,38 @@ from typing import Any
 
 
 class ToolPolicy:
+    """MSTeams 三级工具策略解析器。
+
+    优先级：Sender > Channel > Team > Global deny > Global allow
+    """
+
     def __init__(self, config: dict[str, Any] | None = None):
         config = config or {}
         self._global_allow: set[str] = set(config.get("tools_allow", []))
         self._global_deny: set[str] = set(config.get("tools_deny", []))
-        self._tools_by_sender: dict[str, set[str]] = {}
-        sender_tools = config.get("tools_by_sender", {}) or {}
-        for sender_id, tools in sender_tools.items():
-            self._tools_by_sender[sender_id] = set(tools)
+        self._tools_by_sender: dict[str, dict[str, set[str]]] = {}
+
+        sender_tools = config.get("tools_by_sender", []) or []
+        if isinstance(sender_tools, dict):
+            for sender_id, tools in sender_tools.items():
+                if isinstance(tools, list):
+                    self._tools_by_sender[sender_id] = {"allow": set(tools), "deny": set()}
+                elif isinstance(tools, dict):
+                    self._tools_by_sender[sender_id] = {
+                        "allow": set(tools.get("allow", [])),
+                        "deny": set(tools.get("deny", [])),
+                    }
+        elif isinstance(sender_tools, list):
+            for entry in sender_tools:
+                if isinstance(entry, dict):
+                    sid = entry.get("sender_id", "")
+                    if sid:
+                        self._tools_by_sender[sid] = {
+                            "allow": set(entry.get("allow", [])),
+                            "deny": set(entry.get("deny", [])),
+                        }
+
+        self._teams_config: dict[str, dict[str, Any]] = config.get("teams", {})
 
     def is_tool_allowed(
         self,
@@ -25,9 +50,24 @@ class ToolPolicy:
         team_id: str = "",
         channel_id: str = "",
     ) -> bool:
-        sender_tools = self._tools_by_sender.get(sender_id)
-        if sender_tools is not None:
-            return tool_name in sender_tools
+        sender_rules = self._tools_by_sender.get(sender_id)
+        if sender_rules is not None:
+            if tool_name in sender_rules["deny"]:
+                return False
+            if sender_rules["allow"]:
+                return tool_name in sender_rules["allow"]
+
+        channel_allow, channel_deny = self._resolve_channel_tools(team_id, channel_id)
+        if tool_name in channel_deny:
+            return False
+        if channel_allow and tool_name not in channel_allow:
+            return False
+
+        team_allow, team_deny = self._resolve_team_tools(team_id)
+        if tool_name in team_deny:
+            return False
+        if team_allow and tool_name not in team_allow:
+            return False
 
         if tool_name in self._global_deny:
             return False
@@ -37,14 +77,62 @@ class ToolPolicy:
 
         return True
 
-    def get_allowed_tools(self, sender_id: str = "") -> list[str] | None:
-        sender_tools = self._tools_by_sender.get(sender_id)
-        if sender_tools is not None:
-            return list(sender_tools)
+    def _resolve_team_tools(self, team_id: str) -> tuple[set[str], set[str]]:
+        team_config = self._teams_config.get(team_id) or self._teams_config.get("*") or {}
+        team_tools = team_config.get("tools", {})
+        allow = set(team_tools.get("allow", []))
+        deny = set(team_tools.get("deny", []))
+        return allow, deny
+
+    def _resolve_channel_tools(self, team_id: str, channel_id: str) -> tuple[set[str], set[str]]:
+        if not team_id or not channel_id:
+            return set(), set()
+
+        team_config = self._teams_config.get(team_id) or self._teams_config.get("*") or {}
+        channels_config = team_config.get("channels", {})
+        ch_config = channels_config.get(channel_id) or channels_config.get("*") or {}
+        ch_tools = ch_config.get("tools", {})
+        allow = set(ch_tools.get("allow", []))
+        deny = set(ch_tools.get("deny", []))
+        return allow, deny
+
+    def get_allowed_tools(
+        self,
+        sender_id: str = "",
+        team_id: str = "",
+        channel_id: str = "",
+    ) -> list[str] | None:
+        sender_rules = self._tools_by_sender.get(sender_id)
+        if sender_rules is not None:
+            if sender_rules["allow"]:
+                return sorted(sender_rules["allow"] - sender_rules["deny"])
+
+        channel_allow, channel_deny = self._resolve_channel_tools(team_id, channel_id)
+        team_allow, team_deny = self._resolve_team_tools(team_id)
+
+        if channel_allow:
+            return sorted(channel_allow - channel_deny)
+        if team_allow:
+            return sorted(team_allow - team_deny)
         if self._global_allow:
-            return list(self._global_allow)
+            return sorted(self._global_allow - self._global_deny)
+
         return None
 
+    def get_denied_tools(
+        self,
+        sender_id: str = "",
+        team_id: str = "",
+        channel_id: str = "",
+    ) -> set[str]:
+        sender_rules = self._tools_by_sender.get(sender_id)
+        sender_deny = sender_rules["deny"] if sender_rules else set()
+
+        channel_allow, channel_deny = self._resolve_channel_tools(team_id, channel_id)
+        team_allow, team_deny = self._resolve_team_tools(team_id)
+
+        return sender_deny | channel_deny | team_deny | self._global_deny
+
 
 def resolve_tool_policy(
     config: dict[str, Any],
@@ -53,22 +141,9 @@ def resolve_tool_policy(
     team_id: str = "",
     channel_id: str = "",
 ) -> bool:
-    teams_config = config.get("teams", {}) or {}
-    team_config = (teams_config.get(team_id) or teams_config.get("*")) or {}
+    """快捷函数：三级工具策略解析。
 
-    if team_config:
-        team_tools = team_config.get("tools", {}) or {}
-        if tool_name in team_tools.get("deny", []):
-            return False
-        if team_tools.get("allow"):
-            return tool_name in team_tools["allow"]
-
-    global_deny = set(config.get("tools_deny", []))
-    if tool_name in global_deny:
-        return False
-
-    global_allow = set(config.get("tools_allow", []))
-    if global_allow:
-        return tool_name in global_allow
-
-    return True
+    优先级：Sender > Channel > Team > Global
+    """
+    policy = ToolPolicy(config)
+    return policy.is_tool_allowed(tool_name, sender_id, team_id, channel_id)