refactor(twitch): 重构Twitch适配器，新增Helix API支持与功能优化

本次提交对Twitch适配器进行了全面升级与优化：
1.  修复UTF8截断逻辑，避免越界访问
2.  重构群聊策略配置，标准化mention相关规则
3.  新增消息缓存管理器，支持通过消息ID查询已发送消息
4.  更新配置schema，新增prefer_helix_send开关和deprecated策略自动转换
5.  新增CLEARMSG和ROOMSTATE IRC消息解析，补充事件订阅支持
6.  优化令牌刷新逻辑，增加重试机制与退避策略
7.  新增Helix API聊天消息发送、删除和公告功能
8.  扩展事件订阅类型，新增直播状态、频道更新等系统事件
9.  新增reply、delete_message、announcement等动作支持，完善操作能力
10. 重构流式发送逻辑，新增进度指示器和配置项
11. 优化重连策略，增加指数退避与计数重置

backend/package/yuxi/channels/adapters/twitch/actions.py

@@ -5,8 +5,6 @@ from typing import Any
 from yuxi.channels.models import ChannelIdentity, ChannelResponse, DeliveryResult
 from yuxi.channels.protocols.actions import ActionContext
 
-_ACTION_HANDLERS: dict[str, str] = {}
-
 
 def _build_response(ctx: ActionContext, adapter: Any, content_key: str = "content") -> ChannelResponse | None:
     content = ctx.get(content_key, ctx.get("media_url", ""))
@@ -26,18 +24,97 @@ async def handle_action(ctx: ActionContext, adapter: Any) -> DeliveryResult:
         response = _build_response(ctx, adapter, "content")
         if response is None:
             return DeliveryResult(success=False, error="missing chat_id or content")
+        if adapter.config.get("prefer_helix_send", False):
+            result = await adapter._send_via_helix(adapter.format_outbound(response)["target"], response.content)
+            if result.success:
+                return result
         return await adapter.send(response)
+
     if ctx.action == "send_media":
         response = _build_response(ctx, adapter, "media_url")
         if response is None:
             return DeliveryResult(success=False, error="missing chat_id or media_url")
         return await adapter.send_media(ctx.chat_id, "url", response.content)
 
+    if ctx.action == "reply":
+        chat_id = ctx.get("chat_id", "")
+        content = ctx.get("content", "")
+        reply_to_msg_id = ctx.get("reply_to_msg_id", "")
+        if not chat_id or not content:
+            return DeliveryResult(success=False, error="missing chat_id or content")
+        if not reply_to_msg_id:
+            return DeliveryResult(success=False, error="missing reply_to_msg_id")
+        result = await adapter._send_via_helix(chat_id, content, reply_msg_id=reply_to_msg_id)
+        if result.success:
+            return result
+        prefixed = f"@{ctx.get('reply_to_username', 'user')} {content}"
+        response = ChannelResponse(
+            identity=ChannelIdentity(
+                channel_id="twitch",
+                channel_type=adapter.channel_type,
+                channel_user_id="",
+                channel_chat_id=chat_id,
+            ),
+            content=prefixed,
+        )
+        return await adapter.send(response)
+
+    if ctx.action in ("unsend", "delete_message"):
+        chat_id = ctx.get("chat_id", "")
+        message_id = ctx.get("message_id", "")
+        if not chat_id or not adapter._helix:
+            return DeliveryResult(success=False, error="missing chat_id or helix not available")
+        if not message_id:
+            cached = adapter._outbound_cache.find_by_message_id(ctx.get("content", ""))
+            if cached:
+                message_id = cached.get("message_id", "")
+        if not message_id:
+            return DeliveryResult(success=False, error="missing message_id for message deletion")
+
+        broadcaster_id = await adapter._resolve_broadcaster_id(chat_id)
+        if not broadcaster_id:
+            return DeliveryResult(success=False, error="broadcaster_not_found")
+
+        moderator_id = adapter._bot_user_id
+        if not moderator_id:
+            return DeliveryResult(success=False, error="bot_user_id_unknown")
+
+        success = await adapter._helix.delete_chat_message(
+            broadcaster_id=broadcaster_id,
+            moderator_id=moderator_id,
+            message_id=message_id,
+        )
+        if success:
+            return DeliveryResult(success=True, metadata={"messageId": message_id})
+        return DeliveryResult(success=False, error="helix_delete_failed")
+
+    if ctx.action == "announcement":
+        chat_id = ctx.get("chat_id", "")
+        content = ctx.get("content", "")
+        color = ctx.get("color", "primary")
+        if not chat_id or not content or not adapter._helix:
+            return DeliveryResult(success=False, error="missing chat_id or content or helix not available")
+
+        broadcaster_id = await adapter._resolve_broadcaster_id(chat_id)
+        if not broadcaster_id:
+            return DeliveryResult(success=False, error="broadcaster_not_found")
+
+        moderator_id = adapter._bot_user_id
+        if not moderator_id:
+            return DeliveryResult(success=False, error="bot_user_id_unknown")
+
+        success = await adapter._helix.send_chat_announcement(
+            broadcaster_id=broadcaster_id,
+            moderator_id=moderator_id,
+            message=content,
+            color=color,
+        )
+        if success:
+            return DeliveryResult(success=True)
+        return DeliveryResult(success=False, error="announcement_failed")
+
     unsupported_hints: dict[str, str] = {
-        "reply": "Twitch IRC does not support message replies",
         "edit": "Twitch IRC does not support message editing",
-        "unsend": "Twitch IRC does not support message deletion",
-        "delete_message": "Twitch IRC does not support message deletion",
         "edit_message": "Twitch IRC does not support message editing",
         "reactions": "Twitch IRC does not support reactions",
         "send_reaction": "Twitch IRC does not support reactions",
@@ -51,7 +128,7 @@ async def handle_action(ctx: ActionContext, adapter: Any) -> DeliveryResult:
 
 
 def supports_action(action: str) -> bool:
-    return action in {"send", "send_media"}
+    return action in {"send", "send_media", "reply", "unsend", "delete_message", "announcement"}
 
 
 def describe_message_tool() -> dict[str, Any]:
@@ -70,6 +147,37 @@ def describe_message_tool() -> dict[str, Any]:
                 "media_url": {"type": "string", "description": "Media URL to send"},
             },
         },
