feat(googlechat): 实现 Google Chat 适配器完整功能集

新增 Google Chat 对接的全套工具模块，包括：
- 会话线程管理、消息解析与格式化
- Pub/Sub 消息解码、提及和命令识别
- 权限审批、目录管理和审计日志
- 消息缓存、媒体上传下载和 SSFR 防护
- 策略配置、卡片构建和流式回复支持

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

@@ -0,0 +1,57 @@
+from .adapter import GoogleChatAdapter
+from .approval_auth import (
+    build_exec_approval_context,
+    is_fully_approved,
+    normalize_approver_id,
+    record_approval_decision,
+    resolve_approver_ids,
+)
+from .auth import GoogleChatCertCache, get_cert_cache, verify_project_number_token
+from .directory import list_groups, list_peers, normalize_id
+from .doctor import run_diagnostics
+from .pairing import check_pairing_approval, send_pairing_challenge
+from .proxy import build_google_auth_request, build_proxies_dict, resolve_proxy_config, resolve_tls_config
+from .secret_contract import get_secret_contracts
+from .setup import SetupWizard, get_setup_guide
+from .ssrf import (
+    apply_ssrf_guard,
+    check_auth_response_size,
+    check_auth_response_text,
+    sanitize_google_auth_init,
+    sanitize_request_kwargs,
+)
+from .target import is_googlechat_space_target, is_googlechat_user_target, normalize_googlechat_target, resolve_targets
+
+__all__ = [
+    "GoogleChatAdapter",
+    "send_pairing_challenge",
+    "check_pairing_approval",
+    "sanitize_google_auth_init",
+    "sanitize_request_kwargs",
+    "apply_ssrf_guard",
+    "check_auth_response_size",
+    "check_auth_response_text",
+    "normalize_approver_id",
+    "resolve_approver_ids",
+    "build_exec_approval_context",
+    "record_approval_decision",
+    "is_fully_approved",
+    "list_peers",
+    "list_groups",
+    "normalize_id",
+    "GoogleChatCertCache",
+    "get_cert_cache",
+    "verify_project_number_token",
+    "run_diagnostics",
+    "build_google_auth_request",
+    "build_proxies_dict",
+    "resolve_proxy_config",
+    "resolve_tls_config",
+    "get_secret_contracts",
+    "SetupWizard",
+    "get_setup_guide",
+    "is_googlechat_space_target",
+    "is_googlechat_user_target",
+    "normalize_googlechat_target",
+    "resolve_targets",
+]

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

