refactor(googlechat): 整理并优化Google Chat适配器代码

- 调整依赖导入顺序与清理冗余空行
- 修复setup wizard步骤存储逻辑
- 新增消息编辑/转发标记、卡片控件支持
- 完善目标校验、媒体上传、配对存储功能
- 优化流式响应与错误处理逻辑
- 增强SSRF防护与配置灵活性
- 新增/扩展内置命令与卡片处理能力
- 调整媒体大小限制与类型参数

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

@@ -5,7 +5,7 @@ import json
 import os
 import time
 from collections import defaultdict
-from collections.abc import AsyncIterator
+from collections.abc import AsyncIterator, Callable
 from datetime import UTC
 from typing import Any, ClassVar
 
@@ -25,11 +25,13 @@ from yuxi.channels.models import (
     ChannelStatus,
     ChannelType,
     DeliveryResult,
+    EventType,
     HealthStatus,
 )
 from yuxi.channels.registry import register_builtin_adapter
 from yuxi.utils.logging_config import logger
 
+from . import auth, proxy
 from .formatter import format_outbound
 from .media import download_media as _download_media
 from .normalizer import is_bot_message, normalize_inbound
@@ -46,8 +48,6 @@ from .send import (
     upload_image_message,
 )
 from .streaming import StreamManager
-from . import auth
-from . import proxy
 
 _CONNECT_TIMEOUT_S = 30.0
 _PREAUTH_BODY_MAX_BYTES = 16 * 1024
@@ -106,7 +106,7 @@ class GoogleChatAdapter(BaseChannelAdapter):
     supports_markdown: ClassVar[bool] = True
     supports_streaming: ClassVar[bool] = True
     streaming_modes: ClassVar[list[str]] = ["off", "partial", "block"]
-    max_media_size_mb: ClassVar[int] = 20
+    max_media_size_mb: ClassVar[int] = 200
     min_send_interval_ms: ClassVar[int] = 1000
 
     capabilities = ChannelCapabilities(
@@ -121,7 +121,7 @@ class GoogleChatAdapter(BaseChannelAdapter):
         supports_streaming=True,
         streaming_modes=["off", "partial", "block"],
         text_chunk_limit=4000,
-        max_media_size_mb=20,
+        max_media_size_mb=200,
     )
     meta = ChannelMeta(
         id="googlechat",
@@ -144,10 +144,30 @@ class GoogleChatAdapter(BaseChannelAdapter):
         self._service_account_email: str = ""
         self._pubsub_subscription: str = ""
         self._connected_at: float | None = None
+
+        stream_cfg: dict = {}
+        if config:
+            stream_cfg = config.get("streaming", {}) if isinstance(config.get("streaming"), dict) else {}
+        block_cfg = stream_cfg.get("block", {}) if isinstance(stream_cfg, dict) else {}
+        if not isinstance(block_cfg, dict):
+            block_cfg = {}
+
+        update_interval = (config.get("streamUpdateIntervalMs") if config else None) or block_cfg.get(
+            "update_interval_ms", 800
+        )
+
+        coalesce_min = (config.get("blockStreamingCoalesceMinChars") if config else None) or block_cfg.get(
+            "coalesce_min_chars", 1500
+        )
+
+        coalesce_idle = (config.get("blockStreamingCoalesceIdleMs") if config else None) or block_cfg.get(
+            "coalesce_idle_ms", 1000
+        )
+
         self._stream_mgr = StreamManager(
-            update_interval_ms=config.get("streamUpdateIntervalMs", 800) if config else 800,
+            update_interval_ms=update_interval,
-            coalesce_min_chars=config.get("blockStreamingCoalesceMinChars", 1500) if config else 1500,
+            coalesce_min_chars=coalesce_min,
-            coalesce_idle_ms=config.get("blockStreamingCoalesceIdleMs", 1000) if config else 1000,
+            coalesce_idle_ms=coalesce_idle,
         )
         self._rate_limiter = ChatRateLimiter(ops_per_second=(config or {}).get("minSendIntervalMs", 1000) / 1000.0)
         self._circuit_breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=60.0)
@@ -175,6 +195,7 @@ class GoogleChatAdapter(BaseChannelAdapter):
         self._session_contexts: dict[str, tuple[float, dict]] = {}
         self._session_ttl_s = config.get("sessionTtlS", 1800) if config else 1800
         self._session_cleanup_task: asyncio.Task | None = None
