feat(channels): 新增会话路由、错误分类与聊天能力支持

1. 新增ChannelStreamingProtocol协议的打字指示器、消息编辑支持和 fallback 发送方法
2. 重构SDK导入顺序，调整normalizer和retry的导入位置
3. 新增错误分类模块，实现异常类型归类逻辑
4. 新增SessionRouter抽象基类，实现基础会话路由逻辑
5. 增强重试SDK，添加总超时、熔断和指标回调支持

backend/package/yuxi/channels/protocols/session_router.py

@@ -0,0 +1,64 @@
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from typing import TYPE_CHECKING
+
+from yuxi.channels.models import SessionScope
+
+if TYPE_CHECKING:
+    from yuxi.channels.models import ChannelMessage
+
+
+class SessionRouter(ABC):
+    def __init__(self, channel_type: str, account_id: str = "default"):
+        self.channel_type = channel_type
+        self.account_id = account_id
+
+    @abstractmethod
+    def resolve_thread_key(self, message: ChannelMessage) -> str:
+        """解析消息对应的线程键"""
+
+    @abstractmethod
+    def resolve_chat_id(self, message: ChannelMessage) -> str:
+        """解析聊天ID"""
+
+    @abstractmethod
+    def resolve_session_scope(self, message: ChannelMessage) -> SessionScope:
+        """解析会话范围级别"""
+
+    @abstractmethod
+    def normalize_target(self, target: str) -> str:
+        """标准化目标地址"""
+
+    def resolve_agent_route(self, message: ChannelMessage) -> str:
+        thread_key = self.resolve_thread_key(message)
+        scope = self.resolve_session_scope(message)
+        return f"agent:main:{thread_key}:{scope.value}"
+
+    def _build_key(self, scope: str, type_str: str, id_str: str) -> str:
+        return f"{self.channel_type}:{self.account_id}:{scope}:{type_str}:{id_str}"
+
+
+class BaseSessionRouter(SessionRouter):
+    def normalize_target(self, target: str) -> str:
+        if target.startswith(f"{self.channel_type}:"):
+            return target
+        return f"{self.channel_type}:{target}"
+
+    def resolve_thread_key(self, message: ChannelMessage) -> str:
+        identity = message.identity
+        chat_type = message.chat_type.value if hasattr(message.chat_type, "value") else str(message.chat_type)
+        return self._build_key("chat", chat_type, identity.channel_chat_id)
+
+    def resolve_chat_id(self, message: ChannelMessage) -> str:
+        return message.identity.channel_chat_id
+
+    def resolve_session_scope(self, message: ChannelMessage) -> SessionScope:
+        from yuxi.channels.models import ChatType
+
+        chat_type = message.chat_type
+        if chat_type == ChatType.DIRECT:
+            return SessionScope.DIRECT
+        if chat_type == ChatType.THREAD:
+            return SessionScope.THREAD
+        return SessionScope.GROUP

backend/package/yuxi/channels/protocols/streaming.py

@@ -14,4 +14,16 @@ class ChannelStreamingProtocol(Protocol):
     @property
     def block_streaming(self) -> bool: ...
 
+    @property
+    def supports_typing_indicator(self) -> bool:
+        return getattr(self, "capabilities", None) is not None and getattr(self.capabilities, "typing", False)
+
+    @property
+    def supports_message_edit(self) -> bool:
+        return getattr(self, "capabilities", None) is not None and getattr(self.capabilities, "edit", False)
+
     async def send_stream_chunk(self, chat_id: str, msg_id: str, chunk: str, finished: bool) -> DeliveryResult: ...
+
+    async def send_typing_indicator(self, chat_id: str, active: bool) -> DeliveryResult: ...
+
+    async def fallback_send(self, chat_id: str, text: str) -> DeliveryResult: ...

backend/package/yuxi/channels/sdk/__init__.py

@@ -1,12 +1,5 @@
-from yuxi.channels.sdk.retry import RetryConfig, with_retry
-from yuxi.channels.sdk.dedupe import DedupeConfig, MessageDeduplicator
 from yuxi.channels.sdk.chunking import ChunkConfig, chunk_message