@@ -0,0 +1,71 @@
+from __future__ import annotations
+
+from typing import Any
+
+from yuxi.utils.logging_config import logger
+
+
+def normalize_approver_id(raw: str) -> str:
+    if not raw:
+        return ""
+    if raw.startswith("users/") or raw.startswith("user:"):
+        return raw
+    if "@" in raw:
+        return f"users/{raw}"
+    return f"users/{raw}"
+
+
+def resolve_approver_ids(
+    dm_allow_from: list[str],
+    default_to: str = "",
+) -> list[str]:
+    approvers: list[str] = []
+    for entry in dm_allow_from:
+        if entry == "*":
+            continue
+        normalized = normalize_approver_id(entry)
+        if normalized:
+            approvers.append(normalized)
+    if not approvers and default_to:
+        approvers.append(normalize_approver_id(default_to))
+    return approvers
+
+
+def build_exec_approval_context(
+    action_id: str,
+    approvers: list[str],
+    metadata: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    return {
+        "action_id": action_id,
+        "approvers": approvers,
+        "approved_by": [],
+        "rejected_by": [],
+        "metadata": metadata or {},
+    }
+
+
+def record_approval_decision(
+    context: dict[str, Any],
+    user_id: str,
+    approved: bool,
+) -> bool:
+    normalized = normalize_approver_id(user_id)
+    if normalized not in context.get("approvers", []):
+        logger.warning(f"User {normalized} is not an authorized approver")
+        return False
+    if approved:
+        if normalized in context.get("approved_by", []):
+            return True
+        context.setdefault("approved_by", []).append(normalized)
+    else:
+        if normalized in context.get("rejected_by", []):
+            return True
+        context.setdefault("rejected_by", []).append(normalized)
+    return True
+
+
+def is_fully_approved(context: dict[str, Any]) -> bool:
+    approvers = context.get("approvers", [])
+    approved = context.get("approved_by", [])
+    return bool(approvers) and set(approvers) == set(approved)

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

@@ -0,0 +1,60 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from googleapiclient.discovery import Resource
+
+from yuxi.utils.logging_config import logger
+
+
+async def check_space_membership(
+    chat_service: Resource,
+    space_name: str,
+    user_name: str,
+) -> bool:
+    try:
+        membership = chat_service.spaces().members().get(name=f"{space_name}/members/{user_name}").execute()
+        return membership is not None
+    except Exception as e:
+        _log_audit_error("check_space_membership", space_name, e)
+        return False
+
+
+async def list_space_members(
+    chat_service: Resource,
+    space_name: str,
+) -> list[dict]:
+    try:
+        result = chat_service.spaces().members().list(parent=space_name).execute()
+        return result.get("memberships", [])
+    except Exception as e:
+        _log_audit_error("list_space_members", space_name, e)
+        return []
+
+
+async def get_member_role(
+    chat_service: Resource,
+    space_name: str,
+    user_name: str,
+) -> str | None:
+    try:
+        membership = chat_service.spaces().members().get(name=f"{space_name}/members/{user_name}").execute()
+        return membership.get("role")
+    except Exception as e:
+        _log_audit_error("get_member_role", space_name, e)
+        return None
+
+
+def _log_audit_error(operation: str, space_name: str, error: Exception) -> None:
+    try:
+        from googleapiclient.errors import HttpError
+
+        if isinstance(error, HttpError):
+            logger.warning(
+                f"Audit operation '{operation}' failed for {space_name}: HTTP {error.resp.status} {error.resp.reason}"
+            )
+            return
+    except ImportError:
+        pass
+    logger.warning(f"Audit operation '{operation}' failed for {space_name}: {error}")

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

@@ -0,0 +1,82 @@
+from __future__ import annotations
+
+import json
+import time
+
+import httpx
+
+from yuxi.utils.logging_config import logger
+
+_CERT_URL = "https://www.googleapis.com/service_accounts/v1/metadata/x509/chat@system.gserviceaccount.com"
+_CERT_CACHE_TTL_S = 600
+_CERT_FETCH_TIMEOUT_S = 10.0
+
+
+class GoogleChatCertCache:
+    def __init__(self, ttl_s: int = _CERT_CACHE_TTL_S):
+        self._ttl = ttl_s
+        self._cache: dict[str, str] | None = None
+        self._cache_at: float = 0.0
+
+    @property
+    def is_valid(self) -> bool:
+        return self._cache is not None and (time.monotonic() - self._cache_at) < self._ttl
+
+    async def get_certs(self) -> dict[str, str]:
+        if self.is_valid and self._cache is not None:
+            return dict(self._cache)
+
+        try:
+            async with httpx.AsyncClient(timeout=_CERT_FETCH_TIMEOUT_S) as client:
+                resp = await client.get(_CERT_URL)
+                resp.raise_for_status()
+                certs: dict[str, str] = resp.json()
+                self._cache = certs
+                self._cache_at = time.monotonic()
+                logger.info(f"Google Chat cert cache refreshed: {len(certs)} certificates")
+                return dict(certs)
+        except Exception as e:
+            logger.warning(f"Failed to fetch Google Chat certs: {e}")
+            if self._cache is not None:
+                logger.info("Using stale cert cache")
+                return dict(self._cache)
+            return {}
+
+    def clear(self) -> None:
+        self._cache = None
+        self._cache_at = 0.0
+
+
+def verify_project_number_token(token: str, project_number: str, certs: dict[str, str]) -> bool:
+    try:
+        from google.auth import jwt
+    except ImportError:
+        logger.warning("google-auth not installed, skipping project-number token verification")
+        return True
+
+    if not certs:
+        logger.warning("No certificates available for project-number verification")
+        return False
+
+    certs_json = json.dumps(certs)
+    try:
+        payload = jwt.decode(token, certs=certs_json, audience=project_number)
+        issuer = payload.get("iss", "")
+        expected_issuer = "chat@system.gserviceaccount.com"
+        if issuer != expected_issuer:
+            logger.warning(f"project-number token issuer mismatch: expected '{expected_issuer}', got '{issuer}'")
+            return False
+        return True
+    except Exception as e:
+        logger.warning(f"project-number token verification failed: {e}")
+        return False
+
+
+_global_cert_cache: GoogleChatCertCache | None = None
+
+
+def get_cert_cache() -> GoogleChatCertCache:
+    global _global_cert_cache
+    if _global_cert_cache is None:
+        _global_cert_cache = GoogleChatCertCache()
+    return _global_cert_cache

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

@@ -0,0 +1,218 @@
+from __future__ import annotations
+
+
+def build_approval_card(title: str, action_id: str, confirm_text: str = "确认", reject_text: str = "拒绝") -> dict:
+    return {
+        "cards_v2": [
+            {
+                "card_id": action_id,
+                "card": {
+                    "header": {"title": title, "imageType": "SQUARE"},
+                    "sections": [
+                        {
+                            "widgets": [
+                                {
+                                    "buttonList": {
+                                        "buttons": [
+                                            _make_button(confirm_text, "approve", green=True),
+                                            _make_button(reject_text, "reject", green=False),
+                                        ]
+                                    }
+                                }
+                            ]
+                        }
+                    ],
+                },
+            }
+        ]
+    }
+
+
+def build_poll_card(question: str, options: list[str], action_id: str) -> dict:
+    buttons = [_make_button(opt, f"poll_vote_{i}_{action_id}") for i, opt in enumerate(options)]
+
+    sections: list[dict] = []
+    while buttons:
+        chunk = buttons[:2]
+        buttons = buttons[2:]
+        sections.append({"widgets": [{"buttonList": {"buttons": chunk}}]})
+
+    return {
+        "cards_v2": [
+            {
+                "card_id": action_id,
+                "card": {
+                    "header": {"title": question, "imageType": "SQUARE"},
+                    "sections": sections,
+                },
+            }
+        ]
+    }
+
+
+def build_info_card(title: str, fields: dict[str, str]) -> dict:
+    widgets = []
+    for key, value in fields.items():
+        widgets.append(
+            {
+                "decoratedText": {
+                    "topLabel": key,
+                    "text": value,
+                }
+            }
+        )
+
+    return {
+        "cards_v2": [
+            {
+                "card": {
+                    "header": {"title": title},
+                    "sections": [{"widgets": widgets}],
+                }
+            }
+        ]
+    }
+
+
+def build_form_card(title: str, fields: list[dict], submit_action: str) -> dict:
+    widgets = []
+    for field in fields:
+        field_input_type = field.get("input_type", "SINGLE_LINE")
+
+        if field_input_type in ("SINGLE_SELECT", "MULTI_SELECT"):
+            widgets.append(
+                {
+                    "selectionInput": {
+                        "label": field.get("label", ""),
+                        "name": field.get("name", ""),
+                        "type": field_input_type,
+                        "items": _build_selection_items(field.get("options", [])),
+                    }
+                }
+            )
+        elif field_input_type == "DIVIDER":
+            widgets.append({"divider": {}})
+        else:
+            widgets.append(
+                {
+                    "textInput": {
+                        "label": field.get("label", ""),
+                        "name": field.get("name", ""),
+                        "type": field_input_type,
+                    }
+                }
+            )
+
+    widgets.append({"buttonList": {"buttons": [_make_button("提交", submit_action, green=True)]}})
+
+    return {
+        "cards_v2": [
+            {
+                "card": {
+                    "header": {"title": title},
+                    "sections": [{"widgets": widgets}],
+                }
+            }
+        ]
+    }
+
+
+def build_exec_approval_card(
+    title: str,
+    action_id: str,
+    detail_fields: dict[str, str] | None = None,
+) -> dict:
+    widgets = []
+
+    if detail_fields:
+        for key, value in detail_fields.items():
+            widgets.append(
+                {
+                    "decoratedText": {
+                        "topLabel": key,
+                        "text": value,
+                    }
+                }
+            )
+
+    widgets.append(
+        {
+            "buttonList": {
+                "buttons": [
+                    _make_button("确认执行", f"exec_approve_{action_id}", green=True),
+                    _make_button("拒绝执行", f"exec_reject_{action_id}", green=False),
+                ]
+            }
+        }
+    )
+
+    return {
+        "cards_v2": [
+            {
+                "card": {
+                    "header": {"title": title, "imageType": "SQUARE"},
+                    "sections": [{"widgets": widgets}],
+                }
+            }
+        ]
+    }
+
+
+def build_selection_card(
+    title: str,
+    selection_name: str,
+    label: str,
+    options: list[dict[str, str]],
+    selection_type: str = "SINGLE_SELECT",
+    submit_action: str | None = None,
+) -> dict:
+    widgets = [
+        {
+            "selectionInput": {
+                "label": label,
+                "name": selection_name,
+                "type": selection_type,
+                "items": _build_selection_items(options),
+            }
+        }
+    ]
+
+    if submit_action:
+        widgets.append({"buttonList": {"buttons": [_make_button("提交", submit_action, green=True)]}})
+
+    return {
+        "cards_v2": [
+            {
+                "card": {
+                    "header": {"title": title},
+                    "sections": [{"widgets": widgets}],
+                }
+            }
+        ]
+    }
+
+
+def _make_divider() -> dict:
+    return {"divider": {}}
+
+
+def _build_selection_items(options: list[dict[str, str]]) -> list[dict]:
+    items = []
+    for opt in options:
+        items.append(
+            {
+                "text": opt.get("text", opt.get("label", "")),
+                "value": opt.get("value", ""),
+                "selected": opt.get("selected", False),
+            }
+        )
+    return items
+
+
+def _make_button(text: str, action: str, green: bool = False) -> dict:
+    color = {"red": 0, "green": 0.6, "blue": 0} if green else {"red": 0.6, "green": 0, "blue": 0}
+    return {
+        "text": text,
+        "color": color,
+        "onClick": {"action": {"function": action}},
+    }

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

@@ -0,0 +1,45 @@
+from __future__ import annotations
+
+from typing import Any
+
+
+def list_peers(allow_from: list[str]) -> list[dict[str, Any]]:
+    peers: list[dict[str, Any]] = []
+    for entry in allow_from:
+        if entry == "*":
+            continue
+        normalized = normalize_id(entry)
+        peers.append({"id": normalized, "raw": entry, "type": "user"})
+    return peers
+
+
+def list_groups(groups_config: dict[str, Any]) -> list[dict[str, Any]]:
+    groups: list[dict[str, Any]] = []
+    for space_name, cfg in groups_config.items():
+        if not isinstance(cfg, dict):
+            continue
+        groups.append(
+            {
+                "id": space_name,
+                "type": "space",
+                "name": cfg.get("name", space_name),
+                "enabled": cfg.get("enabled", True),
+                "requires_mention": cfg.get("require_mention", cfg.get("requireMention", False)),
+            }
+        )
+    return groups
+
+
+def normalize_id(raw: str) -> str:
+    if not raw:
+        return ""
+    raw = raw.strip()
+    if raw.startswith("spaces/"):
+        return raw
+    if raw.startswith("users/"):
+        return raw
+    if raw.startswith("user:"):
+        return raw.replace("user:", "users/", 1)
+    if "@" in raw:
+        return f"users/{raw}"
+    return f"users/{raw}"

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

@@ -0,0 +1,164 @@
+from __future__ import annotations
+
+import re
+from typing import Any
+
+
+def run_diagnostics(config: dict[str, Any] | None) -> list[dict[str, Any]]:
+    if not config:
+        return []
+
+    issues: list[dict[str, Any]] = []
+    issues.extend(_check_mutable_allowlist(config))
+    issues.extend(_check_deprecated_formats(config))
+    issues.extend(_check_legacy_stream_mode(config))
+    issues.extend(_check_credential_readiness(config))
+    return issues
+
+
+def _check_mutable_allowlist(config: dict[str, Any]) -> list[dict[str, Any]]:
+    issues: list[dict[str, Any]] = []
+
+    allow_from = _normalize_allowlist(config.get("allowFrom", config.get("allow_from", [])))
+    for entry in allow_from:
+        if "@" in entry and not _is_users_prefix(entry):
+            issues.append(
+                {
+                    "severity": "warning",
+                    "category": "mutable_allowlist",
+                    "field": "allowFrom",
+                    "value": entry,
+                    "message": (
+                        f"allowFrom contains email '{entry}' without 'users/' prefix. "
+                        "Email-based allowlists are mutable; consider migrating to 'users/<email>' format"
+                    ),
+                }
+            )
+
+    group_allow_from = _normalize_allowlist(config.get("groupAllowFrom", config.get("group_allow_from", [])))
+    for entry in group_allow_from:
+        if "@" in entry and not entry.startswith("spaces/"):
+            issues.append(
+                {
+                    "severity": "warning",
+                    "category": "mutable_allowlist",
+                    "field": "groupAllowFrom",
+                    "value": entry,
+                    "message": (
+                        f"groupAllowFrom contains email '{entry}'. "
+                        "Group allowlists should use 'spaces/<id>' format as emails are mutable"
+                    ),
+                }
+            )
+
+    return issues
+
+
+def _check_deprecated_formats(config: dict[str, Any]) -> list[dict[str, Any]]:
+    issues: list[dict[str, Any]] = []
+
+    allow_from = _normalize_allowlist(config.get("allowFrom", config.get("allow_from", [])))
+    for entry in allow_from:
+        if _is_users_prefix_with_email_legacy(entry):
+            issues.append(
+                {
+                    "severity": "warning",
+                    "category": "deprecated_format",
+                    "field": "allowFrom",
+                    "value": entry,
+                    "message": (
+                        f"allowFrom entry '{entry}' uses deprecated format. "
+                        "Remove the trailing '/<email>' portion: use 'users/<userId>' instead"
+                    ),
+                }
+            )
+
+    if config.get("streamMode"):
+        issues.append(
+            {
+                "severity": "info",
+                "category": "deprecated_config",
+                "field": "streamMode",
+                "value": config["streamMode"],
+                "message": (
+                    "streamMode config is deprecated; "
+                    "streaming is configured via 'supports_streaming' capability"
+                ),
+            }
+        )
+
+    if config.get("enableThreadReply"):
+        issues.append(
+            {
+                "severity": "info",
+                "category": "deprecated_config",
+                "field": "enableThreadReply",
+                "value": config["enableThreadReply"],
+                "message": "enableThreadReply is deprecated; use replyToMode instead",
+            }
+        )
+
+    return issues
+
+
+def _check_legacy_stream_mode(config: dict[str, Any]) -> list[dict[str, Any]]:
+    issues: list[dict[str, Any]] = []
+    if config.get("streamMode"):
+        issues.append(
+            {
+                "severity": "info",
+                "category": "legacy_config",
+                "field": "streamMode",
+                "value": config["streamMode"],
+                "message": "Remove 'streamMode' config key; streaming is now controlled by capability flags",
+            }
+        )
+    return issues
+
+
+def _check_credential_readiness(config: dict[str, Any]) -> list[dict[str, Any]]:
+    issues: list[dict[str, Any]] = []
+
+    has_inline = bool(config.get("service_account"))
+    has_file = bool(config.get("service_account_file"))
+    has_secret_ref = bool(config.get("serviceAccountRef", config.get("service_account_ref")))
+    has_env = bool(
+        __import__("os").getenv("GOOGLE_CHAT_SERVICE_ACCOUNT")
+        or __import__("os").getenv("GOOGLE_SERVICE_ACCOUNT_FILE")
+        or __import__("os").getenv("GOOGLE_CHAT_SERVICE_ACCOUNT_FILE")
+    )
+
+    if not (has_inline or has_file or has_secret_ref or has_env):
+        issues.append(
+            {
+                "severity": "error",
+                "category": "credential_readiness",
+                "field": "serviceAccount",
+                "message": (
+                    "No Google Chat service account configured. "
+                    "Provide one via service_account, service_account_file, "
+                    "serviceAccountRef, or environment variables"
+                ),
+            }
+        )
+
+    return issues
+
+
+def _normalize_allowlist(raw: Any) -> list[str]:
+    if isinstance(raw, str):
+        return [x.strip() for x in raw.split(",") if x.strip()]
+    if isinstance(raw, (list, tuple)):
+        return [str(x).strip() for x in raw if x]
+    return []
+
+
+def _is_users_prefix(entry: str) -> bool:
+    return entry.startswith("users/") or entry.startswith("user:")
+
+
+def _is_users_prefix_with_email_legacy(entry: str) -> bool:
+    if not entry.startswith("users/"):
+        return False
+    inner = entry.removeprefix("users/")
+    return bool(re.match(r"^[\w.+-]+@[\w-]+\.[\w.-]+/", inner))

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

@@ -0,0 +1,207 @@
+from __future__ import annotations
+
+import re
+from typing import Any
+
+from yuxi.channels.models import ChannelResponse
+
+from . import cards
+
+_MAX_TEXT_CHARS = 4000
+
+
+def format_outbound(response: ChannelResponse) -> dict[str, Any]:
+    body: dict[str, Any] = {}
+    content = sanitize_text(response.content or "")
+    if not content:
+        content = " "
+    body["text"] = _truncate_text(content)
+
+    body = _apply_card(body, response)
+
+    if response.attachments:
+        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}}]}]}})
+
+    if response.metadata and response.metadata.get("thread_key"):
+        body["messageReplyOption"] = "REPLY_MESSAGE_FALLBACK_TO_NEW_THREAD"
+        body["thread"] = {"threadKey": response.metadata["thread_key"]}
+
+    if response.metadata and response.metadata.get("silent"):
+        body["disableNotification"] = True
+
+    return body
+
+
+def _apply_card(body: dict[str, Any], response: ChannelResponse) -> dict[str, Any]:
+    if response.metadata and response.metadata.get("card"):
+        body["cards_v2"] = response.metadata["card"].get("cards_v2", [])
+        return body
+
+    card_type = (response.metadata or {}).get("card_type", "")
+    if not card_type:
+        return body
+
+    card_params = (response.metadata or {}).get("card_params", {})
+    card_body = _build_card(card_type, response.content, card_params)
+    if card_body:
+        body["cards_v2"] = card_body.get("cards_v2", [])
+
+    return body
+
+
+def _build_card(card_type: str, content: str, params: dict[str, Any]) -> dict | None:
+    action_id = params.get("action_id", "card_action")
+
+    if card_type == "approval":
+        return cards.build_approval_card(
+            title=params.get("title", content or "审批"),
+            action_id=action_id,
+            confirm_text=params.get("confirm_text", "确认"),
+            reject_text=params.get("reject_text", "拒绝"),
+        )
+
+    if card_type == "poll":
+        options = params.get("options", [])
+        if not options:
+            return None
+        return cards.build_poll_card(
+            question=params.get("title", content or "投票"),
+            options=options,
+            action_id=action_id,
+        )
+
+    if card_type == "info":
+        fields = params.get("fields", {})
+        if not fields:
+            return None
+        return cards.build_info_card(
+            title=params.get("title", content or "信息"),
+            fields=fields,
+        )
+
+    if card_type == "form":
+        fields = params.get("fields", [])
+        if not fields:
+            return None
+        return cards.build_form_card(
+            title=params.get("title", content or "表单"),
+            fields=fields,
+            submit_action=params.get("submit_action", action_id),
+        )
+
+    if card_type == "selection":
+        options = params.get("options", [])
+        if not options:
+            return None
+        return cards.build_selection_card(
+            title=params.get("title", content or "选择"),
+            selection_name=params.get("selection_name", "selection"),
+            label=params.get("label", "请选择"),
+            options=options,
+            selection_type=params.get("selection_type", "SINGLE_SELECT"),
+            submit_action=params.get("submit_action"),
+        )
+
+    if card_type == "exec_approval":
+        return cards.build_exec_approval_card(
+            title=params.get("title", content or "执行审批"),
+            action_id=action_id,
+            detail_fields=params.get("detail_fields"),
+        )
+
+    return None
+
+
+def resolve_reply_to_mode(config: dict | None = None, default: str = "off") -> str:
+    if not config:
+        return default
+    mode = str(config.get("reply_to_mode", config.get("replyToMode", default))).strip().lower()
+    if mode in ("first", "all", "off"):
+        return mode
+    return default
+
+
+def sanitize_text(text: str) -> str:
+    text = text.replace("\x00", "").rstrip()
+    text = re.sub(r"[\x00-\x08\x0b\x0c\x0e-\x1f]", "", text)
+    return text
+
+
+def chunk_text_for_outbound(text: str, chunk_limit: int = _MAX_TEXT_CHARS) -> list[str]:
+    if not text:
+        return []
+    if len(text) <= chunk_limit:
+        return [text]
+
+    chunks: list[str] = []
+    remaining = text
+
+    while remaining:
+        if len(remaining) <= chunk_limit:
+            chunks.append(remaining)
+            break
+
+        split_at = _find_split_point(remaining, chunk_limit)
+        chunk = remaining[:split_at].rstrip()
+        chunks.append(chunk)
+        remaining = remaining[split_at:].lstrip()
+
+    return chunks
+
+
+def _find_split_point(text: str, limit: int) -> int:
+    window = text[:limit]
+
+    code_block_match = re.search(r"```[\s\S]*?```", window)
+    if code_block_match and code_block_match.end() <= limit:
+        after_end = code_block_match.end()
+        if after_end < limit:
+            return after_end
+
+    open_fence = re.search(r"```\w*\n", window)
+    if open_fence:
+        fence_start = open_fence.start()
+        closing = text.find("```", fence_start + 3)
+        if fence_start > 0 and (closing == -1 or closing >= limit):
+            return fence_start
+
+    for pat in [r"\n\n", r"\n", r"\. ", r"。", r"\.\n"]:
+        matches = list(re.finditer(pat, window))
+        if matches:
+            last = matches[-1]
+            boundary = last.end()
+            if boundary > limit * 0.6:
+                return boundary
+
+    for pat_word in [r"\s", r"[,，;；:：!！?？]"]:
+        m = re.search(pat_word + r"[^\s]*$", window)
+        if m:
+            return m.start()
+
+    return limit
+
+
+def _truncate_text(text: str) -> str:
+    if len(text) <= _MAX_TEXT_CHARS:
+        return text
+
+    truncated = text[: _MAX_TEXT_CHARS - 3]
+
+    markdown_boundary = _find_markdown_break(truncated)
+    if markdown_boundary > _MAX_TEXT_CHARS * 0.8:
+        truncated = truncated[:markdown_boundary].rstrip()
+
+    return truncated + "..."
+
+
+def _find_markdown_break(text: str) -> int:
+    best = 0
+    for pat in [r"\n\n", r"\n", r"\. ", r"。"]:
+        matches = list(re.finditer(pat, text))
+        if matches:
+            best = matches[-1].end()
+            break
+    return best if best > 0 else len(text)

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