+        self._card_handlers: dict[str, Callable] = {}
         cfg = config or {}
         self._connect_timeout_s = cfg.get("connectTimeout", _CONNECT_TIMEOUT_S)
         self._preauth_body_max_bytes = cfg.get("preauthBodyMaxBytes", _PREAUTH_BODY_MAX_BYTES)
@@ -208,6 +229,7 @@ class GoogleChatAdapter(BaseChannelAdapter):
             status = 0
             try:
                 import googleapiclient.errors
+
                 if isinstance(e, googleapiclient.errors.HttpError):
                     status = e.resp.status
             except Exception:
@@ -257,6 +279,7 @@ class GoogleChatAdapter(BaseChannelAdapter):
         self._chat_service = None
         self._credentials = None
         self._session_contexts.clear()
+        self._card_handlers.clear()
         if self._session_cleanup_task:
             self._session_cleanup_task.cancel()
             self._session_cleanup_task = None
@@ -284,6 +307,9 @@ class GoogleChatAdapter(BaseChannelAdapter):
                 error_detail = f"HTTP {status_code}: {error_detail}"
             return {"ok": False, "status": "error", "error": error_detail, "status_code": status_code}
 
+    def register_card_handler(self, action_name: str, handler: Callable) -> None:
+        self._card_handlers[action_name] = handler
+
     async def send(self, response: ChannelResponse) -> DeliveryResult:
         chat_id = response.identity.channel_chat_id
         space_name = chat_id.split("/threads/")[0]
@@ -349,13 +375,24 @@ class GoogleChatAdapter(BaseChannelAdapter):
         except CircuitBreakerOpenError:
             return DeliveryResult(success=False, error="Circuit breaker open")
 
-    async def edit_message(self, chat_id: str, msg_id: str, content: str) -> DeliveryResult:
+    async def edit_message(
+        self,
+        chat_id: str,
+        msg_id: str,
+        content: str,
+        cards: list[dict] | None = None,
+        update_fields: list[str] | None = None,
+    ) -> DeliveryResult:
         space_name = chat_id.split("/threads/")[0]
         await self._rate_limiter.acquire(space_name)
         body: dict[str, Any] = {"text": content}
+        if cards:
+            body["cards_v2"] = cards
+            update_fields = list(set((update_fields or ["text"]) + ["cardsV2"]))
+        mask = ",".join(update_fields or ["text"])
 
         async def _do_send():
-            return await update_message(self._chat_service, msg_id, body)
+            return await update_message(self._chat_service, msg_id, body, update_mask=mask)
 
         try:
             return await self._circuit_breaker.call(_do_send)
@@ -374,7 +411,7 @@ class GoogleChatAdapter(BaseChannelAdapter):
         except CircuitBreakerOpenError:
             return DeliveryResult(success=False, error="Circuit breaker open")
 
-    async def send_reaction(self, chat_id: str, msg_id: str, emoji: str) -> DeliveryResult:
+    async def send_reaction(self, chat_id: str, msg_id: str, emoji: str, custom_emoji_uid: str = "") -> DeliveryResult:
         reactions_enabled = self._actions.get("reactions", True)
         if not reactions_enabled:
             return DeliveryResult(success=False, error="Agent reactions disabled by actions.reactions config")
@@ -383,7 +420,7 @@ class GoogleChatAdapter(BaseChannelAdapter):
         await self._rate_limiter.acquire(space_name)
 
         async def _do_send():
-            return await send_reaction(self._chat_service, msg_id, emoji)
+            return await send_reaction(self._chat_service, msg_id, emoji, custom_emoji_uid=custom_emoji_uid)
 
         try:
             return await self._circuit_breaker.call(_do_send)
@@ -410,25 +447,10 @@ class GoogleChatAdapter(BaseChannelAdapter):
     async def send_typing_indicator(self, chat_id: str) -> DeliveryResult:
         space_name = chat_id.split("/threads/")[0]
         await self._rate_limiter.acquire(space_name)