-from yuxi.channels.sdk.normalizer import (
-    extract_mentions,
-    extract_urls,
-    normalize_timestamp,
-    normalize_user_id,
-)
+from yuxi.channels.sdk.dedupe import DedupeConfig, MessageDeduplicator
 from yuxi.channels.sdk.format import (
     TRUNCATION_MARKER,
     build_basic_outbound,
@@ -14,6 +7,13 @@ from yuxi.channels.sdk.format import (
     strip_html_tags,
     truncate_text,
 )
+from yuxi.channels.sdk.normalizer import (
+    extract_mentions,
+    extract_urls,
+    normalize_timestamp,
+    normalize_user_id,
+)
+from yuxi.channels.sdk.retry import RetryConfig, with_retry
 
 __all__ = [
     "RetryConfig",

backend/package/yuxi/channels/sdk/error_classifier.py

@@ -0,0 +1,52 @@
+from __future__ import annotations
+
+from enum import StrEnum
+
+from yuxi.channels.exceptions import (
+    ChannelAuthenticationError,
+    ChannelException,
+    ChannelNotConnectedError,
+    ChannelRateLimitError,
+    ChannelTimeoutError,
+    MessageTooLargeError,
+    TokenExpiredError,
+)
+
+
+class ErrorCategory(StrEnum):
+    TRANSIENT = "transient"
+    PERMANENT = "permanent"
+    RATE_LIMIT = "rate_limit"
+    AUTH_EXPIRED = "auth_expired"
+    CIRCUIT_OPEN = "circuit_open"
+    TIMEOUT = "timeout"
+
+
+_ERROR_CATEGORY_MAP: dict[type[Exception], ErrorCategory] = {
+    ChannelNotConnectedError: ErrorCategory.TRANSIENT,
+    ChannelRateLimitError: ErrorCategory.RATE_LIMIT,
+    TokenExpiredError: ErrorCategory.AUTH_EXPIRED,
+    ChannelAuthenticationError: ErrorCategory.PERMANENT,
+    ChannelTimeoutError: ErrorCategory.TIMEOUT,
+    MessageTooLargeError: ErrorCategory.PERMANENT,
+    TimeoutError: ErrorCategory.TIMEOUT,
+}
+
+
+def classify_error(error: Exception) -> ErrorCategory:
+    error_type = type(error)
+    if error_type in _ERROR_CATEGORY_MAP:
+        return _ERROR_CATEGORY_MAP[error_type]
+
+    try:
+        from yuxi.channels.infra.circuit_breaker import CircuitBreakerOpenError
+
+        if isinstance(error, CircuitBreakerOpenError):
+            return ErrorCategory.CIRCUIT_OPEN
+    except ImportError:
+        pass
+
+    if isinstance(error, ChannelException):
+        return ErrorCategory.TRANSIENT if error.retryable else ErrorCategory.PERMANENT
+
+    return ErrorCategory.TRANSIENT

backend/package/yuxi/channels/sdk/retry.py

@@ -2,13 +2,22 @@ from __future__ import annotations
 
 import asyncio
 import random
+import time
 from collections.abc import Awaitable, Callable
 from dataclasses import dataclass
+from enum import StrEnum
 from typing import Any
 
 from yuxi.utils.logging_config import logger
 
 
+class RetryDecision(StrEnum):
+    RETRY = "retry"
+    CIRCUIT_BREAK = "circuit_break"
+    FALLBACK = "fallback"
+    FAIL = "fail"
+
+
 @dataclass
 class RetryConfig:
     max_retries: int = 3
@@ -17,6 +26,9 @@ class RetryConfig:
     backoff_factor: float = 2.0
     jitter: bool = True
     retryable_exceptions: tuple[type[Exception], ...] = (Exception,)
+    total_timeout: float | None = None
+    circuit_breaker: Any = None
+    metrics_callback: Callable[[str, dict], None] | None = None
 
 
 async def with_retry(
@@ -25,18 +37,41 @@ async def with_retry(
 ) -> Any:
     cfg = config or RetryConfig()
     last_exception: Exception | None = None
+    start_time = time.monotonic()
 
     for attempt in range(cfg.max_retries + 1):
+        if cfg.total_timeout is not None and (time.monotonic() - start_time) >= cfg.total_timeout:
+            raise last_exception or Exception("Retry total_timeout exceeded")
+
+        if cfg.circuit_breaker is not None:
+            cb = cfg.circuit_breaker
+            if hasattr(cb, "state") and getattr(cb, "state", None) is not None:
+                from yuxi.channels.infra.circuit_breaker import CircuitState
+
+                if cb.state == CircuitState.OPEN:
+                    if cfg.metrics_callback:
+                        cfg.metrics_callback("circuit_breaker_skip", {"attempt": attempt})
+                    raise last_exception or Exception("Circuit breaker is OPEN, skipping retry")
+
         try:
-            return await func()
+            result = await func()
+            if cfg.metrics_callback:
+                cfg.metrics_callback("retry_success", {"attempt": attempt})
+            return result
         except cfg.retryable_exceptions as e:
             last_exception = e
             if attempt == cfg.max_retries:
                 break
+            if cfg.total_timeout is not None and (time.monotonic() - start_time) >= cfg.total_timeout:
+                break
             delay = min(cfg.base_delay * (cfg.backoff_factor**attempt), cfg.max_delay)
             if cfg.jitter:
                 delay *= random.uniform(0.5, 1.5)
+            if cfg.metrics_callback:
+                cfg.metrics_callback("retry_attempt", {"attempt": attempt + 1, "error": str(e), "delay": delay})
             logger.debug(f"[SDK/Retry] Attempt {attempt + 1}/{cfg.max_retries} failed: {e}, retrying in {delay:.2f}s")
             await asyncio.sleep(delay)
 
+    if cfg.metrics_callback:
+        cfg.metrics_callback("retry_exhausted", {"attempts": cfg.max_retries, "error": str(last_exception)})
     raise last_exception  # type: ignore[misc]