@@ -0,0 +1,153 @@
+from __future__ import annotations
+
+import io
+import json
+from typing import TYPE_CHECKING
+
+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)
+    except (json.JSONDecodeError, TypeError):
+        raise DeliveryFailedError(f"Invalid file_id format: {file_id}")
+
+    resource_name = file_info.get("name", file_info.get("resource_name", ""))
+    if not resource_name:
+        raise DeliveryFailedError("file_id missing attachment resource name")
+
+    try:
+        attachment = chat_service.spaces().messages().attachments().get(name=resource_name).execute()
+    except Exception as e:
+        raise DeliveryFailedError(f"Failed to get attachment metadata: {e}")
+
+    drive_data = attachment.get("driveDataRef", {})
+    if drive_data and drive_data.get("driveFileId"):
+        try:
+            return await _download_drive_file(chat_service, drive_data["driveFileId"])
+        except Exception as e:
+            raise DeliveryFailedError(f"Drive file download failed (file_id={drive_data.get('driveFileId')}): {e}")
+
+    try:
+        creds = _get_credentials(chat_service)
+        token = await _get_access_token(creds)
+
+        import httpx
+
+        url = f"https://chat.googleapis.com/v1/{resource_name}?alt=media"
+        async with httpx.AsyncClient(timeout=httpx.Timeout(60)) as client:
+            resp = await client.get(
+                url,
+                headers={"Authorization": f"Bearer {token}"},
+                follow_redirects=True,
+            )
+            if resp.status_code == 200:
+                content_length = int(resp.headers.get("content-length", 0))
+                if content_length > _MAX_DOWNLOAD_SIZE_BYTES:
+                    raise DeliveryFailedError(
+                        f"File too large: {content_length} bytes (max {_MAX_DOWNLOAD_SIZE_BYTES})"
+                    )
+                return resp.content
+            raise DeliveryFailedError(f"Media download returned HTTP {resp.status_code}")
+    except DeliveryFailedError:
+        raise
+    except Exception as e:
+        logger.warning(f"Direct media download failed, falling back: {e}")
+
+    raise DeliveryFailedError(f"Unable to download media for attachment: {resource_name}")
+
+
+async def _download_drive_file(chat_service: Resource, drive_file_id: str) -> bytes:
+    from googleapiclient.discovery import build
+    from googleapiclient.http import MediaIoBaseDownload
+
+    creds = _get_credentials(chat_service)
+    drive_service = build("drive", "v3", credentials=creds, cache_discovery=False)
+
+    request = drive_service.files().get_media(fileId=drive_file_id)
+    buf = io.BytesIO()
+    downloader = MediaIoBaseDownload(buf, request)
+    done = False
+    while not done:
+        _, done = downloader.next_chunk()
+    return buf.getvalue()
+
+
+def _get_credentials(chat_service: Resource):
+    http = getattr(chat_service, "_http", None)
+    if http is None:
+        raise DeliveryFailedError("No HTTP transport on chat_service")
+    creds = getattr(http, "credentials", None)
+    if creds is None:
+        raise DeliveryFailedError("No credentials on chat_service HTTP transport")
+    return creds
+
+
+async def _get_access_token(credentials) -> str:
+    from google.auth.transport.requests import Request as GARequest
+
+    if credentials.valid:
+        return credentials.token
+    credentials.refresh(GARequest())
+    return credentials.token

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

@@ -0,0 +1,25 @@
+from __future__ import annotations
+
+from yuxi.channels.models import MentionsInfo
+
+
+def parse_mentions(message: dict, bot_user: str = "") -> MentionsInfo:
+    annotations = message.get("annotations", [])
+    mentioned_ids = [
+        a.get("userMention", {}).get("user", {}).get("name", "") for a in annotations if a.get("type") == "USER_MENTION"
+    ]
+    is_at_bot = False
+    for a in annotations:
+        if a.get("type") != "USER_MENTION":
+            continue
+        um = a.get("userMention", {})
+        if um.get("type") == "BOT":
+            is_at_bot = True
+            break
+        if bot_user and um.get("user", {}).get("name", "") == bot_user:
+            is_at_bot = True
+            break
+    return MentionsInfo(
+        mentioned_user_ids=mentioned_ids,
+        is_bot_mentioned=is_at_bot,
+    )

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

@@ -0,0 +1,81 @@
+from __future__ import annotations
+
+import time
+from collections import OrderedDict
+from typing import Any
+
+from yuxi.utils.logging_config import logger
+
+_MAX_CACHE_SIZE = 128
+_MAX_AGE_S = 3600
+
+
+class SentMessageCache:
+    def __init__(self, max_size: int = _MAX_CACHE_SIZE, max_age_s: int = _MAX_AGE_S):
+        self._cache: OrderedDict[str, tuple[float, dict[str, Any]]] = OrderedDict()
+        self._max_size = max_size
+        self._max_age_s = max_age_s
+
+    def put(self, msg_id: str, metadata: dict[str, Any] | None = None) -> None:
+        now = time.monotonic()
+        if len(self._cache) >= self._max_size:
+            self._cache.popitem(last=False)
+        self._cache[msg_id] = (now, metadata or {})
+
+    def get(self, msg_id: str) -> dict[str, Any] | None:
+        entry = self._cache.get(msg_id)
+        if entry is None:
+            return None
+        ts, meta = entry
+        if time.monotonic() - ts > self._max_age_s:
+            self._cache.pop(msg_id, None)
+            return None
+        return meta
+
+    def remove(self, msg_id: str) -> None:
+        self._cache.pop(msg_id, None)
+
+    def update(self, msg_id: str, metadata: dict[str, Any]) -> None:
+        entry = self._cache.get(msg_id)
+        if entry is None:
+            return
+        ts, existing = entry
+        existing.update(metadata)
+        self._cache[msg_id] = (ts, existing)
+
+    def size(self) -> int:
+        return len(self._cache)
+
+    def clear(self) -> None:
+        self._cache.clear()
+
+    def expired_count(self) -> int:
+        now = time.monotonic()
+        expired = 0
+        for _msg_id, (ts, _) in self._cache.items():
+            if now - ts > self._max_age_s:
+                expired += 1
+        return expired
+
+
+def track_sent_message(
+    cache: SentMessageCache,
+    msg_id: str,
+    chat_id: str = "",
+    message_type: str = "text",
+) -> None:
+    cache.put(msg_id, {"chat_id": chat_id, "message_type": message_type, "sent_at": time.monotonic()})
+    logger.debug(f"Sent message cached: {msg_id}")
+
+
+def update_sent_message_status(
+    cache: SentMessageCache,
+    msg_id: str,
+    status: str,
+    metadata: dict[str, Any] | None = None,
+) -> None:
+    update_data = {"status": status}
+    if metadata:
+        update_data.update(metadata)
+    cache.update(msg_id, update_data)
+    logger.debug(f"Sent message status updated: {msg_id} -> {status}")

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