-        body = format_outbound(
+        msg_id = await self._stream_mgr.create_typing_message(self._chat_service, chat_id, self._bot_user or "Bot")
-            ChannelResponse(
+        if msg_id:
-                identity=ChannelIdentity(
+            return DeliveryResult(success=True, message_id=msg_id)
-                    channel_id=self.channel_id,
+        return DeliveryResult(success=False, error="Failed to create typing indicator")
-                    channel_type=self.channel_type,
-                    channel_user_id="",
-                    channel_chat_id=chat_id,
-                ),
-                content="",
-            )
-        )
-
-        async def _do_send():
-            return await send_message(self._chat_service, chat_id, body)
-
-        try:
-            return await self._circuit_breaker.call(_do_send)
-        except CircuitBreakerOpenError:
-            return DeliveryResult(success=False, error="Circuit breaker open")
 
     async def send_stream_chunk(
         self,
@@ -876,6 +898,29 @@ class GoogleChatAdapter(BaseChannelAdapter):
             raise ChannelNotConnectedError()
         return await _download_media(self._chat_service, file_id)
 
+    async def get_message(self, msg_id: str) -> dict | None:
+        if not self._chat_service:
+            raise ChannelNotConnectedError()
+        from .send import get_message as _get_message
+
+        return await _get_message(self._chat_service, msg_id)
+
+    async def list_messages(
+        self, space_name: str, page_size: int = 50, page_token: str | None = None
+    ) -> tuple[list[dict], str | None]:
+        if not self._chat_service:
+            raise ChannelNotConnectedError()
+        from .send import list_messages as _list_messages
+
+        return await _list_messages(self._chat_service, space_name, page_size, page_token)
+
+    async def list_spaces(self, page_size: int = 50) -> tuple[list[dict], str | None]:
+        if not self._chat_service:
+            raise ChannelNotConnectedError()
+        from .directory import list_spaces as _list_spaces
+
+        return await _list_spaces(self._chat_service, page_size)
+
     async def _refresh_token_if_needed(self) -> bool:
         if not self._credentials or not self._credentials.valid:
             return False
@@ -947,6 +992,16 @@ class GoogleChatAdapter(BaseChannelAdapter):
                                 queue.task_done()
                                 continue
 
+                    if msg.event_type == EventType.CARD_ACTION:
+                        action_data = json.loads(msg.content) if msg.content else {}
+                        card_handler = self._card_handlers.get(action_data.get("function", ""))
+                        if card_handler:
+                            await card_handler(msg, action_data)
+                        else:
+                            logger.warning(f"No handler for card action: {action_data}")
+                        queue.task_done()
+                        continue
+
                     if msg.message_type == "command" and msg.metadata.get("slash_command"):
                         handled = await self._handle_slash_command(msg, space_name)
                         if handled:
@@ -974,10 +1029,7 @@ class GoogleChatAdapter(BaseChannelAdapter):
             while self._status == ChannelStatus.CONNECTED:
                 await asyncio.sleep(300)
                 now = time.monotonic()
-                expired = [
+                expired = [k for k, (ts, _) in self._session_contexts.items() if now - ts > self._session_ttl_s]
-                    k for k, (ts, _) in self._session_contexts.items()
-                    if now - ts > self._session_ttl_s
-                ]
                 for k in expired:
                     del self._session_contexts[k]
                 if expired:
@@ -996,22 +1048,24 @@ class GoogleChatAdapter(BaseChannelAdapter):
 
     async def _handle_slash_command(self, msg: ChannelMessage, space_name: str) -> bool:
         command = msg.metadata.get("slash_command", "")
+        if not command:
+            return False
+
         chat_id = msg.identity.channel_chat_id
 
+        def _make_identity():
+            return ChannelIdentity(
+                channel_id=self.channel_id,
+                channel_type=self.channel_type,
+                channel_user_id="",
+                channel_chat_id=chat_id,
+            )
+
         if command == "/help":
             from .slash_commands import get_command_help
 
             help_text = get_command_help()
-            response = ChannelResponse(
+            await self.send(ChannelResponse(identity=_make_identity(), content=help_text))
-                identity=ChannelIdentity(
-                    channel_id=self.channel_id,
-                    channel_type=self.channel_type,
-                    channel_user_id="",
-                    channel_chat_id=chat_id,
-                ),
-                content=help_text,
-            )
-            await self.send(response)
             return True
 
         if command == "/status":
@@ -1024,19 +1078,57 @@ class GoogleChatAdapter(BaseChannelAdapter):
                 f"• 待处理流消息: {self._stream_mgr.pending_count}\n"
                 f"• 熔断器: {self._circuit_breaker.state}"
             )
-            response = ChannelResponse(
+            await self.send(ChannelResponse(identity=_make_identity(), content=status_text))
-                identity=ChannelIdentity(
-                    channel_id=self.channel_id,
-                    channel_type=self.channel_type,
-                    channel_user_id="",
-                    channel_chat_id=chat_id,
-                ),
-                content=status_text,
-            )
-            await self.send(response)
             return True
 
