📄 gateway/platforms/base.py (第 1793-2277 行)

class BasePlatformAdapter(ABC):
    """Base class for platform adapters."""

    # ── 必须实现的抽象方法 ──
    @abstractmethod
    async def connect(self) -> bool:
        """连接到平台并开始接收消息。"""

    @abstractmethod
    async def disconnect(self) -> None:
        """断开连接。"""

    @abstractmethod
    async def send(
        self,
        chat_id: str,
        content: str,
        reply_to: Optional[str] = None,
        metadata: Optional[Dict[str, Any]] = None
    ) -> SendResult:
        """发送消息到聊天。返回 SendResult(success, message_id, error)。"""

    # ── 可选覆盖的方法（有默认实现） ──
    async def edit_message(self, ...) -> SendResult:
        """编辑已发送消息。默认返回 success=False。"""
        return SendResult(success=False, error="Not supported")

    async def delete_message(self, ...) -> bool:
        """删除消息。默认返回 False。"""
        return False

    async def send_typing(self, chat_id, metadata=None) -> None:
        """发送输入指示器。默认 no-op。"""
        pass

    async def send_draft(self, ...) -> SendResult:
        """流式草稿预览 (Telegram Bot API 9.5+)。默认 raise。"""
        raise NotImplementedError(...)

    # ── 能力标志 ──
    supports_code_blocks: bool = False
        # 平台是否渲染 Markdown 代码块
    typed_command_prefix: str = "/"
        # 用户可输入的命令前缀 (Slack/Matrix 用 "!")
    REQUIRES_EDIT_FINALIZE: bool = False
        # 是否需要显式 finalize (钉钉 AI Cards 等)

📄 gateway/platforms/base.py (第 1990-2051 行)

class BasePlatformAdapter(ABC):
    # ── 结构化流事件渲染 ──
    # 适配器决定 *如何* 呈现每个流事件
    # 默认实现 1:1 复现历史行为, 适配器覆盖以实现原生渲染

    def render_message_event(self, event, sink) -> None:
        """渲染 MessageChunk / MessageStop / Commentary。

        默认: 映射到 stream consumer 的已有原语。
        Telegram 覆盖: 用 MarkdownV2 渲染 ```bash``` 代码块。
        iMessage 覆盖: 丢弃工具 chrome (纯文本平台无法格式化)。
        """
        if isinstance(event, MessageChunk):
            sink.on_delta(event.text)
        elif isinstance(event, MessageStop):
            if not event.final:
                sink.on_segment_break()
        elif isinstance(event, Commentary):
            sink.on_commentary(event.text)

    def format_tool_event(self, event, *, mode="all",
                          preview_max_len=40) -> Optional[str]:
        """返回工具调用的渲染 chrome, 或 None 表示"吃掉"该事件。

        默认: emoji + tool_name + 短预览 (40 字符限制)。
        纯文本平台: 返回 None (不显示工具进度)。
        verbose 模式: 显示完整 args dict。
        """
        emoji = get_tool_emoji(event.tool_name, default="⚙️")
        if mode == "verbose":
            return f"{emoji} {event.tool_name}({keys})\n{args}"
        preview = event.preview or ""
        if len(preview) > 40:
            preview = preview[:37] + "..."
        return f"{emoji} {event.tool_name}: \"{preview}\""

📄 gateway/run.py (第 71-136 行)

# 第一层: 瞬态网络错误白名单
# 防止 TimedOut/NetworkError 杀死整个网关进程
_transient_class_names = {
    "TimedOut", "NetworkError", "ReadError", "WriteError",
    "ConnectError", "ConnectTimeout", "ReadTimeout", ...
}

# 第二层: Provider 错误分类与脱敏
# 用户看到分类摘要, 不看到原始 HTTP body
_GATEWAY_PROVIDER_ERROR_RE = re.compile(
    r"api\s+(?:call\s+)?failed|provider\s+authentication\s+failed|...")

# 第三层: 密钥脱敏
_GATEWAY_SECRET_PATTERNS = (
    re.compile(r"\bsk-[A-Za-z0-9][A-Za-z0-9_\-]{12,}\b"),     # OpenAI
    re.compile(r"\bgh[pousr]_[A-Za-z0-9_]{20,}\b"),           # GitHub
    re.compile(r"\bxox[baprs]-[A-Za-z0-9\-]{20,}\b"),         # Slack
    re.compile(r"\bhf_[A-Za-z0-9]{20,}\b"),                   # HuggingFace
    re.compile(r"\bglpat-[A-Za-z0-9_\-]{20,}\b"),             # GitLab
    re.compile(r"(?i)\b(Bearer\s+)[A-Za-z0-9._\-]{20,}\b"),   # Generic
)

# 第四层: Telegram 噪声过滤
# 压缩摘要/重试/降级等内部状态不推送到聊天
_TELEGRAM_NOISY_STATUS_RE = re.compile(
    r"auxiliary\s+.+?\s+failed|compression\s+summary\s+failed|...")

# 第五层: 媒体投递路径验证
# validate_media_delivery_path() 在 base.py 中
# 白名单 (缓存目录) + 黑名单 (凭证目录) + 时间窗口