@@ -0,0 +1,249 @@
+from __future__ import annotations
+
+import json
+
+from yuxi.channels.models import (
+    Attachment,
+    ChannelIdentity,
+    ChannelMessage,
+    ChannelType,
+    ChatType,
+    EventType,
+    MessageType,
+)
+
+from .mentions import parse_mentions
+from .slash_commands import extract_command
+
+
+def map_event_type(event_str: str) -> EventType:
+    _map = {
+        "MESSAGE": EventType.MESSAGE_RECEIVED,
+        "MESSAGE_UPDATE": EventType.MESSAGE_UPDATED,
+        "MESSAGE_DELETE": EventType.MESSAGE_DELETED,
+        "ADDED_TO_SPACE": EventType.BOT_ADDED,
+        "REMOVED_FROM_SPACE": EventType.BOT_REMOVED,
+        "CARD_CLICKED": EventType.CARD_ACTION,
+    }
+    return _map.get(event_str, EventType.MESSAGE_RECEIVED)
+
+
+def normalize_inbound(
+    channel_id: str,
+    channel_type: ChannelType,
+    raw_payload: dict,
+    bot_user: str = "",
+) -> ChannelMessage:
+    event = raw_payload.get("event", {})
+    event_type_str = event.get("type", "")
+    message = event.get("message", {})
+    space = event.get("space", {})
+    user = event.get("user", {})
+    sender = message.get("sender", {})
+
+    event_type = map_event_type(event_type_str)
+
+    space_name = space.get("name", "")
+    space_type = space.get("spaceType", "SPACE")
+    chat_type = _resolve_chat_type(space_type, message, raw_payload, event_type)
+
+    thread_key = message.get("thread", {}).get("threadKey", "")
+    if thread_key:
+        chat_type = ChatType.THREAD
+        chat_id = f"{space_name}/threads/{thread_key}"
+    else:
+        chat_id = space_name
+
+    mentions = parse_mentions(message, bot_user)
+
+    content = _extract_content(message, event_type, raw_payload)
+    msg_type = map_msg_type(message, event_type)
+    attachments = extract_attachments(message)
+    argument_text = _extract_argument_text(message)
+    context = _build_context(channel_id, event, message, space, user, sender)
+    route_envelope = resolve_route_envelope(chat_type, space_name, user.get("name", ""))
+
+    command, command_args = extract_command(content)
+
+    return ChannelMessage(
+        identity=ChannelIdentity(
+            channel_id=channel_id,
+            channel_type=channel_type,
+            channel_user_id=user.get("name", ""),
+            channel_chat_id=chat_id,
+            channel_message_id=message.get("name", ""),
+        ),
+        chat_type=chat_type,
+        event_type=event_type,
+        message_type=MessageType.COMMAND if command else msg_type,
+        content=content,
+        attachments=attachments,
+        mentions=mentions,
+        metadata={
+            "space_name": space_name,
+            "space_type": space_type,
+            "thread_key": thread_key,
+            "sender_name": sender.get("name", ""),
+            "sender_type": sender.get("type", ""),
+            "argument_text": argument_text,
+            "context": context,
+            "route_envelope": route_envelope,
+            "slash_command": command,
+            "slash_args": command_args,
+        },
+    )
+
+
+def is_bot_message(message: dict) -> bool:
+    sender = message.get("sender", {})
+    return sender.get("type", "") == "BOT"
+
+
+def _extract_argument_text(message: dict) -> str:
+    annotations = message.get("annotations", [])
+    for annotation in annotations:
+        if annotation.get("type") == "SLASH_COMMAND":
+            return message.get("argumentText", "")
+    return ""
+
+
+def _build_context(
+    channel_id: str,
+    event: dict,
+    message: dict,
+    space: dict,
+    user: dict,
+    sender: dict,
+) -> dict:
+    return {
+        "channel": channel_id,
+        "accountId": event.get("accountId", ""),
+        "messageId": message.get("name", ""),
+        "from": sender.get("name", ""),
+        "sender": sender,
+        "conversation": space,
+        "reply": message.get("thread", {}),
+    }
+
+
+def _resolve_chat_type(
+    space_type: str,
+    message: dict,
+    raw_payload: dict,
+    event_type: EventType,
+) -> ChatType:
+    if space_type == "DIRECT_MESSAGE":
+        return ChatType.DIRECT
+    if raw_payload.get("is_card_clicked"):
+        return ChatType.SPACE
+    if event_type == EventType.CARD_ACTION:
+        return ChatType.SPACE
+    return ChatType.SPACE
+
+
+def is_forwarded_message(message: dict) -> bool:
+    return bool(message.get("retentionSettings")) or bool(message.get("lastUpdateTime") and message.get("createTime"))
+
+
+def detect_message_edit(message: dict) -> bool:
+    create_time = message.get("createTime", "")
+    update_time = message.get("lastUpdateTime", "")
+    if create_time and update_time and create_time != update_time:
+        return True
+    return False
+
+
+def resolve_route_envelope(chat_type: ChatType, space_name: str, user_name: str) -> dict:
+    if chat_type == ChatType.DIRECT:
+        user_id = user_name.removeprefix("users/") if user_name.startswith("users/") else user_name
+        return {
+            "peer_kind": "direct",
+            "peer_id": user_name,
+            "peer_key": f"direct:{user_id}",
+            "route": {
+                "session_key": f"direct:{user_id}",
+                "content": {"type": "direct", "id": user_name},
+            },
+        }
+    space_id = space_name.removeprefix("spaces/") if space_name.startswith("spaces/") else space_name
+    return {
+        "peer_kind": "group",
+        "peer_id": space_name,
+        "peer_key": f"space:{space_id}",
+        "route": {
+            "session_key": f"space:{space_id}",
+            "content": {"type": "space", "id": space_name},
+        },
+    }
+
+
+def map_msg_type(message: dict, event_type: EventType) -> MessageType:
+    if event_type == EventType.CARD_ACTION:
+        return MessageType.CARD
+
+    attachments = message.get("attachment", message.get("attachments", []))
+    if attachments:
+        first = attachments[0] if attachments else {}
+        content_type = first.get("contentType", "")
+        if content_type.startswith("image/"):
+            return MessageType.IMAGE
+        if content_type.startswith("video/"):
+            return MessageType.VIDEO
+        if content_type.startswith("audio/"):
+            return MessageType.AUDIO
+        if content_type:
+            return MessageType.FILE
+
+    text = message.get("text", "")
+    if message.get("cardsV2") or message.get("cards_v2"):
+        return MessageType.CARD
+    if text:
+        return MessageType.TEXT
+
+    return MessageType.TEXT
+
+
+def extract_attachments(message: dict) -> list[Attachment]:
+    attachments = message.get("attachment", message.get("attachments", []))
+    if not attachments:
+        return []
+
+    result: list[Attachment] = []
+    for att in attachments:
+        name = att.get("name", "")
+        content_name = att.get("contentName", "")
+        content_type = att.get("contentType", "")
+
+        att_type = "file"
+        if content_type.startswith("image/"):
+            att_type = "image"
+        elif content_type.startswith("video/"):
+            att_type = "video"
+        elif content_type.startswith("audio/"):
+            att_type = "audio"
+
+        result.append(
+            Attachment(
+                type=att_type,
+                filename=content_name,
+                mime_type=content_type,
+                file_id=_encode_file_id(name),
+                metadata={"attachment_name": name},
+            )
+        )
+    return result
+
+
+def _encode_file_id(attachment_name: str) -> str:
+    return json.dumps({"name": attachment_name})
+
+
+def _extract_content(
+    message: dict,
+    event_type: EventType,
+    raw_payload: dict,
+) -> str:
+    if event_type == EventType.CARD_ACTION:
+        action = raw_payload.get("action", {})
+        return json.dumps(action)
+    return message.get("text", "")

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

@@ -0,0 +1,49 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from googleapiclient.discovery import Resource
+
+from yuxi.channels.models import DeliveryResult
+from yuxi.utils.logging_config import logger
+
+_PAIRING_CHALLENGE_MESSAGE = (
+    "{agent_name} Approval Required\n\n"
+    "To continue, send the following:\n"
+    "```\n/approve {user_id}\n```\n"
+    "This confirmation is required to pair with this bot account."
+)
+
+
+async def send_pairing_challenge(
+    chat_service: Resource,
+    space_name: str,
+    user_id: str,
+    agent_name: str = "Bot",
+) -> DeliveryResult:
+    content = _PAIRING_CHALLENGE_MESSAGE.format(
+        agent_name=agent_name,
+        user_id=user_id,
+    )
+    body = {"text": content}
+    try:
+        result = chat_service.spaces().messages().create(parent=space_name, body=body).execute()
+        msg_id = result.get("name", "")
+        logger.info(f"Pairing challenge sent to {space_name}: user_id={user_id}")
+        return DeliveryResult(success=True, message_id=msg_id)
+    except Exception as e:
+        logger.warning(f"Failed to send pairing challenge to {space_name}: {e}")
+        return DeliveryResult(success=False, error=str(e))
+
+
+async def check_pairing_approval(
+    chat_service: Resource,
+    space_name: str,
+    user_id: str,
+    message_text: str,
+) -> bool:
+    approved = f"/approve {user_id}" in message_text.strip().lower()
+    if approved:
+        logger.info(f"Pairing approved for {space_name}: user_id={user_id}")
+    return approved

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