+        "reply": {
+            "description": "Reply to a message in a Twitch channel via Helix API",
+            "parameters": {
+                "chat_id": {"type": "string", "description": "Channel name (e.g. #channel)"},
+                "content": {"type": "string", "description": "Reply text content"},
+                "reply_to_msg_id": {"type": "string", "description": "Message ID to reply to"},
+                "reply_to_username": {"type": "string", "description": "Username being replied to (for IRC fallback)"},
+            },
+        },
+        "unsend": {
+            "description": "Delete a message from a Twitch channel via Helix API",
+            "parameters": {
+                "chat_id": {"type": "string", "description": "Channel name (e.g. #channel)"},
+                "message_id": {"type": "string", "description": "Message ID to delete"},
+            },
+        },
+        "delete_message": {
+            "description": "Delete a message from a Twitch channel via Helix API",
+            "parameters": {
+                "chat_id": {"type": "string", "description": "Channel name (e.g. #channel)"},
+                "message_id": {"type": "string", "description": "Message ID to delete"},
+            },
+        },
+        "announcement": {
+            "description": "Send a colored announcement to a Twitch channel",
+            "parameters": {
+                "chat_id": {"type": "string", "description": "Channel name (e.g. #channel)"},
+                "content": {"type": "string", "description": "Announcement text"},
+                "color": {"type": "string", "description": "Announcement color: blue/green/orange/purple/primary"},
+            },
+        },
     }
 
 
@@ -86,19 +194,16 @@ def resolve_execution_mode(action: str) -> str:
 
 def get_action_stats() -> dict[str, list[str]]:
     return {
-        "implemented": ["send", "send_media"],
+        "implemented": ["send", "send_media", "reply", "unsend", "delete_message", "announcement"],
         "planned": [],
         "unsupported": [
-            "reply",
             "edit",
-            "unsend",
             "reactions",
             "polls",
             "native_commands",
             "pin",
             "unpin",
             "send_reaction",
-            "delete_message",
             "edit_message",
         ],
     }

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

@@ -10,11 +10,11 @@ from typing import Any, ClassVar
 
 from yuxi.channels.base import BaseChannelAdapter
 from yuxi.channels.capabilities import ChannelCapabilities