📄 gateway/stream_events.py (第 43-145 行)

# ── 消息事件 ──
@dataclass(frozen=True)
class MessageChunk:
    """流式助手文本增量。"""
    text: str

@dataclass(frozen=True)
class MessageStop:
    """当前助手消息段完成。"""
    final: bool = False  # True = 整个 turn 结束

@dataclass(frozen=True)
class Commentary:
    """工具迭代间的完整中间消息。"""
    text: str

# ── 工具调用事件 ──
@dataclass(frozen=True)
class ToolCallChunk:
    """工具调用开始。"""
    tool_name: str
    preview: Optional[str] = None
    args: Optional[Dict[str, Any]] = None
    index: int = 0  # 单调索引, 关联 start/finish

@dataclass(frozen=True)
class ToolCallFinished:
    """工具调用完成。"""
    tool_name: str
    duration: float = 0.0
    ok: bool = True
    index: int = 0

# ── 网关控制事件 ──
@dataclass(frozen=True)
class LongToolHint:
    """工具运行超时的新手引导。"""
    tool_name: str = ""
    duration: float = 0.0

@dataclass(frozen=True)
class GatewayNotice:
    """网关级控制消息 (restart, online, long_run)。"""
    kind: str
    text: str = ""
    extra: Dict[str, Any] = field(default_factory=dict)

StreamEvent = Union[
    MessageChunk, MessageStop, Commentary,
    ToolCallChunk, ToolCallFinished,
    LongToolHint, GatewayNotice,
]  # 显式联合, 穷尽 match 时缺失 case = 类型错误

📄 gateway/stream_dispatch.py (第 40-132 行)

class GatewayEventDispatcher:
    """将类型化流事件通过适配器路由到投递 sink。"""

    def __init__(self, adapter, sink=None, *,
                 enqueue_tool_line=None,
                 tool_mode="all",
                 preview_max_len=40,
                 on_long_tool=None,
                 on_notice=None):
        ...

    def dispatch(self, event: StreamEvent) -> None:
        """路由单个事件。绝不向 Agent 工作线程抛出异常。"""
        try:
            self._dispatch(event)
        except Exception:  # 呈现层绝不能破坏 Agent 循环
            logger.debug("stream-event dispatch error", exc_info=True)

    def _dispatch(self, event: StreamEvent) -> None:
        if isinstance(event, (MessageChunk, MessageStop, Commentary)):
            # 消息事件 → 适配器.render_message_event → sink
            self.adapter.render_message_event(event, self.sink)

        elif isinstance(event, ToolCallChunk):
            # 工具事件 → 适配器.format_tool_event → 进度队列
            line = self.adapter.format_tool_event(
                event, mode=self.tool_mode, ...)
            if line:  # None = 适配器选择"吃掉"
                self._enqueue_tool_line(line)

📄 gateway/session.py (第 70-193 行)

@dataclass
class SessionSource:
    """描述消息来源。"""
    platform: Platform
    chat_id: str
    chat_name: Optional[str] = None
    chat_type: str = "dm"  # "dm", "group", "channel", "thread"
    user_id: Optional[str] = None
    user_name: Optional[str] = None
    thread_id: Optional[str] = None  # Forum topics, Discord threads
    guild_id: Optional[str] = None   # Discord guild / Slack workspace
    parent_chat_id: Optional[str] = None
    message_id: Optional[str] = None
    role_authorized: bool = False

    @property
    def description(self) -> str:
        """人类可读描述, 注入系统提示词。"""
        # "DM with Alice" / "group: #dev-channel" / "channel: general"

@dataclass
class SessionContext:
    """完整会话上下文, 用于动态系统提示词注入。"""
    source: SessionSource
    connected_platforms: List[Platform]
    home_channels: Dict[Platform, HomeChannel]
    shared_multi_user_session: bool = False

📄 gateway/config.py (第 274-299 行)

@dataclass
class SessionResetPolicy:
    """控制会话何时重置 (丢失上下文)。"""
    mode: str = "both"  # "daily", "idle", "both", "none"
    at_hour: int = 4     # 每日重置时间 (0-23, 本地时区)
    idle_minutes: int = 1440  # 空闲超时 (24 小时)
    notify: bool = True    # 自动重置时通知用户
    notify_exclude_platforms: tuple = ("api_server", "webhook")

📄 gateway/mirror.py (第 25-81 行)

def mirror_to_session(
    platform: str, chat_id: str, message_text: str,
    source_label: str = "cli", thread_id: Optional[str] = None,
    user_id: Optional[str] = None,
) -> bool:
    """将投递镜像写入目标会话的对话记录。"""
    session_id = _find_session_id(platform, chat_id, ...)
    if not session_id:
        return False

    mirror_msg = {
        "role": "assistant",
        "content": message_text,
        "timestamp": datetime.now().isoformat(),
        "mirror": True,
        "mirror_source": source_label,
    }
    _append_to_sqlite(session_id, mirror_msg)
    return True

# 回放时, mirror=True 的消息被标记:
# "[Delivered from cli] <content>"
# 目标 Agent 知道这是从另一个会话投递的