@@ -0,0 +1,185 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from enum import StrEnum
+from typing import Any
+
+from yuxi.channels.models import ChatType, ChannelMessage
+from yuxi.utils.logging_config import logger
+
+
+class DmPolicy(StrEnum):
+    OPEN = "open"
+    PAIRING = "pairing"
+    ALLOWLIST = "allowlist"
+    DISABLED = "disabled"
+
+
+class GroupPolicy(StrEnum):
+    OPEN = "open"
+    ALLOWLIST = "allowlist"
+    DISABLED = "disabled"
+
+
+DM_POLICY_VALUES = frozenset(p.value for p in DmPolicy)
+GROUP_POLICY_VALUES = frozenset(p.value for p in GroupPolicy)
+
+
+@dataclass
+class PerGroupConfig:
+    require_mention: bool = False
+    enabled: bool = True
+    users: list[str] = field(default_factory=list)
+    system_prompt: str = ""
+
+
+@dataclass
+class GoogleChatPolicy:
+    dm_policy: DmPolicy = DmPolicy.OPEN
+    group_policy: GroupPolicy = GroupPolicy.OPEN
+    allow_from: list[str] = field(default_factory=list)
+    group_allow_from: list[str] = field(default_factory=list)
+    require_mention: bool = False
+    allow_bots: bool = False
+    bot_user: str = ""
+    groups: dict[str, PerGroupConfig] = field(default_factory=dict)
+    dangerously_allow_name_matching: bool = False
+
+    @classmethod
+    def from_config(cls, config: dict[str, Any] | None) -> GoogleChatPolicy:
+        if not config:
+            return cls()
+
+        dm_raw = str(config.get("dm_policy", config.get("dmPolicy", "open"))).strip().lower()
+        dm_policy = DmPolicy(dm_raw) if dm_raw in DM_POLICY_VALUES else DmPolicy.OPEN
+
+        group_raw = str(config.get("group_policy", config.get("groupPolicy", "open"))).strip().lower()
+        group_policy = GroupPolicy(group_raw) if group_raw in GROUP_POLICY_VALUES else GroupPolicy.OPEN
+
+        allow_from = _normalize_list(config.get("allowFrom", config.get("allow_from", [])))
+        group_allow_from = _normalize_list(config.get("groupAllowFrom", config.get("group_allow_from", [])))
+        require_mention = bool(config.get("require_mention", config.get("requireMention", False)))
+        allow_bots = bool(config.get("allow_bots", config.get("allowBots", False)))
+        bot_user = str(config.get("bot_user", config.get("botUser", ""))).strip()
+        groups = _parse_per_group_config(config.get("groups", {}))
+        dangerously_allow_name_matching = bool(
+            config.get("dangerously_allow_name_matching", config.get("dangerouslyAllowNameMatching", False))
+        )
+
+        return cls(
+            dm_policy=dm_policy,
+            group_policy=group_policy,
+            allow_from=allow_from,
+            group_allow_from=group_allow_from,
+            require_mention=require_mention,
+            allow_bots=allow_bots,
+            bot_user=bot_user,
+            groups=groups,
+            dangerously_allow_name_matching=dangerously_allow_name_matching,
+        )
+
+    def check_dm_access(self, user_id: str) -> bool:
+        if self.dm_policy == DmPolicy.DISABLED:
+            return False
+        if self.dm_policy == DmPolicy.OPEN:
+            return True
+        if self.dm_policy == DmPolicy.PAIRING:
+            return True
+        if self.dm_policy == DmPolicy.ALLOWLIST:
+            return self._match_allowlist(user_id, self.allow_from)
+        return False
+
+    def check_group_access(self, space_id: str) -> bool:
+        per_group = self._resolve_per_group(space_id)
+        if per_group is not None and not per_group.enabled:
+            return False
+        if self.group_policy == GroupPolicy.DISABLED:
+            return False
+        if self.group_policy == GroupPolicy.OPEN:
+            return True
+        if self.group_policy == GroupPolicy.ALLOWLIST:
+            return self._match_allowlist(space_id, self.group_allow_from)
+        return False
+
+    def check_inbound(self, channel_msg: ChannelMessage) -> bool:
+        chat_type = channel_msg.chat_type
+
+        if chat_type == ChatType.DIRECT:
+            user_id = channel_msg.identity.channel_user_id
+            return self.check_dm_access(user_id)
+
+        space_id = channel_msg.metadata.get("space_name", "")
+        if not self.check_group_access(space_id):
+            return False
+
+        per_group = self._resolve_per_group(space_id)
+
+        if per_group is not None and per_group.users:
+            user_id = channel_msg.identity.channel_user_id
+            if user_id not in per_group.users:
+                logger.debug(f"Group message filtered: user {user_id} not in per-group users whitelist")
+                return False
+
+        require_mention = self.require_mention
+        if per_group is not None:
+            require_mention = per_group.require_mention
+
+        if require_mention and chat_type != ChatType.DIRECT:
+            mentions = channel_msg.mentions
+            if mentions and not mentions.is_bot_mentioned:
+                logger.debug("Group message filtered: bot not mentioned")
+                return False
+
+        return True
+
+    def get_per_group_system_prompt(self, space_id: str) -> str:
+        per_group = self._resolve_per_group(space_id)
+        if per_group and per_group.system_prompt:
+            return per_group.system_prompt
+        return ""
+
+    def _resolve_per_group(self, space_id: str) -> PerGroupConfig | None:
+        if not space_id or not self.groups:
+            return None
+        return self.groups.get(space_id)
+
+    def _match_allowlist(self, target: str, allowlist: list[str]) -> bool:
+        if not allowlist:
+            return False
+        if "*" in allowlist:
+            return True
+        if target in allowlist:
+            return True
+        if self.dangerously_allow_name_matching and "@" in target:
+            email_local = target.split("@")[0].lower()
+            for entry in allowlist:
+                if "@" in entry and entry.split("@")[0].lower() == email_local:
+                    logger.warning(
+                        f"Dangerously matched target '{target}' to allowlist entry '{entry}' via email local-part"
+                    )
+                    return True
+        return False
+
+
+def _normalize_list(raw: Any) -> list[str]:
+    if isinstance(raw, str):
+        return [x.strip() for x in raw.split(",") if x.strip()]
+    if isinstance(raw, (list, tuple)):
+        return [str(x).strip() for x in raw if x]
+    return []
+
+
+def _parse_per_group_config(groups_raw: Any) -> dict[str, PerGroupConfig]:
+    if not isinstance(groups_raw, dict):
+        return {}
+    result: dict[str, PerGroupConfig] = {}
+    for space_name, cfg in groups_raw.items():
+        if not isinstance(cfg, dict):
+            continue
+        result[space_name] = PerGroupConfig(
+            require_mention=bool(cfg.get("require_mention", cfg.get("requireMention", False))),
+            enabled=bool(cfg.get("enabled", True)),
+            users=_normalize_list(cfg.get("users", [])),
+            system_prompt=str(cfg.get("system_prompt", cfg.get("systemPrompt", ""))),
+        )
+    return result

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

@@ -0,0 +1,87 @@
+from __future__ import annotations
+
+import os
+from typing import Any
+
+from yuxi.utils.logging_config import logger
+
+
+def resolve_proxy_config(config: dict[str, Any] | None = None) -> dict[str, str | None]:
+    proxy_config: dict[str, str | None] = {
+        "http": None,
+        "https": None,
+        "no_proxy": None,
+    }
+
+    if config:
+        proxy_config["http"] = config.get("httpProxy", config.get("HTTP_PROXY"))
+        proxy_config["https"] = config.get("httpsProxy", config.get("HTTPS_PROXY"))
+        proxy_config["no_proxy"] = config.get("noProxy", config.get("NO_PROXY"))
+
+    if not proxy_config["http"]:
+        proxy_config["http"] = os.getenv("HTTP_PROXY")
+    if not proxy_config["https"]:
+        proxy_config["https"] = os.getenv("HTTPS_PROXY")
+    if not proxy_config["no_proxy"]:
+        proxy_config["no_proxy"] = os.getenv("NO_PROXY")
+
+    if proxy_config["http"] and not proxy_config["https"]:
+        proxy_config["https"] = proxy_config["http"]
+
+    return proxy_config
+
+
+def build_proxies_dict(proxy_config: dict[str, str | None]) -> dict[str, str] | None:
+    proxies: dict[str, str] = {}
+    if proxy_config.get("http"):
+        proxies["http"] = proxy_config["http"]
+    if proxy_config.get("https"):
+        proxies["https"] = proxy_config["https"]
+    if proxy_config.get("no_proxy"):
+        proxies["no_proxy"] = proxy_config["no_proxy"]
+
+    if proxies:
+        logger.debug(f"Proxy configured: https={proxies.get('https')}")
+        return proxies
+    return None
+
+
+def build_google_auth_request(proxy_config: dict[str, str | None] | None = None) -> Any:
+    try:
+        from google.auth.transport.requests import Request as GARequest
+    except ImportError:
+        return None
+
+    proxies = build_proxies_dict(proxy_config or {})
+    if not proxies:
+        return GARequest()
+
+    try:
+        result = GARequest()
+        result.session.proxies.update(proxies)
+        return result
+    except Exception:
+        logger.debug("Failed to apply proxy to Google Auth request, using default")
+        return GARequest()
+
+
+def resolve_tls_config(config: dict[str, Any] | None = None) -> dict[str, str | None]:
+    tls_config: dict[str, str | None] = {
+        "cert": None,
+        "key": None,
+    }
+
+    if config:
+        tls_config["cert"] = config.get("cert", config.get("tlsCert"))
+        tls_config["key"] = config.get("key", config.get("tlsKey"))
+
+    if not tls_config["cert"]:
+        tls_config["cert"] = os.getenv("GOOGLE_CHAT_TLS_CERT")
+    if not tls_config["key"]:
+        tls_config["key"] = os.getenv("GOOGLE_CHAT_TLS_KEY")
+
+    return tls_config
+
+
+def has_tls_config(tls_config: dict[str, str | None]) -> bool:
+    return bool(tls_config.get("cert"))

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

@@ -0,0 +1,21 @@
+from __future__ import annotations
+
+import json
+import base64
+
+from yuxi.utils.logging_config import logger
+
+
+def decode_pubsub_push(body: dict) -> dict | None:
+    message = body.get("message", {})
+    data_b64 = message.get("data", "")
+
+    if not data_b64:
+        return None
+
+    try:
+        decoded = base64.b64decode(data_b64).decode("utf-8")
+        return json.loads(decoded)
+    except Exception as e:
+        logger.warning(f"Failed to decode Pub/Sub push data: {e}")
+        return None

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

@@ -0,0 +1,54 @@
+from __future__ import annotations
+
+from typing import Any
+
+from yuxi.utils.logging_config import logger
+
+_SECRET_CONTRACTS: dict[str, dict[str, Any]] = {
+    "serviceAccount": {
+        "name": "serviceAccount",
+        "description": "Google Chat service account JSON credentials (inline)",
+        "required": True,
+        "sensitive": True,
+        "validate": "_validate_service_account_json",
+    },
+    "serviceAccountFile": {
+        "name": "serviceAccountFile",
+        "description": "Path to Google Chat service account JSON key file",
+        "required": False,
+        "sensitive": True,
+    },
+    "serviceAccountRef": {
+        "name": "serviceAccountRef",
+        "description": "Secret reference to Google Chat service account credentials",
+        "required": False,
+        "sensitive": True,
+    },
+    "webhookSecret": {
+        "name": "webhookSecret",
+        "description": "Webhook secret for Pub/Sub push endpoint (if applicable)",
+        "required": False,
+        "sensitive": True,
+    },
+}
+
+
+def get_secret_contracts() -> dict[str, dict[str, Any]]:
+    return dict(_SECRET_CONTRACTS)
+
+
+def register_secret_contract(name: str, contract: dict[str, Any]) -> None:
+    _SECRET_CONTRACTS[name] = contract
+    logger.debug(f"Registered secret contract: {name}")
+
+
+def get_required_secrets() -> list[str]:
+    return [name for name, contract in _SECRET_CONTRACTS.items() if contract.get("required")]
+
+
+def get_sensitive_fields() -> set[str]:
+    return {name for name, contract in _SECRET_CONTRACTS.items() if contract.get("sensitive")}
+
+
+def is_sensitive_field(field_name: str) -> bool:
+    return field_name in get_sensitive_fields() or field_name.endswith("_key") or field_name.endswith("_secret")

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