-        return False
+        if command == "/reset":
+            thread_id = msg.metadata.get("thread_id", chat_id)
+            self._session_contexts.pop(thread_id, None)
+            self._touch_session(thread_id)
+            await self.send(
+                ChannelResponse(
+                    identity=_make_identity(),
+                    content="会话上下文已清除，可以开始新的对话。",
+                )
+            )
+            return True
+
+        if command == "/history":
+            thread_id = msg.metadata.get("thread_id", chat_id)
+            _, ctx = self._session_contexts.get(thread_id, (0, {}))
+            history = ctx.get("history_summary", "暂无对话历史")
+            await self.send(
+                ChannelResponse(
+                    identity=_make_identity(),
+                    content=f"*对话历史摘要*\n{history}",
+                )
+            )
+            return True
+
+        if command == "/context":
+            thread_id = msg.metadata.get("thread_id", chat_id)
+            _, ctx = self._session_contexts.get(thread_id, (0, {}))
+            ctx_info = json.dumps(ctx, ensure_ascii=False, indent=2)
+            await self.send(
+                ChannelResponse(
+                    identity=_make_identity(),
+                    content=f"*当前上下文*\n```json\n{ctx_info}\n```",
+                )
+            )
+            return True
+
+        if command == "/summary":
+            msg.metadata["request_summary"] = True
+            await self._handle_message(msg)
+            return True
+
+        await self.send(
+            ChannelResponse(
+                identity=_make_identity(),
+                content=f"命令 `{command}` 暂未实现，即将支持。发送 `/help` 查看可用命令。",
+            )
+        )
+        return True
 
     def _convert_addon_event(self, common_event: dict) -> dict:
         event_type = common_event.get("type", "MESSAGE")
@@ -1431,6 +1523,21 @@ class GoogleChatAdapter(BaseChannelAdapter):
                 push_config = pubsub_v1.PushConfig(push_endpoint=webhook_url)
                 topic_path = subscriber.topic_path(project_id, topic)
 
+                publisher = pubsub_v1.PublisherClient(credentials=self._credentials)
+                try:
+                    publisher.get_topic(topic=topic_path)
+                    logger.info(f"Pub/Sub topic exists: {topic_path}")
+                except Exception:
+                    try:
+                        publisher.create_topic(name=topic_path)
+                        logger.info(f"Created Pub/Sub topic: {topic_path}")
+                    except Exception as topic_err:
+                        if hasattr(topic_err, "code") and getattr(topic_err, "code") == 409:
+                            logger.info(f"Pub/Sub topic already exists (race): {topic_path}")
+                        else:
+                            logger.error(f"Failed to create Pub/Sub topic {topic}: {topic_err}")
+                            return
+
                 try:
                     subscriber.create_subscription(
                         name=sub_path,

backend/package/yuxi/channels/adapters/googlechat/cards.py

@@ -90,6 +90,15 @@ def build_form_card(title: str, fields: list[dict], submit_action: str) -> dict:
                     }
                 }
             )
+        elif field_input_type in ("DATE_AND_TIME", "DATE_ONLY", "TIME_ONLY"):
+            widgets.append(
+                _make_date_time_picker(
+                    name=field.get("name", ""),
+                    label=field.get("label", ""),
+                    type_=field_input_type,
+                    value_ms_epoch=field.get("value_ms_epoch"),
+                )
+            )
         elif field_input_type == "DIVIDER":
             widgets.append({"divider": {}})
         else:
@@ -216,3 +225,41 @@ def _make_button(text: str, action: str, green: bool = False) -> dict:
         "color": color,
         "onClick": {"action": {"function": action}},
     }
+
+
+def _make_date_time_picker(
+    name: str,
+    label: str = "",
+    type_: str = "DATE_AND_TIME",
+    value_ms_epoch: int | None = None,
+) -> dict:
+    widget = {
+        "dateTimePicker": {
+            "name": name,
+            "label": label,
+            "type": type_,
+        }
+    }
+    if value_ms_epoch is not None:
+        widget["dateTimePicker"]["valueMsEpoch"] = str(value_ms_epoch)
+    return widget
+
+
+def _make_grid(column_count: int, items: list[dict]) -> dict:
+    return {
+        "grid": {
+            "columnCount": column_count,
+            "items": [{"widgets": [item]} for item in items],
+        }
+    }
+
+
+def _make_columns(column_items: list[list[dict]]) -> dict:
+    return {"columns": {"columnItems": [{"widgets": col_widgets} for col_widgets in column_items]}}
+
+
+def _make_link_button(text: str, url: str) -> dict:
+    return {
+        "text": text,
+        "onClick": {"openLink": {"url": url}},
+    }