-from yuxi.channels.meta import ChannelMeta
-from yuxi.channels.infra.circuit_breaker import CircuitBreaker, CircuitBreakerOpenError
 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 (
     ChannelMessage,
     ChannelResponse,
@@ -26,28 +26,38 @@ from yuxi.channels.models import (
 from yuxi.channels.registry import register_builtin_adapter
 from yuxi.utils.logging_config import logger
 
+from .actions import (
+    describe_message_tool,
+    extract_target_from_args,
+    handle_action,
+    resolve_execution_mode,
+)
+from .actions import (
+    supports_action as _supports_action,
+)
 from .auth_provider import create_auth_provider
 from .deduplicator import MessageDeduplicator
 from .eventsub_client import EventSubListener
 from .helix import HelixClient
-from .irc_parser import parse_badges, parse_irc_line, extract_nick
+from .irc_parser import extract_nick, parse_badges, parse_irc_line
 from .markdown_utils import strip_twitch_markdown
+from .normalizer import normalize_irc_message, resolve_mentions
 from .outbound_cache import OutboundCacheManager
 from .pairing import PairingStore, check_pairing_policy
-from .actions import (
-    handle_action,
-    supports_action as _supports_action,
-    describe_message_tool,
-    extract_target_from_args,
-    resolve_execution_mode,
-)
-from .normalizer import normalize_irc_message, resolve_mentions
 from .probe import get_app_access_token, validate_token
 from .rate_limiter import RateLimiter
-from .send import format_privmsg_line, format_action_line, format_pong, format_cap_req, find_utf8_cut
+from .send import find_utf8_cut, format_action_line, format_cap_req, format_pong, format_privmsg_line
 from .session import check_allowed_roles, check_group_policy
 from .token_utils import ensure_oauth_prefix
 
+PROGRESS_BAR_WIDTH = 8
+
+
+def _progress_bar(text_len_hint: int, max_chars: int) -> str:
+    ratio = min(text_len_hint / max(max_chars, 1), 1.0)
+    filled = int(ratio * PROGRESS_BAR_WIDTH)
+    return "▓" * filled + "░" * (PROGRESS_BAR_WIDTH - filled)
+
 
 @register_builtin_adapter
 class TwitchAdapter(BaseChannelAdapter):
@@ -63,15 +73,15 @@ class TwitchAdapter(BaseChannelAdapter):
     delivery_mode: ClassVar[str] = "direct"
 
     capabilities = ChannelCapabilities(
-        chat_types=["group"],
+        chat_types=["group", "direct"],
         supports_markdown=False,
         supports_streaming=True,
         streaming_modes=["off", "block"],
         text_chunk_limit=500,
-        max_media_size_mb=0,  # IRC protocol does not support media upload; URLs sent as text
-        reply=False,
+        max_media_size_mb=0,
+        reply=True,
         edit=False,
-        unsend=False,
+        unsend=True,
         reactions=False,
         polls=False,
         native_commands=False,
@@ -102,9 +112,11 @@ class TwitchAdapter(BaseChannelAdapter):
 
         self._bot_user_id: str | None = None
         self._bot_username: str | None = None
+        self._broadcaster_ids: dict[str, str] = {}
 
         self._irc_task: asyncio.Task | None = None
         self._reconnect_task: asyncio.Task | None = None
+        self._reconnect_attempts: int = 0
 
         self._deduplicator = MessageDeduplicator()
 
@@ -275,6 +287,14 @@ class TwitchAdapter(BaseChannelAdapter):
         if not content.strip():
             return DeliveryResult(success=True, metadata={"messageId": "skipped"})
 
+        prefer_helix = self.config.get("prefer_helix_send", False)
+        if prefer_helix:
+            target = self.format_outbound(response)["target"]
+            result = await self._send_via_helix(target, content)
+            if result.success:
+                return result
+            logger.info("[Twitch] Helix send failed, falling back to IRC PRIVMSG")
+
         async def _do_send():
             target = self.format_outbound(response)["target"]
             chunks = self._split_irc_text(content, target)
@@ -291,8 +311,19 @@ class TwitchAdapter(BaseChannelAdapter):
         if not chunk.strip():
             return DeliveryResult(success=True, metadata={"messageId": "skipped"})
 
-        coalesce_min_chars = self.config.get("stream_coalesce_min_chars", 30)
-        coalesce_max_delay_ms = self.config.get("stream_coalesce_max_delay_ms", 0)
+        stream_config = self.config.get("streaming", {})
+        block_cfg = stream_config.get("block", {}) if isinstance(stream_config, dict) else {}
+
+        coalesce_min_chars = self.config.get(
+            "stream_coalesce_min_chars",
+            block_cfg.get("coalesce_min_chars", 30),
+        )
+        coalesce_max_delay_ms = self.config.get(
+            "stream_coalesce_max_delay_ms",
+            block_cfg.get("coalesce_idle_ms", 0),
+        )
+
+        show_progress = stream_config.get("progress_indicator", True)
 
         buffer_key = f"{chat_id}:{msg_id}"
         if coalesce_max_delay_ms > 0 and not finished:
@@ -307,12 +338,12 @@ class TwitchAdapter(BaseChannelAdapter):
             buf["chunks"].append(chunk)
             total = "".join(buf["chunks"])
             if len(total.encode("utf-8")) >= coalesce_min_chars:
-                return await self._send_with_protection(self._flush_stream_buffer, buffer_key)
+                return await self._send_with_protection(self._flush_stream_buffer, buffer_key, progress=show_progress)
             if buf["task"] is None or buf["task"].done():
                 buf["task"] = asyncio.create_task(self._delayed_flush(buffer_key, coalesce_max_delay_ms / 1000.0))
             return DeliveryResult(success=True)
         elif finished and buffer_key in self._stream_buffers:
-            return await self._send_with_protection(self._flush_stream_buffer, buffer_key)
+            return await self._send_with_protection(self._flush_stream_buffer, buffer_key, progress=show_progress)
         else:
             if finished:
 
@@ -330,8 +361,10 @@ class TwitchAdapter(BaseChannelAdapter):
                     silent = self.config.get("silent", False)
                     fmt = format_action_line if silent else format_privmsg_line
                     prefix_len = len(f"PRIVMSG {chat_id} :")
-                    suffix = "…"
-                    text = chunk[: 510 - prefix_len - len(suffix.encode("utf-8"))] + suffix
+                    max_chars = 500 - prefix_len
+                    suffix = _progress_bar(len(chunk), max_chars) if show_progress else "…"
+                    suffix_len = len(suffix.encode("utf-8"))
+                    text = chunk[: max_chars - suffix_len] + suffix
                     if not await self._rate_limiter.acquire():
                         return DeliveryResult(success=False, error="rate_limit_exceeded")
                     self._writer.write(fmt(chat_id, text).encode("utf-8") + b"\r\n")
@@ -344,7 +377,7 @@ class TwitchAdapter(BaseChannelAdapter):
         await asyncio.sleep(delay_sec)
         await self._flush_stream_buffer(buffer_key)
 
-    async def _flush_stream_buffer(self, buffer_key: str) -> DeliveryResult:
+    async def _flush_stream_buffer(self, buffer_key: str, progress: bool = True) -> DeliveryResult:
         buf = self._stream_buffers.pop(buffer_key, None)
         if buf is None:
             return DeliveryResult(success=True)
@@ -385,6 +418,45 @@ class TwitchAdapter(BaseChannelAdapter):
     def _record_outbound(self, target: str, content: str) -> None:
         self._outbound_cache.record(target, content)
 
+    async def _resolve_broadcaster_id(self, channel: str) -> str | None:
+        channel_name = channel.lstrip("#").lower()
+        if channel_name in self._broadcaster_ids:
+            return self._broadcaster_ids[channel_name]
+
+        if not self._helix:
+            return None
+
+        user = await self._helix.get_user_by_name(channel_name)
+        if user:
+            self._broadcaster_ids[channel_name] = user["id"]
+            return user["id"]
+        return None
+
+    async def _send_via_helix(self, target: str, content: str, reply_msg_id: str | None = None) -> DeliveryResult:
+        if not self._helix:
+            return DeliveryResult(success=False, error="helix_not_available")
+
+        broadcaster_id = await self._resolve_broadcaster_id(target)
+        if not broadcaster_id:
+            return DeliveryResult(success=False, error="broadcaster_not_found")
+
+        sender_id = self._bot_user_id
+        if not sender_id:
+            return DeliveryResult(success=False, error="bot_user_id_unknown")
+
+        result = await self._helix.send_chat_message(
+            broadcaster_id=broadcaster_id,
+            sender_id=sender_id,
+            message=content,
+            reply_parent_msg_id=reply_msg_id,
+        )
+        if result is None:
+            return DeliveryResult(success=False, error="helix_send_failed")
+
+        message_id = result.get("message_id", "")
+        self._outbound_cache.record(target, content, message_id)
+        return DeliveryResult(success=True, metadata={"messageId": message_id})
+
     def get_outbound_cache(self) -> list[dict[str, Any]]:
         return self._outbound_cache.get_all()
 
@@ -838,10 +910,14 @@ class TwitchAdapter(BaseChannelAdapter):
 
         await self._refresh_token_if_needed()
 
-        await asyncio.sleep(3)
+        delay = min(3 * (2**self._reconnect_attempts), 120)
+        self._reconnect_attempts += 1
+        logger.info(f"[Twitch] IRC reconnecting in {delay}s (attempt {self._reconnect_attempts})")
+        await asyncio.sleep(delay)
         try:
             await self._connect_irc()
             self._status = ChannelStatus.CONNECTED
+            self._reconnect_attempts = 0
             logger.info("[Twitch] IRC reconnected")
         except asyncio.CancelledError:
             self._status = ChannelStatus.DISCONNECTED

backend/package/yuxi/channels/adapters/twitch/auth_provider.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import asyncio
 import time
 from abc import ABC, abstractmethod
 from typing import Any
@@ -74,26 +75,41 @@ class RefreshingAuthProvider(AuthProvider):
     async def _refresh(self) -> bool:
         if not self._client_id or not self._client_secret or not self._refresh_token:
             return False
-        try:
-            new_tokens = await refresh_access_token(self._client_id, self._client_secret, self._refresh_token)
-            if new_tokens is None:
-                return False
-            new_access = new_tokens.get("access_token", "")
-            new_refresh = new_tokens.get("refresh_token", "")
-            expires_in = new_tokens.get("expires_in", 0)
-            if new_access:
-                self._access_token = new_access
-            if new_refresh:
-                self._refresh_token = new_refresh
-            if expires_in:
-                self._token_obtained_at = time.time()
-                self._token_expires_in = expires_in
-                self._token_expires_at = self._token_obtained_at + expires_in - 300
-            logger.info(f"[TwitchAuth] Token refreshed, expires in {expires_in}s")
-            return True
-        except Exception as e:
-            logger.error(f"[TwitchAuth] Token refresh error: {e}")
-            return False
+
+        max_retries = 3
+        for attempt in range(max_retries):
+            try:
+                new_tokens = await refresh_access_token(self._client_id, self._client_secret, self._refresh_token)
+                if new_tokens is None:
+                    if attempt < max_retries - 1:
+                        delay = 2**attempt
+                        logger.warning(f"[TwitchAuth] Token refresh attempt {attempt + 1} failed, retrying in {delay}s")
+                        await asyncio.sleep(delay)
+                        continue
+                    return False
+                new_access = new_tokens.get("access_token", "")
+                new_refresh = new_tokens.get("refresh_token", "")
+                expires_in = new_tokens.get("expires_in", 0)
+                if new_access:
+                    self._access_token = new_access
+                if new_refresh:
+                    self._refresh_token = new_refresh
+                if expires_in:
+                    self._token_obtained_at = time.time()
+                    self._token_expires_in = expires_in
+                    self._token_expires_at = self._token_obtained_at + expires_in - 300
+                logger.info(f"[TwitchAuth] Token refreshed, expires in {expires_in}s")
+                return True
+            except Exception as e:
+                if attempt < max_retries - 1:
+                    delay = 2**attempt
+                    logger.warning(
+                        f"[TwitchAuth] Token refresh error (attempt {attempt + 1}): {e}, retrying in {delay}s"
+                    )
+                    await asyncio.sleep(delay)
+                else:
+                    logger.error(f"[TwitchAuth] Token refresh failed after {max_retries} attempts: {e}")
+        return False
 
     def _is_expired(self) -> bool:
         if self._token_expires_at is None:

backend/package/yuxi/channels/adapters/twitch/config_schema.py

@@ -32,6 +32,7 @@ class TwitchAccountSchema(BaseModel):
     silent: bool = False
     dm_policy: str = "pairing"
     probe_timeout_ms: int = 10000
+    prefer_helix_send: bool = False
 
 
 class TwitchConfigSchema(BaseModel):
@@ -41,7 +42,7 @@ class TwitchConfigSchema(BaseModel):
     client_secret: str = ""
     refresh_token: str = ""
     channels: list[str] = []
-    group_policy: str = Field(default="open", pattern=r"^(open|allowlist|disabled|mention_only)$")
+    group_policy: str = Field(default="open", pattern=r"^(open|allowlist|disabled|mention|mention_only)$")
     require_mention: bool = True
     allowedRoles: list[str] = []
     group_allow_from: list[str] = []
@@ -59,6 +60,7 @@ class TwitchConfigSchema(BaseModel):
     silent: bool = False
     dm_policy: str = "pairing"
     probe_timeout_ms: int = 10000
+    prefer_helix_send: bool = False
     accounts: dict[str, TwitchAccountSchema] = {}
     defaultAccount: str = ""
 
@@ -82,7 +84,7 @@ def validate_twitch_config(config: dict[str, Any]) -> list[str]:
         errors.append("at least one channel is required in 'channels' list")
 
     group_policy = config.get("group_policy", "open")
-    valid_policies = {"open", "allowlist", "disabled", "mention_only"}
+    valid_policies = {"open", "allowlist", "disabled", "mention", "mention_only"}
     if group_policy not in valid_policies:
         errors.append(f"group_policy value invalid: '{group_policy}', valid values: {valid_policies}")
 
@@ -146,6 +148,10 @@ def super_refine_twitch_config(config: dict[str, Any]) -> dict[str, Any]:
         logger.info("[TwitchConfig] Converting legacy 'allowall' to 'open'")
         refined["group_policy"] = "open"
 
+    if group_policy == "mention_only":
+        logger.info("[TwitchConfig] 'mention_only' is deprecated, use 'mention' instead (auto-converted)")
+        refined["group_policy"] = "mention"
+
     group_allow_from = refined.get("group_allow_from", refined.get("groupAllowFrom", []))
     if group_policy == "open" and ("*" in group_allow_from):
         logger.info("[TwitchConfig] group_policy is 'open', wildcard in allow_from is redundant")

backend/package/yuxi/channels/adapters/twitch/eventsub_client.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import asyncio
-from datetime import datetime, UTC
+from datetime import UTC, datetime
 from typing import Any
 
 import aiohttp
@@ -41,6 +41,11 @@ class EventSubListener:
         {"type": "channel.unban", "version": "1"},
         {"type": "channel.moderator.add", "version": "1"},
         {"type": "channel.moderator.remove", "version": "1"},
+        {"type": "stream.online", "version": "1"},
+        {"type": "stream.offline", "version": "1"},
+        {"type": "channel.update", "version": "2"},
+        {"type": "channel.chat.clear", "version": "1"},
+        {"type": "channel.chat.message_delete", "version": "1"},
     ]
 
     def __init__(
@@ -231,6 +236,18 @@ class EventSubListener:
         if subscription_type == "channel.moderator.remove":
             return self._normalize_mod_remove(event, broadcaster_id)
 
+        if subscription_type == "stream.online":
+            return self._normalize_stream_online(event, broadcaster_id)
+        if subscription_type == "stream.offline":
+            return self._normalize_stream_offline(event, broadcaster_id)
+
+        if subscription_type == "channel.update":
+            return self._normalize_channel_update(event, broadcaster_id)
+        if subscription_type == "channel.chat.clear":
+            return self._normalize_chat_clear(event, broadcaster_id)
+        if subscription_type == "channel.chat.message_delete":
+            return self._normalize_chat_message_delete(event, broadcaster_id)
+
         return None
 
     @staticmethod
@@ -527,3 +544,125 @@ class EventSubListener:
             metadata={"event": "moderator_remove"},
             timestamp=datetime.now(UTC),
         )
+
+    @staticmethod
+    def _normalize_stream_online(event: dict, broadcaster_id: str) -> ChannelMessage:
+        broadcaster_name = event.get("broadcaster_user_name", "")
+        stream_type = event.get("type", "live")
+        started_at = event.get("started_at", "")
+
+        return ChannelMessage(
+            identity=ChannelIdentity(
+                channel_id="twitch",
+                channel_type=ChannelType.TWITCH,
+                channel_user_id=broadcaster_id,
+                channel_chat_id=f"#broadcaster_{broadcaster_id}",
+                channel_message_id=f"eventsub:stream_online:{event.get('id')}",
+            ),
+            event_type=EventType.SYSTEM_EVENT,
+            message_type=MessageType.TEXT,
+            chat_type=ChatType.GROUP,
+            content=f"🔴 {broadcaster_name} went live!",
+            metadata={
+                "event": "stream_online",
+                "type": stream_type,
+                "started_at": started_at,
+            },
+            timestamp=datetime.now(UTC),
+        )
+
+    @staticmethod
+    def _normalize_stream_offline(event: dict, broadcaster_id: str) -> ChannelMessage:
+        broadcaster_name = event.get("broadcaster_user_name", "")
+
+        return ChannelMessage(
+            identity=ChannelIdentity(
+                channel_id="twitch",
+                channel_type=ChannelType.TWITCH,
+                channel_user_id=broadcaster_id,
+                channel_chat_id=f"#broadcaster_{broadcaster_id}",
+                channel_message_id=f"eventsub:stream_offline:{event.get('id')}",
+            ),
+            event_type=EventType.SYSTEM_EVENT,
+            message_type=MessageType.TEXT,
+            chat_type=ChatType.GROUP,
+            content=f"⚫ {broadcaster_name} went offline",
+            metadata={
+                "event": "stream_offline",
+            },
+            timestamp=datetime.now(UTC),
+        )
+
+    @staticmethod
+    def _normalize_channel_update(event: dict, broadcaster_id: str) -> ChannelMessage:
+        broadcaster_name = event.get("broadcaster_user_name", "")
+        title = event.get("title", "")
+        category_name = event.get("category_name", "")
+        language = event.get("language", "")
+
+        return ChannelMessage(
+            identity=ChannelIdentity(
+                channel_id="twitch",
+                channel_type=ChannelType.TWITCH,
+                channel_user_id=broadcaster_id,
+                channel_chat_id=f"#broadcaster_{broadcaster_id}",
+                channel_message_id=f"eventsub:channel_update:{event.get('id')}",
+            ),
+            event_type=EventType.SYSTEM_EVENT,
+            message_type=MessageType.TEXT,
+            chat_type=ChatType.GROUP,
+            content=f"📺 {broadcaster_name} updated channel: {title}",
+            metadata={
+                "event": "channel_update",
+                "title": title,
+                "category": category_name,
+                "language": language,
+            },
+            timestamp=datetime.now(UTC),
+        )
+
+    @staticmethod
+    def _normalize_chat_clear(event: dict, broadcaster_id: str) -> ChannelMessage:
+        return ChannelMessage(
+            identity=ChannelIdentity(
+                channel_id="twitch",
+                channel_type=ChannelType.TWITCH,
+                channel_user_id="twitch_system",
+                channel_chat_id=f"#broadcaster_{broadcaster_id}",
+                channel_message_id=f"eventsub:chat_clear:{event.get('id')}",
+            ),
+            event_type=EventType.SYSTEM_EVENT,
+            message_type=MessageType.TEXT,
+            chat_type=ChatType.GROUP,
+            content=f"🧹 Chat cleared in #{event.get('broadcaster_user_name', '')}",
+            metadata={
+                "event": "chat_clear",
+            },
+            timestamp=datetime.now(UTC),
+        )
+
+    @staticmethod
+    def _normalize_chat_message_delete(event: dict, broadcaster_id: str) -> ChannelMessage:
+        target_user_name = event.get("target_user_name", "")
+        target_user_id = event.get("target_user_id", "")
+        message_id = event.get("message_id", "")
+
+        return ChannelMessage(
+            identity=ChannelIdentity(
+                channel_id="twitch",
+                channel_type=ChannelType.TWITCH,
+                channel_user_id=target_user_id or "twitch_system",
+                channel_chat_id=f"#broadcaster_{broadcaster_id}",
+                channel_message_id=f"eventsub:msg_delete:{event.get('id')}",
+            ),
+            event_type=EventType.MESSAGE_DELETED,
+            message_type=MessageType.TEXT,
+            chat_type=ChatType.GROUP,
+            content="",
+            metadata={
+                "event": "message_delete",
+                "target_user_name": target_user_name,
+                "target_message_id": message_id,
+            },
+            timestamp=datetime.now(UTC),
+        )

backend/package/yuxi/channels/adapters/twitch/helix.py

@@ -46,16 +46,15 @@ class HelixClient:
         }
 
     async def _request_with_backoff(self, method: str, url: str, **kwargs) -> aiohttp.ClientResponse | None:
-        last_status: int | None = None
         for attempt in range(self.MAX_RETRIES):
             try:
                 async with self._ensure_session.request(method, url, **kwargs) as resp:
                     if resp.status not in self.RETRY_STATUSES or attempt == self.MAX_RETRIES - 1:
                         return resp
-                    last_status = resp.status
             except (TimeoutError, aiohttp.ClientError) as e:
                 if attempt == self.MAX_RETRIES - 1:
-                    raise
+                    logger.error(f"Helix {method} {url} connection failed after {self.MAX_RETRIES} retries: {e}")
+                    return None
                 logger.warning(f"Helix {method} {url} attempt {attempt + 1} failed: {e}")
 
             delay = min(self.BASE_BACKOFF * (2**attempt), self.MAX_BACKOFF)
@@ -190,3 +189,90 @@ class HelixClient:
         except aiohttp.ClientError as e:
             logger.error(f"EventSub delete subscription error: {e}")
             return False
+
+    async def send_chat_message(
+        self,
+        broadcaster_id: str,
+        sender_id: str,
+        message: str,
+        reply_parent_msg_id: str | None = None,
+    ) -> dict[str, Any] | None:
+        url = f"{self.BASE_URL}/chat/messages"
+        body: dict[str, Any] = {
+            "broadcaster_id": broadcaster_id,
+            "sender_id": sender_id,
+            "message": message,
+        }
+        if reply_parent_msg_id:
+            body["reply_parent_msg_id"] = reply_parent_msg_id
+
+        resp = await self._request_with_backoff(
+            "POST",
+            url,
+            headers={**self._headers(), "Content-Type": "application/json"},
+            json=body,
+        )
+        if resp is None:
+            return None
+        try:
+            if resp.status == 200:
+                data = await resp.json()
+                return data.get("data", [{}])[0] if data.get("data") else None
+            logger.error(f"Helix send_chat_message failed: {resp.status} {await resp.text()}")
+            return None
+        except aiohttp.ClientError as e:
+            logger.error(f"Helix send_chat_message connection error: {e}")
+            return None
+
+    async def delete_chat_message(
+        self,
+        broadcaster_id: str,
+        moderator_id: str,
+        message_id: str,
+    ) -> bool:
+        url = f"{self.BASE_URL}/chat/messages"
+        params = {
+            "broadcaster_id": broadcaster_id,
+            "moderator_id": moderator_id,
+            "message_id": message_id,
+        }
+        resp = await self._request_with_backoff("DELETE", url, headers=self._headers(), params=params)
+        if resp is None:
+            return False
+        try:
+            return resp.status in (200, 204)
+        except aiohttp.ClientError as e:
+            logger.error(f"Helix delete_chat_message error: {e}")
+            return False
+
+    async def send_chat_announcement(
+        self,
+        broadcaster_id: str,
+        moderator_id: str,
+        message: str,
+        color: str = "primary",
+    ) -> bool:
+        url = f"{self.BASE_URL}/chat/announcements"
+        valid_colors = {"blue", "green", "orange", "purple", "primary"}
+        if color not in valid_colors:
+            color = "primary"
+
+        body = {
+            "broadcaster_id": broadcaster_id,
+            "moderator_id": moderator_id,
+            "message": message,
+            "color": color,
+        }
+        resp = await self._request_with_backoff(
+            "POST",
+            url,
+            headers={**self._headers(), "Content-Type": "application/json"},
+            json=body,
+        )
+        if resp is None:
+            return False
+        try:
+            return resp.status in (200, 204)
+        except aiohttp.ClientError as e:
+            logger.error(f"Helix send_chat_announcement error: {e}")
+            return False

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

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from datetime import datetime, UTC
+from datetime import UTC, datetime
 from typing import Any
 
 from yuxi.channels.models import (
@@ -16,9 +16,9 @@ from yuxi.channels.models import (
 from .irc_parser import (
     ParsedIRCMessage,
     extract_nick,
+    make_irc_message_id,
     parse_badges,
     parse_emotes,
-    make_irc_message_id,
 )
 
 _FILTERED_NOTICE_MSG_IDS = frozenset(
@@ -49,7 +49,7 @@ _EVENT_MAP: dict[str, str] = {
     "announcement": "announcement",
 }
 
-_SKIP_COMMANDS = frozenset({"JOIN", "PART", "HOSTTARGET", "ROOMSTATE", "USERSTATE"})
+_SKIP_COMMANDS = frozenset({"JOIN", "PART", "HOSTTARGET", "USERSTATE"})
 
 
 def _make_identity(
@@ -178,6 +178,27 @@ def normalize_irc_message(parsed: ParsedIRCMessage) -> ChannelMessage | None:
             timestamp=datetime.now(UTC),
         )
 
+    # ---- CLEARMSG: 单条消息删除 ----
+    if command == "CLEARMSG":
+        channel = parsed.params[0] if parsed.params else ""
+        target_msg_id = tags.get("target-msg-id", "")
+        login = tags.get("login", "")
+        target_user_id = tags.get("target-user-id", "")
+
+        return ChannelMessage(
+            identity=_make_identity(target_user_id or "twitch_system", channel, target_msg_id),
+            event_type=EventType.MESSAGE_DELETED,
+            message_type=MessageType.TEXT,
+            chat_type=ChatType.GROUP,
+            content="",
+            metadata={
+                "event": "clearmsg",
+                "target_msg_id": target_msg_id,
+                "login": login,
+            },
+            timestamp=datetime.now(UTC),
+        )
+
     # ---- WHISPER: 私信 ----
     if command == "WHISPER":
         user_id = tags.get("user-id") or extract_nick(parsed.prefix)
@@ -204,6 +225,30 @@ def normalize_irc_message(parsed: ParsedIRCMessage) -> ChannelMessage | None:
             timestamp=_make_timestamp(tags.get("tmi-sent-ts")),
         )
 
+    # ---- ROOMSTATE: 房间状态变更 ----
+    if command == "ROOMSTATE":
+        channel = parsed.params[0] if parsed.params else ""
+        room_state_flags = {
+            "emote_only": tags.get("emote-only") == "1",
+            "subs_only": tags.get("subs-only") == "1",
+            "followers_only": tags.get("followers-only", "-1"),
+            "slow": tags.get("slow", "0"),
+            "r9k": tags.get("r9k") == "1",
+        }
+
+        return ChannelMessage(
+            identity=_make_identity("twitch_system", channel),
+            event_type=EventType.SYSTEM_EVENT,
+            message_type=MessageType.TEXT,
+            chat_type=ChatType.GROUP,
+            content="Room state changed",
+            metadata={
+                "event": "roomstate",
+                "room_state": room_state_flags,
+            },
+            timestamp=datetime.now(UTC),
+        )
+
     # ---- NOTICE: 系统通知 ----
     if command == "NOTICE":
         channel = parsed.params[0] if parsed.params else ""

backend/package/yuxi/channels/adapters/twitch/outbound_cache.py

@@ -10,16 +10,25 @@ class OutboundCacheManager:
         self._cache: OrderedDict[str, dict[str, Any]] = OrderedDict()
         self._max_size = max_size
 
-    def record(self, target: str, content: str) -> None:
+    def record(self, target: str, content: str, message_id: str | None = None) -> str:
         entry = {
             "channel": target,
             "content": content,
             "timestamp": time.time(),
         }
-        cache_key = f"{target}:{len(self._cache)}"
+        cache_key = message_id or f"{target}:{len(self._cache)}"
+        if message_id:
+            entry["message_id"] = message_id
         self._cache[cache_key] = entry
         while len(self._cache) > self._max_size:
             self._cache.popitem(last=False)
+        return cache_key
 
     def get_all(self) -> list[dict[str, Any]]:
         return list(self._cache.values())
+
+    def find_by_message_id(self, message_id: str) -> dict[str, Any] | None:
+        for entry in self._cache.values():
+            if entry.get("message_id") == message_id:
+                return entry
+        return None

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

@@ -18,6 +18,7 @@ def format_cap_req(capabilities: list[str]) -> str:
 
 
 def find_utf8_cut(encoded: bytes, byte_limit: int) -> int:
+    byte_limit = min(byte_limit, len(encoded))
     cut = byte_limit
     while cut > 0 and (encoded[cut - 1] & 0xC0) == 0x80:
         cut -= 1

backend/package/yuxi/channels/adapters/twitch/session.py

@@ -97,11 +97,10 @@ def check_group_policy(
             return True
         case "disabled":
             return False
-        case "mention_only":
-            require_mention = config.get("require_mention", True)
-            if not require_mention:
-                return True
+        case "mention":
             return bool(bot_name) and bot_name in content.lower()
+        case "mention_only":
+            return not config.get("require_mention", True) or (bool(bot_name) and bot_name in content.lower())
         case "allowlist":
             allowlist = config.get("group_allow_from", [])
             channels_config = config.get("channels_config", {})