@@ -0,0 +1,314 @@
+from __future__ import annotations
+
+import asyncio
+import io
+import random
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from googleapiclient.discovery import Resource
+
+from yuxi.channels.exceptions import (
+    ChannelAuthenticationError,
+    ChannelRateLimitError,
+    DeliveryFailedError,
+)
+from yuxi.channels.models import DeliveryResult
+from yuxi.utils.logging_config import logger
+
+_MAX_RETRIES = 5
+_MAX_BACKOFF_S = 64
+_AUTH_401_MAX_BACKOFF_S = 300
+_AUTH_401_COOLDOWN: dict[str, float] = {}
+
+
+def _check_401_cooldown(account_id: str = "default") -> bool:
+    now = __import__("time").monotonic()
+    cooldown_until = _AUTH_401_COOLDOWN.get(account_id, 0)
+    if now < cooldown_until:
+        remaining = cooldown_until - now
+        logger.warning(f"Auth 401 cooldown active for {remaining:.0f}s")
+        return False
+    return True
+
+
+def _apply_401_cooldown(account_id: str = "default", backoff_s: float = 60.0) -> None:
+    now = __import__("time").monotonic()
+    capped = min(backoff_s, _AUTH_401_MAX_BACKOFF_S)
+    _AUTH_401_COOLDOWN[account_id] = now + capped
+    logger.info(f"Auth 401 backoff applied: {capped:.0f}s cooldown")
+
+
+def _extend_401_cooldown(account_id: str = "default") -> None:
+    now = __import__("time").monotonic()
+    existing = _AUTH_401_COOLDOWN.get(account_id, 0)
+    remaining = max(existing - now, 0)
+    extension = min(remaining * 2 + 30, _AUTH_401_MAX_BACKOFF_S)
+    _AUTH_401_COOLDOWN[account_id] = now + extension
+    logger.warning(f"Auth 401 cooldown extended to {extension:.0f}s")
+
+
+def _classify_error(e: Exception, account_id: str = "default") -> None:
+    try:
+        from googleapiclient.errors import HttpError
+
+        if isinstance(e, HttpError):
+            status = e.resp.status
+            if status == 429:
+                raise ChannelRateLimitError()
+            if status in (401, 403):
+                reason = _extract_rate_limit_reason(e)
+                if reason:
+                    raise ChannelRateLimitError()
+                if status == 401:
+                    _extend_401_cooldown(account_id)
+                raise ChannelAuthenticationError(str(e))
+            raise DeliveryFailedError(str(e)) from e
+    except ImportError:
+        raise DeliveryFailedError(str(e)) from None
+    raise DeliveryFailedError(str(e)) from e
+
+
+def _extract_rate_limit_reason(e: Exception) -> bool:
+    try:
+        content = e.content if hasattr(e, "content") else b""
+        if isinstance(content, bytes):
+            content = content.decode("utf-8", errors="replace")
+        if isinstance(content, str) and ("rateLimitExceeded" in content or "userRateLimitExceeded" in content):
+            return True
+    except Exception:
+        pass
+    return False
+
+
+def _backoff_with_jitter(retry_count: int) -> float:
+    delay = min((2**retry_count) + random.uniform(0, 1), _MAX_BACKOFF_S)
+    return delay
+
+
+async def _retry_with_backoff(func, *args, account_id: str = "default", **kwargs) -> DeliveryResult:
+    last_error = None
+    for attempt in range(_MAX_RETRIES):
+        try:
+            return await func(*args, **kwargs)
+        except ChannelRateLimitError as e:
+            last_error = e
+            delay = _backoff_with_jitter(attempt)
+            logger.warning(f"Rate limited (attempt {attempt + 1}/{_MAX_RETRIES}), backing off {delay:.1f}s")
+            await asyncio.sleep(delay)
+        except (ChannelAuthenticationError, DeliveryFailedError):
+            raise
+        except Exception as e:
+            last_error = e
+            if attempt < _MAX_RETRIES - 1:
+                delay = _backoff_with_jitter(attempt)
+                logger.warning(f"Request failed (attempt {attempt + 1}/{_MAX_RETRIES}): {e}, retrying in {delay:.1f}s")
+                await asyncio.sleep(delay)
+            else:
+                break
+
+    error_msg = str(last_error) if last_error else "Max retries exceeded"
+    logger.error(f"send failed after {_MAX_RETRIES} retries: {error_msg}")
+    return DeliveryResult(success=False, error=error_msg)
+
+
+async def send_message(
+    chat_service: Resource,
+    chat_id: str,
+    body: dict[str, Any],
+    account_id: str = "default",
+) -> DeliveryResult:
+    async def _do_send():
+        result = chat_service.spaces().messages().create(parent=chat_id, body=body).execute()
+        return DeliveryResult(success=True, message_id=result.get("name", ""))
+
+    try:
+        return await _retry_with_backoff(_do_send, account_id=account_id)
+    except (ChannelAuthenticationError, DeliveryFailedError, ChannelRateLimitError) as e:
+        return DeliveryResult(success=False, error=str(e))
+    except Exception as e:
+        _classify_error(e, account_id)
+        return DeliveryResult(success=False, error=str(e))
+
+
+async def update_message(
+    chat_service: Resource,
+    message_id: str,
+    body: dict[str, Any],
+    update_mask: str = "text",
+    account_id: str = "default",
+) -> DeliveryResult:
+    async def _do_send():
+        result = chat_service.spaces().messages().update(name=message_id, updateMask=update_mask, body=body).execute()
+        return DeliveryResult(success=True, message_id=result.get("name", message_id))
+
+    try:
+        return await _retry_with_backoff(_do_send, account_id=account_id)
+    except (ChannelAuthenticationError, DeliveryFailedError, ChannelRateLimitError) as e:
+        return DeliveryResult(success=False, error=str(e))
+    except Exception as e:
+        _classify_error(e, account_id)
+        return DeliveryResult(success=False, error=str(e))
+
+
+async def delete_message(
+    chat_service: Resource,
+    message_id: str,
+    account_id: str = "default",
+) -> DeliveryResult:
+    async def _do_send():
+        chat_service.spaces().messages().delete(name=message_id).execute()
+        return DeliveryResult(success=True)
+
+    try:
+        return await _retry_with_backoff(_do_send, account_id=account_id)
+    except (ChannelAuthenticationError, DeliveryFailedError, ChannelRateLimitError) as e:
+        return DeliveryResult(success=False, error=str(e))
+    except Exception as e:
+        _classify_error(e, account_id)
+        return DeliveryResult(success=False, error=str(e))
+
+
+async def send_media(
+    chat_service: Resource,
+    chat_id: str,
+    media_url: str,
+    caption: str = "",
+    account_id: str = "default",
+) -> DeliveryResult:
+    async def _do_send():
+        body: dict[str, Any] = {"text": caption or " "}
+        body["cards_v2"] = [{"card": {"sections": [{"widgets": [{"image": {"imageUrl": media_url}}]}]}}]
+        result = chat_service.spaces().messages().create(parent=chat_id, body=body).execute()
+        return DeliveryResult(success=True, message_id=result.get("name", ""))
+
+    try:
+        return await _retry_with_backoff(_do_send, account_id=account_id)
+    except (ChannelAuthenticationError, DeliveryFailedError, ChannelRateLimitError) as e:
+        return DeliveryResult(success=False, error=str(e))
+    except Exception as e:
+        _classify_error(e, account_id)
+        return DeliveryResult(success=False, error=str(e))
+
+
+async def send_reaction(
+    chat_service: Resource,
+    message_id: str,
+    emoji: str,
+    account_id: str = "default",
+) -> DeliveryResult:
+    async def _do_send():
+        body = {"emoji": {"unicode": emoji}}
+        (chat_service.spaces().messages().reactions().create(parent=message_id, body=body).execute())
+        return DeliveryResult(success=True)
+
+    try:
+        return await _retry_with_backoff(_do_send, account_id=account_id)
+    except (ChannelAuthenticationError, DeliveryFailedError, ChannelRateLimitError) as e:
+        return DeliveryResult(success=False, error=str(e))
+    except Exception as e:
+        _classify_error(e, account_id)
+        return DeliveryResult(success=False, error=str(e))
+
+
+async def list_reactions(
+    chat_service: Resource,
+    message_id: str,
+) -> list[dict[str, Any]]:
+    try:
+        result = chat_service.spaces().messages().reactions().list(parent=message_id).execute()
+        return result.get("reactions", [])
+    except Exception as e:
+        logger.warning(f"list_reactions failed for {message_id}: {e}")
+        return []
+
+
+async def delete_reaction(
+    chat_service: Resource,
+    reaction_id: str,
+    account_id: str = "default",
+) -> DeliveryResult:
+    async def _do_send():
+        chat_service.spaces().messages().reactions().delete(name=reaction_id).execute()
+        return DeliveryResult(success=True)
+
+    try:
+        return await _retry_with_backoff(_do_send, account_id=account_id)
+    except (ChannelAuthenticationError, DeliveryFailedError, ChannelRateLimitError) as e:
+        return DeliveryResult(success=False, error=str(e))
+    except Exception as e:
+        _classify_error(e, account_id)
+        return DeliveryResult(success=False, error=str(e))
+
+
+async def upload_file_message(
+    chat_service: Resource,
+    chat_id: str,
+    file_data: bytes,
+    filename: str,
+    mime_type: str = "application/octet-stream",
+    caption: str = "",
+    account_id: str = "default",
+) -> DeliveryResult:
+    from googleapiclient.http import MediaIoBaseUpload
+
+    async def _do_send():
+        media = MediaIoBaseUpload(
+            io.BytesIO(file_data),
+            mimetype=mime_type,
+            resumable=True,
+        )
+        body: dict[str, Any] = {"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", ""))
+
+    try:
+        return await _retry_with_backoff(_do_send, account_id=account_id)
+    except (ChannelAuthenticationError, DeliveryFailedError, ChannelRateLimitError) as e:
+        return DeliveryResult(success=False, error=str(e))
+    except Exception as e:
+        _classify_error(e, account_id)
+        return DeliveryResult(success=False, error=str(e))
+
+
+async def upload_image_message(
+    chat_service: Resource,
+    chat_id: str,
+    image_data: bytes,
+    filename: str = "image.png",
+    caption: str = "",
+    account_id: str = "default",
+) -> DeliveryResult:
+    return await upload_file_message(
+        chat_service, chat_id, image_data, filename, "image/png", caption, account_id=account_id
+    )
+
+
+async def find_direct_message_space(
+    chat_service: Resource,
+    user_name: str,
+) -> str | None:
+    try:
+        result = chat_service.spaces().findDirectMessage(query=user_name).execute()
+        return result.get("name", "")
+    except Exception as e:
+        logger.warning(f"findDirectMessage failed for user {user_name}: {e}")
+        return None
+
+
+async def resolve_outbound_space(
+    chat_service: Resource,
+    target: str,
+) -> str | None:
+    if target.startswith("spaces/"):
+        try:
+            result = chat_service.spaces().get(name=target).execute()
+            return result.get("name", "")
+        except Exception:
+            return target
+
+    dm_result = await find_direct_message_space(chat_service, target)
+    if dm_result:
+        return dm_result
+
+    return None

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

@@ -0,0 +1,9 @@
+from __future__ import annotations
+
+
+def build_thread_id(agent_id: str, space_name: str, user_email: str = "") -> str:
+    if space_name.startswith("spaces/"):
+        space_id = space_name.removeprefix("spaces/")
+        return f"agent:{agent_id}:googlechat:space:{space_id}"
+
+    return f"agent:main:googlechat:dm:{user_email}"

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