backend/package/yuxi/channels/adapters/googlechat/directory.py

@@ -1,6 +1,30 @@
 from __future__ import annotations
 
-from typing import Any
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from googleapiclient.discovery import Resource
+
+from yuxi.utils.logging_config import logger
+
+
+async def list_spaces(
+    chat_service: Resource,
+    page_size: int = 50,
+    page_token: str | None = None,
+    filter_str: str | None = None,
+) -> tuple[list[dict], str | None]:
+    kwargs: dict[str, Any] = {"pageSize": page_size}
+    if page_token:
+        kwargs["pageToken"] = page_token
+    if filter_str:
+        kwargs["filter"] = filter_str
+    try:
+        result = chat_service.spaces().list(**kwargs).execute()
+        return result.get("spaces", []), result.get("nextPageToken")
+    except Exception as e:
+        logger.warning(f"list_spaces failed: {e}")
+        return [], None
 
 
 def list_peers(allow_from: list[str]) -> list[dict[str, Any]]:

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

@@ -23,7 +23,10 @@ def format_outbound(response: ChannelResponse) -> dict[str, Any]:
         for att in response.attachments:
             if att.url and att.type in ("image", "IMAGE", "video", "VIDEO"):
                 body.setdefault("cards_v2", [])
-                body["cards_v2"].append({"card": {"sections": [{"widgets": [{"image": {"imageUrl": att.url}}]}]}})
+                img_widget = {"image": {"imageUrl": att.url}}
+                if att.filename:
+                    img_widget["image"]["altText"] = att.filename
+                body["cards_v2"].append({"card": {"sections": [{"widgets": [img_widget]}]}})
 
     if response.metadata and response.metadata.get("thread_key"):
         body["messageReplyOption"] = "REPLY_MESSAGE_FALLBACK_TO_NEW_THREAD"

backend/package/yuxi/channels/adapters/googlechat/media.py

@@ -8,65 +8,11 @@ if TYPE_CHECKING:
     from googleapiclient.discovery import Resource
 
 from yuxi.channels.exceptions import DeliveryFailedError
-from yuxi.channels.models import DeliveryResult
 from yuxi.utils.logging_config import logger
 
 _MAX_DOWNLOAD_SIZE_BYTES = 100 * 1024 * 1024
 
 
-async def upload_image(
-    chat_service: Resource,
-    chat_id: str,
-    image_data: bytes,
-    filename: str = "image.png",
-    caption: str = "",
-) -> DeliveryResult:
-    from googleapiclient.http import MediaIoBaseUpload
-
-    try:
-        media = MediaIoBaseUpload(
-            io.BytesIO(image_data),
-            mimetype="image/png",
-            resumable=True,
-        )
-        body = {"text": caption or " "}
-        result = chat_service.spaces().messages().create(parent=chat_id, body=body, media_body=media).execute()
-        return DeliveryResult(
-            success=True,
-            message_id=result.get("name", ""),
-        )
-    except Exception as e:
-        logger.error(f"upload_image failed: {e}")
-        return DeliveryResult(success=False, error=str(e))
-
-
-async def upload_file(
-    chat_service: Resource,
-    chat_id: str,
-    file_data: bytes,
-    filename: str,
-    mime_type: str = "application/octet-stream",
-    caption: str = "",
-) -> DeliveryResult:
-    from googleapiclient.http import MediaIoBaseUpload
-
-    try:
-        media = MediaIoBaseUpload(
-            io.BytesIO(file_data),
-            mimetype=mime_type,
-            resumable=True,
-        )
-        body = {"text": caption or " "}
-        result = chat_service.spaces().messages().create(parent=chat_id, body=body, media_body=media).execute()
-        return DeliveryResult(
-            success=True,
-            message_id=result.get("name", ""),
-        )
-    except Exception as e:
-        logger.error(f"upload_file failed: {e}")
-        return DeliveryResult(success=False, error=str(e))
-
-
 async def download_media(chat_service: Resource, file_id: str) -> bytes:
     try:
         file_info = json.loads(file_id)

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

@@ -90,6 +90,8 @@ def normalize_inbound(
             "route_envelope": route_envelope,
             "slash_command": command,
             "slash_args": command_args,
+            "is_edited": detect_message_edit(message),
+            "is_forwarded": is_forwarded_message(message),
         },
     )
 