@@ -0,0 +1,339 @@
+from __future__ import annotations
+
+from typing import Any
+
+
+_SETUP_GUIDE = """
+Google Chat Setup Guide
+=======================
+
+1. Go to Google Cloud Console → APIs & Services → Enable APIs
+   Enable the "Google Chat API"
+
+2. Go to IAM & Admin → Service Accounts
+   Create a new service account or use an existing one
+   Generate a JSON key and download it
+
+3. Go to Google Chat API → Configuration
+   Set up the bot:
+   - App name: Your bot name
+   - Avatar URL: (optional)
+   - Description: Your bot description
+   - Functionality: Enable "Receive messages" and "Join spaces and group conversations"
+   - Connection settings:
+     * App URL: https://your-domain.com/api/webhook/googlechat
+     * (Or use Pub/Sub for production)
+
+4. Configure the adapter with your service account credentials
+   using one of the supported methods (file, inline, or environment variable).
+
+5. Optional: Set audienceType to "project-number" if using project-number auth,
+   and configure the audience accordingly.
+
+6. Configure security policies:
+   - dmPolicy: "open" | "pairing" | "allowlist" | "disabled"
+   - groupPolicy: "open" | "allowlist" | "disabled"
+   - allowFrom: list of users/spaces allowed to interact
+   - requireMention: require @mention in group messages
+"""
+
+
+class SetupWizard:
+    def __init__(self, adapter=None):
+        self._adapter = adapter
+        self._steps: list[dict[str, Any]] = []
+        self._current_step = 0
+        self._results: dict[str, Any] = {}
+
+    @property
+    def current_step(self) -> dict[str, Any] | None:
+        if 0 <= self._current_step < len(self._steps):
+            return self._steps[self._current_step]
+        return None
+
+    def start(self) -> dict[str, Any]:
+        self._results = {}
+        self._current_step = 0
+        self._steps = [
+            self._step_credential_mode(),
+            self._step_credential_input(),
+            self._step_audience_config(),
+            self._step_policy_config(),
+            self._step_review(),
+        ]
+        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:
+            self._results[self._steps[self._current_step - 1]["id"]] = answer
+
+        self._current_step += 1
+        if self._current_step >= len(self._steps):
+            return {"step": self._current_step, "done": True, "config": self._build_config()}
+
+        return {"step": self._current_step, "steps": len(self._steps), "data": self.current_step}
+
+    def _step_credential_mode(self) -> dict[str, Any]:
+        return {
+            "id": "credential_mode",
+            "question": "How would you like to provide service account credentials?",
+            "type": "select",
+            "options": [
+                {
+                    "id": "env_file",
+                    "label": "Environment (GOOGLE_SERVICE_ACCOUNT_FILE)",
+                    "description": (
+                        "Set the path to a service account JSON key file "
+                        "via the GOOGLE_SERVICE_ACCOUNT_FILE environment variable"
+                    ),
+                },
+                {
+                    "id": "env_json",
+                    "label": "Environment (GOOGLE_CHAT_SERVICE_ACCOUNT)",
+                    "description": (
+                        "Paste the full service account JSON content "
+                        "into the GOOGLE_CHAT_SERVICE_ACCOUNT environment variable"
+                    ),
+                },
+                {
+                    "id": "file",
+                    "label": "File (serviceAccountFile)",
+                    "description": "Specify the path to a service account JSON key file on disk in the config",
+                },
+                {
+                    "id": "inline",
+                    "label": "Inline (serviceAccount)",
+                    "description": "Paste the full service account JSON content directly into the config",
+                },
+                {
+                    "id": "secret_ref",
+                    "label": "Secret Reference (serviceAccountRef)",
+                    "description": "Reference a secret via the secret management system",
+                },
+            ],
+        }
+
+    def _step_credential_input(self) -> dict[str, Any]:
+        mode = self._results.get("credential_mode", "")
+        if mode == "file":
+            return {
+                "id": "credential_value",
+                "question": "Please provide the path to your Google Cloud service account JSON key file:",
+                "type": "text",
+                "required": True,
+                "hint": "e.g., /etc/secrets/google-service-account.json or ~/.config/gcp-sa-key.json",
+            }
+        if mode == "inline":
+            return {
+                "id": "credential_value",
+                "question": "Please paste the full content of your Google Cloud service account JSON key:",
+                "type": "text",
+                "required": True,
+                "hint": (
+                    "This file can be downloaded from Google Cloud Console "
+                    "→ IAM & Admin → Service Accounts → Create Key → JSON"
+                ),
+            }
+        if mode == "secret_ref":
+            return {
+                "id": "credential_value",
+                "question": "Please provide the secret reference identifier:",
+                "type": "text",
+                "required": True,
+                "hint": "e.g., myapp/gcp/googlechat-sa-key",
+            }
+        return {
+            "id": "credential_value",
+            "question": (
+                    "Credential will be read from environment variables. "
+                    "Ensure GOOGLE_CHAT_SERVICE_ACCOUNT or "
+                    "GOOGLE_SERVICE_ACCOUNT_FILE is set."
+                ),
+            "type": "info",
+            "required": False,
+        }
+
+    def _step_audience_config(self) -> dict[str, Any]:
+        return {
+            "id": "audience_config",
+            "question": "Configure audience validation for webhook JWT:",
+            "type": "form",
+            "fields": [
+                {
+                    "id": "audience_type",
+                    "question": "Audience type:",
+                    "type": "select",
+                    "options": [
+                        {"id": "app-url", "label": "App URL (recommended)", "description": "Match the app's HTTPS URL"},
+                        {
+                            "id": "project-number",
+                            "label": "Project Number",
+                            "description": "Match the GCP project number",
+                        },
+                    ],
+                    "default": "app-url",
+                },
+                {
+                    "id": "audience",
+                    "question": "Audience value:",
+                    "type": "text",
+                    "hint": "For app-url: your HTTPS app URL. For project-number: your GCP project number.",
+                },
+            ],
+        }
+
+    def _step_policy_config(self) -> dict[str, Any]:
+        return {
+            "id": "policy_config",
+            "question": "Configure access policies:",
+            "type": "form",
+            "fields": [
+                {
+                    "id": "dm_policy",
+                    "question": "DM Policy:",
+                    "type": "select",
+                    "options": [
+                        {"id": "open", "label": "Open", "description": "Any user can DM the bot"},
+                        {"id": "pairing", "label": "Pairing", "description": "Users must complete a pairing challenge"},
+                        {"id": "allowlist", "label": "Allowlist", "description": "Only users in allowFrom can DM"},
+                        {"id": "disabled", "label": "Disabled", "description": "No DM access"},
+                    ],
+                    "default": "open",
+                },
+                {
+                    "id": "group_policy",
+                    "question": "Group Policy:",
+                    "type": "select",
+                    "options": [
+                        {"id": "open", "label": "Open", "description": "Bot responds in all groups"},
+                        {
+                            "id": "allowlist",
+                            "label": "Allowlist",
+                            "description": "Only spaces in groupAllowFrom are active",
+                        },
+                        {"id": "disabled", "label": "Disabled", "description": "No group access"},
+                    ],
+                    "default": "open",
+                },
+                {
+                    "id": "require_mention",
+                    "question": "Require @mention in groups?",
+                    "type": "boolean",
+                    "default": False,
+                },
+            ],
+        }
+
+    def _step_review(self) -> dict[str, Any]:
+        return {
+            "id": "review",
+            "question": "Review configuration before saving:",
+            "type": "review",
+            "summary": self._build_config(),
+        }
+
+    def _build_config(self) -> dict[str, Any]:
+        config: dict[str, Any] = {}
+        mode = self._results.get("credential_mode", "")
+        credential_value = self._results.get("credential_value", "")
+
+        if mode == "inline" and credential_value:
+            config["serviceAccount"] = credential_value
+        elif mode == "file" and credential_value:
+            config["serviceAccountFile"] = credential_value
+        elif mode == "secret_ref" and credential_value:
+            config["serviceAccountRef"] = credential_value
+
+        audience = self._results.get("audience_config", {})
+        if isinstance(audience, dict):
+            if audience.get("audience_type"):
+                config["audienceType"] = audience["audience_type"]
+            if audience.get("audience"):
+                config["audience"] = audience["audience"]
+
+        policy = self._results.get("policy_config", {})
+        if isinstance(policy, dict):
+            if policy.get("dm_policy"):
+                config["dmPolicy"] = policy["dm_policy"]
+            if policy.get("group_policy"):
+                config["groupPolicy"] = policy["group_policy"]
+            if policy.get("require_mention") is not None:
+                config["requireMention"] = policy["require_mention"]
+
+        config["enabled"] = True
+        return config
+
+
+def wizard_prompt_credential_mode() -> dict[str, Any]:
+    return {
+        "step": "select_credential_mode",
+        "question": "How would you like to provide service account credentials?",
+        "options": [
+            {
+                "id": "env_file",
+                "label": "Environment (GOOGLE_SERVICE_ACCOUNT_FILE)",
+                "description": (
+                        "Set the path to a service account JSON key file "
+                        "via the GOOGLE_SERVICE_ACCOUNT_FILE environment variable"
+                    ),
+                },
+                {
+                    "id": "env_json",
+                    "label": "Environment (GOOGLE_CHAT_SERVICE_ACCOUNT)",
+                    "description": (
+                        "Paste the full service account JSON content "
+                        "into the GOOGLE_CHAT_SERVICE_ACCOUNT environment variable"
+                    ),
+            },
+            {
+                "id": "file",
+                "label": "File (serviceAccountFile)",
+                "description": "Specify the path to a service account JSON key file on disk",
+            },
+            {
+                "id": "inline",
+                "label": "Inline (serviceAccount)",
+                "description": "Paste the full service account JSON content directly into the config",
+            },
+        ],
+    }
+
+
+def wizard_prompt_file_path() -> dict[str, Any]:
+    return {
+        "step": "provide_credential_path",
+        "question": "Please provide the path to your Google Cloud service account JSON key file:",
+        "required": True,
+    }
+
+
+def wizard_prompt_service_account_json() -> dict[str, Any]:
+    return {
+        "step": "provide_credential_json",
+        "question": (
+            "Please paste the full content of your Google Cloud service account "
+            "JSON key (or set the environment variable directly):"
+        ),
+        "required": True,
+        "hint": (
+            "This file can be downloaded from Google Cloud Console "
+            "→ IAM & Admin → Service Accounts → Create Key → JSON"
+        ),
+    }
+
+
+def wizard_validate_service_account(cred_json: dict) -> dict[str, Any]:
+    errors: list[str] = []
+    if cred_json.get("type") != "service_account":
+        errors.append("Invalid credential type: expected 'service_account'")
+    if not cred_json.get("private_key"):
+        errors.append("Missing 'private_key' field")
+    if not cred_json.get("client_email"):
+        errors.append("Missing 'client_email' field")
+    if not cred_json.get("token_uri"):
+        errors.append("Missing 'token_uri' field")
+    return {"valid": len(errors) == 0, "errors": errors}
+
+
+def get_setup_guide() -> str:
+    return _SETUP_GUIDE.strip()

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

@@ -0,0 +1,33 @@
+from __future__ import annotations
+
+
+SLASH_COMMAND_MAP: dict[str, str] = {
+    "/reset": "清除当前会话上下文，重新开始对话",
+    "/history": "查看当前会话的对话历史摘要",
+    "/context": "查看当前会话的上下文信息",
+    "/summary": "生成当前会话的总结",
+    "/help": "显示可用命令列表",
+    "/status": "查看 Bot 状态",
+}
+
+
+def extract_command(content: str) -> tuple[str | None, str]:
+    stripped = content.strip()
+    if not stripped.startswith("/"):
+        return None, content
+
+    parts = stripped.split(maxsplit=1)
+    command = parts[0].lower()
+    args = parts[1] if len(parts) > 1 else ""
+
+    if command in SLASH_COMMAND_MAP:
+        return command, args
+
+    return None, content
+
+
+def get_command_help() -> str:
+    lines = ["可用命令："]
+    for cmd, desc in SLASH_COMMAND_MAP.items():
+        lines.append(f"  {cmd} - {desc}")
+    return "\n".join(lines)

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

@@ -0,0 +1,71 @@
+from __future__ import annotations
+
+from typing import Any
+
+from yuxi.utils.logging_config import logger
+
+_UNSAFE_TRANSPORT_FIELDS = frozenset(
+    {
+        "agent",
+        "cert",
+        "cert_file",
+        "key",
+        "key_file",
+        "dispatcher",
+        "proxy",
+        "session",
+    }
+)
+
+_AUTH_RESPONSE_MAX_BYTES = 1 * 1024 * 1024
+
+
+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:
+            removed.append(key)
+            continue
+        cleaned[key] = value
+    if removed:
+        logger.warning(f"SSRF guard: removed unsafe transport fields: {removed}")
+    return cleaned
+
+
+def sanitize_request_kwargs(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:
+            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 {
+        "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:
+        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")
+    truncated = check_auth_response_size(data)
+    return truncated.decode("utf-8", errors="replace")

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

@@ -0,0 +1,190 @@
+from __future__ import annotations
+
+import time
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from googleapiclient.discovery import Resource
+
+from yuxi.channels.models import DeliveryResult
+from yuxi.utils.logging_config import logger
+
+from .send import update_message, send_message, delete_message
+
+DEFAULT_UPDATE_INTERVAL_MS = 800
+DEFAULT_COALESCE_MIN_CHARS = 1500
+DEFAULT_COALESCE_IDLE_MS = 1000
+
+
+class StreamManager:
+    def __init__(
+        self,
+        update_interval_ms: int = DEFAULT_UPDATE_INTERVAL_MS,
+        coalesce_min_chars: int = DEFAULT_COALESCE_MIN_CHARS,
+        coalesce_idle_ms: int = DEFAULT_COALESCE_IDLE_MS,
+    ):
+        self._messages: dict[str, str] = {}
+        self._texts: dict[str, str] = {}
+        self._last_update: dict[str, float] = {}
+        self._update_interval_ms = update_interval_ms
+        self._coalesce_min_chars = coalesce_min_chars
+        self._coalesce_idle_ms = coalesce_idle_ms
+
+        self._typing_messages: dict[str, str] = {}
+        self._typing_replaced: set[str] = set()
+
+    @property
+    def update_interval_ms(self) -> int:
+        return self._update_interval_ms
+
+    def has_pending(self, chat_id: str) -> bool:
+        return chat_id in self._messages
+
+    def get_message_name(self, chat_id: str) -> str | None:
+        return self._messages.get(chat_id)
+
+    def get_accumulated_text(self, chat_id: str) -> str:
+        return self._texts.get(chat_id, "")
+
+    def register_message(self, chat_id: str, message_name: str, text: str) -> None:
+        self._messages[chat_id] = message_name
+        self._texts[chat_id] = text
+        self._last_update[chat_id] = time.monotonic()
+
+    def register_first_chunk(self, chat_id: str, message_name: str, text: str) -> None:
+        self._messages[chat_id] = message_name
+        self._texts[chat_id] = text
+        self._last_update[chat_id] = time.monotonic()
+
+        typing_id = self._typing_messages.pop(chat_id, None)
+        if typing_id:
+            self._typing_replaced.add(chat_id)
+            logger.debug(f"Stream: replaced typing message {typing_id} with first chunk {message_name}")
+
+    def append_text(self, chat_id: str, chunk: str) -> str:
+        current = self._texts.get(chat_id, "")
+        current += chunk
+        self._texts[chat_id] = current
+        return current
+
+    def should_update(self, chat_id: str) -> bool:
+        last = self._last_update.get(chat_id, 0)
+        elapsed_ms = (time.monotonic() - last) * 1000
+
+        if elapsed_ms >= self._update_interval_ms:
+            return True
+
+        if self._coalesce_idle_ms > 0:
+            idle_threshold = max(self._coalesce_idle_ms, self._update_interval_ms)
+            if elapsed_ms >= idle_threshold:
+                new_chars_since_last = 0
+                return new_chars_since_last >= self._coalesce_min_chars
+
+        return False
+
+    def mark_update(self, chat_id: str) -> None:
+        self._last_update[chat_id] = time.monotonic()
+
+    async def create_typing_message(
+        self,
+        chat_service: Resource,
+        chat_id: str,
+        bot_name: str = "Bot",
+    ) -> str | None:
+        if chat_id in self._typing_messages:
+            return self._typing_messages[chat_id]
+
+        try:
+            body = {"text": f"*{bot_name} is typing...*"}
+            result = await send_message(chat_service, chat_id, body)
+            if result.success and result.message_id:
+                self._typing_messages[chat_id] = result.message_id
+                return result.message_id
+        except Exception as e:
+            logger.warning(f"Failed to create typing message in {chat_id}: {e}")
+
+        return None
+
+    async def clear_typing_message(
+        self,
+        chat_service: Resource,
+        chat_id: str,
+    ) -> None:
+        msg_id = self._typing_messages.pop(chat_id, None)
+        if msg_id and chat_id not in self._typing_replaced:
+            try:
+                await delete_message(chat_service, msg_id)
+            except Exception as e:
+                logger.warning(f"Failed to delete typing message {msg_id}: {e}")
+        self._typing_replaced.discard(chat_id)
+
+    async def handle_error(
+        self,
+        chat_service: Resource,
+        chat_id: str,
+    ) -> None:
+        await self.clear_typing_message(chat_service, chat_id)
+        self._messages.pop(chat_id, None)
+        self._texts.pop(chat_id, None)
+        self._last_update.pop(chat_id, None)
+
+    async def finalize(
+        self,
+        chat_service: Resource,
+        chat_id: str,
+    ) -> DeliveryResult:
+        await self.clear_typing_message(chat_service, chat_id)
+
+        message_name = self._messages.pop(chat_id, None)
+        text = self._texts.pop(chat_id, None)
+        self._last_update.pop(chat_id, None)
+
+        if not message_name or text is None:
+            return DeliveryResult(success=False, error="No pending stream message")
+
+        return await update_message(chat_service, message_name, {"text": text})
+
+    async def send_update(
+        self,
+        chat_service: Resource,
+        chat_id: str,
+        finished: bool = False,
+    ) -> DeliveryResult | None:
+        message_name = self._messages.get(chat_id)
+        text = self._texts.get(chat_id)
+        if not message_name or text is None:
+            return None
+
+        if chat_id in self._typing_messages and text:
+            typing_id = self._typing_messages.pop(chat_id, None)
+            if typing_id:
+                try:
+                    await delete_message(chat_service, typing_id)
+                except Exception:
+                    pass
+            self._typing_replaced.add(chat_id)
+
+        body = {"text": text if finished else text + " ⏳"}
+        result = await update_message(chat_service, message_name, body)
+
+        if result.success:
+            if finished:
+                self._messages.pop(chat_id, None)
+                self._texts.pop(chat_id, None)
+                self._last_update.pop(chat_id, None)
+                await self.clear_typing_message(chat_service, chat_id)
+            else:
+                self._last_update[chat_id] = time.monotonic()
+
+        return result
+
+    def clear(self) -> None:
+        self._messages.clear()
+        self._texts.clear()
+        self._last_update.clear()
+        self._typing_messages.clear()
+        self._typing_replaced.clear()
+
+    @property
+    def pending_count(self) -> int:
+        return len(self._messages)

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

@@ -0,0 +1,92 @@
+from __future__ import annotations
+
+import re
+from typing import Any
+
+
+_CHANNEL_PREFIXES = ("googlechat:", "gchat:", "google-chat:")
+
+
+def normalize_googlechat_target(target: str) -> dict[str, Any]:
+    if not target:
+        return {"raw": target, "type": "unknown"}
+
+    target = _strip_channel_prefix(target)
+
+    if "@" in target and not target.startswith(("spaces/", "users/")):
+        target = _convert_email_to_user(target)
+
+    if target.startswith("spaces/"):
+        parts = target.removeprefix("spaces/").split("/")
+        space_id = parts[0]
+        thread_key = parts[2] if len(parts) > 2 and parts[1] == "threads" else None
+        return {
+            "raw": target,
+            "type": "thread" if thread_key else "space",
+            "space_id": space_id,
+            "space_name": f"spaces/{space_id}",
+            "thread_key": thread_key,
+            "full_target": target,
+        }
+
+    if target.startswith("users/"):
+        user_id = target.removeprefix("users/")
+        cleaned = user_id.removeprefix("user:")
+        return {
+            "raw": target,
+            "type": "user",
+            "user_id": cleaned,
+            "user_name": f"users/{cleaned}",
+            "full_target": target,
+        }
+
+    if target.startswith("user:"):
+        user_id = target.removeprefix("user:")
+        return {
+            "raw": target,
+            "type": "user",
+            "user_id": user_id,
+            "user_name": f"users/{user_id}",
+            "full_target": f"users/{user_id}",
+        }
+
+    return {"raw": target, "type": "unknown"}
+
+
+def is_googlechat_user_target(target: str) -> bool:
+    target = _strip_channel_prefix(target)
+    return target.startswith("users/") or target.startswith("user:")
+
+
+def is_googlechat_space_target(target: str) -> bool:
+    target = _strip_channel_prefix(target)
+    return target.startswith("spaces/")
+
+
+def resolve_targets(inputs: list[str], kind: str | None = None) -> list[dict[str, Any]]:
+    results: list[dict[str, Any]] = []
+
+    for target in inputs:
+        normalized = normalize_googlechat_target(target)
+        target_type = normalized.get("type", "unknown")
+
+        if kind and target_type != kind:
+            continue
+
+        results.append(normalized)
+
+    return results
+
+
+def _strip_channel_prefix(target: str) -> str:
+    for prefix in _CHANNEL_PREFIXES:
+        if target.lower().startswith(prefix):
+            return target[len(prefix) :]
+    return target
+
+
+def _convert_email_to_user(email: str) -> str:
+    email_match = re.match(r"^[\w.+-]+@[\w-]+\.[\w.-]+$", email)
+    if email_match:
+        return f"users/{email}"
+    return f"users/{email}"

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

@@ -0,0 +1,18 @@
+from __future__ import annotations
+
+
+def parse_thread_key(message: dict) -> str | None:
+    return message.get("thread", {}).get("threadKey")
+
+
+def extract_thread_metadata(message: dict) -> dict:
+    thread = message.get("thread", {})
+    return {
+        "thread_key": thread.get("threadKey"),
+        "thread_name": thread.get("name"),
+        "is_thread_root": not bool(thread.get("threadKey")),
+    }
+
+
+def is_thread_message(message: dict) -> bool:
+    return bool(message.get("thread", {}).get("threadKey"))