backend/package/yuxi/channels/adapters/googlechat/pairing.py

@@ -1,5 +1,8 @@
 from __future__ import annotations
 
+import json
+import os
+import threading
 from typing import TYPE_CHECKING
 
 if TYPE_CHECKING:
@@ -16,6 +19,74 @@ _PAIRING_CHALLENGE_MESSAGE = (
 )
 
 
+class PairingStore:
+    def __init__(self, storage_path: str = ""):
+        self._paired_users: dict[str, str] = {}
+        self._lock = threading.Lock()
+        self._storage_path = storage_path or os.path.join(
+            os.path.expanduser("~"), ".forcepilot", "googlechat_pairing.json"
+        )
+        self._load()
+
+    def _load(self) -> None:
+        try:
+            if os.path.exists(self._storage_path):
+                with open(self._storage_path, encoding="utf-8") as f:
+                    self._paired_users = json.load(f)
+                logger.debug(f"Loaded {len(self._paired_users)} paired users from {self._storage_path}")
+        except Exception as e:
+            logger.warning(f"Failed to load pairing store: {e}")
+            self._paired_users = {}
+
+    def _save(self) -> None:
+        try:
+            os.makedirs(os.path.dirname(self._storage_path), exist_ok=True)
+            with open(self._storage_path, "w", encoding="utf-8") as f:
+                json.dump(self._paired_users, f, indent=2)
+        except Exception as e:
+            logger.warning(f"Failed to save pairing store: {e}")
+
+    def is_paired(self, space_id: str) -> bool:
+        with self._lock:
+            return space_id in self._paired_users
+
+    def pair(self, space_id: str, user_id: str) -> None:
+        with self._lock:
+            self._paired_users[space_id] = user_id
+            self._save()
+
+    def unpair(self, space_id: str) -> bool:
+        with self._lock:
+            if space_id in self._paired_users:
+                del self._paired_users[space_id]
+                self._save()
+                return True
+            return False
+
+    def get_paired_user(self, space_id: str) -> str | None:
+        with self._lock:
+            return self._paired_users.get(space_id)
+
+    def list_paired_spaces(self) -> list[str]:
+        with self._lock:
+            return list(self._paired_users.keys())
+
+    def clear(self) -> None:
+        with self._lock:
+            self._paired_users.clear()
+            self._save()
+
+
+_global_pairing_store: PairingStore | None = None
+
+
+def get_pairing_store(storage_path: str = "") -> PairingStore:
+    global _global_pairing_store
+    if _global_pairing_store is None:
+        _global_pairing_store = PairingStore(storage_path)
+    return _global_pairing_store
+
+
 async def send_pairing_challenge(
     chat_service: Resource,
     space_name: str,

backend/package/yuxi/channels/adapters/googlechat/policy.py

@@ -4,7 +4,7 @@ from dataclasses import dataclass, field
 from enum import StrEnum
 from typing import Any
 
-from yuxi.channels.models import ChatType, ChannelMessage
+from yuxi.channels.models import ChannelMessage, ChatType
 from yuxi.utils.logging_config import logger
 
 

backend/package/yuxi/channels/adapters/googlechat/pubsub_.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
-import json
 import base64
+import json
 
 from yuxi.utils.logging_config import logger
 

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

@@ -55,6 +55,9 @@ def _classify_error(e: Exception, account_id: str = "default") -> None:
         if isinstance(e, HttpError):
             status = e.resp.status
             if status == 429:
+                retry_after = _extract_retry_after(e)
+                if retry_after:
+                    raise ChannelRateLimitError(f"Rate limited, retry after {retry_after}s")
                 raise ChannelRateLimitError()
             if status in (401, 403):
                 reason = _extract_rate_limit_reason(e)
@@ -81,6 +84,19 @@ def _extract_rate_limit_reason(e: Exception) -> bool:
     return False
 
 
+def _extract_retry_after(e: Exception) -> int | None:
+    try:
+        resp = e.resp if hasattr(e, "resp") else None
+        if resp is None:
+            return None
+        retry_after = resp.get("Retry-After") or resp.get("retry-after")
+        if retry_after:
+            return int(retry_after)
+    except (ValueError, TypeError):
+        pass
+    return None
+
+
 def _backoff_with_jitter(retry_count: int) -> float:
     delay = min((2**retry_count) + random.uniform(0, 1), _MAX_BACKOFF_S)
     return delay
@@ -195,10 +211,14 @@ async def send_reaction(
     chat_service: Resource,
     message_id: str,
     emoji: str,
+    custom_emoji_uid: str = "",
     account_id: str = "default",
 ) -> DeliveryResult:
     async def _do_send():
-        body = {"emoji": {"unicode": emoji}}
+        if custom_emoji_uid:
+            body = {"emoji": {"customEmoji": {"uid": custom_emoji_uid}}}
+        else:
+            body = {"emoji": {"unicode": emoji}}
         (chat_service.spaces().messages().reactions().create(parent=message_id, body=body).execute())
         return DeliveryResult(success=True)
 
@@ -302,6 +322,39 @@ async def find_direct_message_space(
         return None
 
 
+async def get_message(
+    chat_service: Resource,
+    message_id: str,
+) -> dict | None:
+    try:
+        result = chat_service.spaces().messages().get(name=message_id).execute()
+        return result
+    except Exception as e:
+        logger.warning(f"get_message failed for {message_id}: {e}")
+        return None
+
+
+async def list_messages(
+    chat_service: Resource,
+    space_name: str,
+    page_size: int = 50,
+    page_token: str | None = None,
+    filter_str: str | None = None,
+) -> tuple[list[dict], str | None]:
+    kwargs: dict[str, Any] = {"parent": space_name, "pageSize": page_size}
+    if page_token:
+        kwargs["pageToken"] = page_token
+    if filter_str:
+        kwargs["filter"] = filter_str
+
+    try:
+        result = chat_service.spaces().messages().list(**kwargs).execute()
+        return result.get("messages", []), result.get("nextPageToken")
+    except Exception as e:
+        logger.warning(f"list_messages failed for {space_name}: {e}")
+        return [], None
+
+
 async def resolve_outbound_space(
     chat_service: Resource,
     target: str,

backend/package/yuxi/channels/adapters/googlechat/setup.py

@@ -2,7 +2,6 @@ from __future__ import annotations
 
 from typing import Any
 
-
 _SETUP_GUIDE = """
 Google Chat Setup Guide
 =======================
@@ -64,8 +63,8 @@ class SetupWizard:
         return {"step": self._current_step, "steps": len(self._steps), "data": self.current_step}
 
     def next(self, answer: Any = None) -> dict[str, Any]:
-        if self._current_step > 0 and answer is not None:
+        if answer is not None:
-            self._results[self._steps[self._current_step - 1]["id"]] = answer
+            self._results[self._steps[self._current_step]["id"]] = answer
 
         self._current_step += 1
         if self._current_step >= len(self._steps):

backend/package/yuxi/channels/adapters/googlechat/slash_commands.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-
 SLASH_COMMAND_MAP: dict[str, str] = {
     "/reset": "清除当前会话上下文，重新开始对话",
     "/history": "查看当前会话的对话历史摘要",

backend/package/yuxi/channels/adapters/googlechat/ssrf.py

@@ -2,70 +2,69 @@ from __future__ import annotations
 
 from typing import Any
 
-from yuxi.utils.logging_config import logger
+from yuxi.channels.auth.ssrf_guard import (
-
+    apply_ssrf_guard_defaults,
-_UNSAFE_TRANSPORT_FIELDS = frozenset(
+)
-    {
+from yuxi.channels.auth.ssrf_guard import (
-        "agent",
+    check_response_size as _check_response_size,
-        "cert",
+)
-        "cert_file",
+from yuxi.channels.auth.ssrf_guard import (
-        "key",
+    check_response_text as _check_response_text,
-        "key_file",
+)
-        "dispatcher",
+from yuxi.channels.auth.ssrf_guard import (
-        "proxy",
+    is_private_url as _is_private_url,
-        "session",
+)
-    }
+from yuxi.channels.auth.ssrf_guard import (
+    sanitize_transport_kwargs as _sanitize_transport_kwargs,
+)
+from yuxi.channels.auth.ssrf_guard import (
+    validate_url_with_whitelist as _validate_url_with_whitelist,
 )
 
-_AUTH_RESPONSE_MAX_BYTES = 1 * 1024 * 1024
+_GOOGLE_CHAT_ALLOWED_HOSTS = [
+    "*.googleapis.com",
+    "*.google.com",
+    "chat.googleapis.com",
+    "oauth2.googleapis.com",
+]
 
 
 def sanitize_google_auth_init(kwargs: dict[str, Any]) -> dict[str, Any]:
     cleaned: dict[str, Any] = {}
-    removed: list[str] = []
     for key, value in kwargs.items():
-        if key in _UNSAFE_TRANSPORT_FIELDS:
+        if key not in apply_ssrf_guard_defaults():
-            removed.append(key)
+            cleaned[key] = value
-            continue
+    safe = _sanitize_transport_kwargs(cleaned)
-        cleaned[key] = value
+    unsafe_defaults = apply_ssrf_guard_defaults()
-    if removed:
+    for k, v in unsafe_defaults.items():
-        logger.warning(f"SSRF guard: removed unsafe transport fields: {removed}")
+        safe.setdefault(k, v)
-    return cleaned
+    return safe
 
 
 def sanitize_request_kwargs(kwargs: dict[str, Any]) -> dict[str, Any]:
-    cleaned: dict[str, Any] = {}
+    return _sanitize_transport_kwargs(kwargs)
-    removed: list[str] = []
-    for key, value in kwargs.items():
-        if key in _UNSAFE_TRANSPORT_FIELDS:
-            removed.append(key)
-            continue
-        cleaned[key] = value
-    if removed:
-        logger.warning(f"SSRF guard: removed unsafe request fields: {removed}")
-    return cleaned
 
 
 def apply_ssrf_guard() -> dict[str, Any]:
-    return {
+    return apply_ssrf_guard_defaults()
-        "agent": None,
-        "cert": None,
-        "cert_file": None,
-        "key": None,
-        "key_file": None,
-    }
 
 
 def check_auth_response_size(data: bytes) -> bytes:
-    if len(data) > _AUTH_RESPONSE_MAX_BYTES:
+    return _check_response_size(data)
-        logger.warning(
-            f"Google Auth response exceeds size limit: {len(data)} bytes > {_AUTH_RESPONSE_MAX_BYTES} bytes, truncating"
-        )
-        return data[:_AUTH_RESPONSE_MAX_BYTES]
-    return data
 
 
 def check_auth_response_text(text: str) -> str:
-    data = text.encode("utf-8", errors="replace")
+    return _check_response_text(text)
-    truncated = check_auth_response_size(data)
+
-    return truncated.decode("utf-8", errors="replace")
+
+def validate_outbound_url(url: str, allow_private: bool = False) -> str:
+    return _validate_url_with_whitelist(
+        url,
+        allowed_hosts=_GOOGLE_CHAT_ALLOWED_HOSTS,
+        enforce_https=True,
+        allow_private=allow_private,
+    )
+
+
+def is_private_target(url: str) -> bool:
+    return _is_private_url(url)

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

@@ -10,7 +10,7 @@ if TYPE_CHECKING:
 from yuxi.channels.models import DeliveryResult
 from yuxi.utils.logging_config import logger
 
-from .send import update_message, send_message, delete_message
+from .send import delete_message, send_message, update_message
 
 DEFAULT_UPDATE_INTERVAL_MS = 800
 DEFAULT_COALESCE_MIN_CHARS = 1500
@@ -183,7 +183,10 @@ class StreamManager:
                 self._typing_messages.pop(chat_id, None)
                 self._typing_replaced.add(chat_id)
 
-        body = {"text": text if finished else text + " ⏳"}
+        if self._coalesce_min_chars > 0 and not finished:
+            body = {"text": text + "\n\n_正在生成中..._"}
+        else:
+            body = {"text": text if finished else text + " ⏳"}
         result = await update_message(chat_service, message_name, body)
 
         if result.success:

backend/package/yuxi/channels/adapters/googlechat/target.py

@@ -3,9 +3,24 @@ from __future__ import annotations
 import re
 from typing import Any
 
-
 _CHANNEL_PREFIXES = ("googlechat:", "gchat:", "google-chat:")
 
+_VALID_GC_RESOURCE_PREFIXES = frozenset({"spaces/", "users/"})
+_MESSAGES_SUFFIX_RE = re.compile(r"/messages/[^/]+$")
+
+
+def strip_message_suffix(target: str) -> tuple[str, bool]:
+    stripped = _MESSAGES_SUFFIX_RE.sub("", target)
+    return stripped, stripped != target
+
+
+def detect_deprecated_target(target: str) -> str | None:
+    cleaned = _strip_channel_prefix(target)
+    for prefix in _VALID_GC_RESOURCE_PREFIXES:
+        if cleaned.startswith(prefix):
+            return None
+    return f"Target '{target}' does not use a valid Google Chat resource prefix (spaces/ or users/)"
+
 
 def normalize_googlechat_target(target: str) -> dict[str, Any]:
     if not target: