diff --git a/backend/package/yuxi/channel/extensions/googlechat/__init__.py b/backend/package/yuxi/channel/extensions/googlechat/__init__.py
new file mode 100644
index 00000000..ce3ef20a
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/googlechat/__init__.py
@@ -0,0 +1,344 @@
+from __future__ import annotations
+
+from yuxi.channel.capabilities import ChannelCapabilities
+from yuxi.channel.extensions.base import BaseChannelPlugin
+from yuxi.channel.extensions.googlechat.access_policy import apply_inbound_access_policy
+from yuxi.channel.extensions.googlechat.config import GoogleChatConfigAdapter
+from yuxi.channel.extensions.googlechat.dedupe import GoogleChatDedupeStore
+from yuxi.channel.extensions.googlechat.gateway import GoogleChatGatewayAdapter
+from yuxi.channel.extensions.googlechat.monitor import GoogleChatMonitor
+from yuxi.channel.extensions.googlechat.outbound import GoogleChatOutboundAdapter
+from yuxi.channel.extensions.googlechat.pairing import GoogleChatPairingAdapter
+from yuxi.channel.extensions.googlechat.security import GoogleChatSecurityAdapter
+from yuxi.channel.extensions.googlechat.session import GoogleChatSessionAdapter
+from yuxi.channel.extensions.googlechat.status import GoogleChatStatusAdapter
+from yuxi.channel.plugins.registry import ChannelPluginRegistry
+
+
+class GoogleChatPlugin(BaseChannelPlugin):
+    id = "googlechat"
+    name = "Google Chat"
+    order = 55
+    label = "Google Chat (Chat API)"
+    aliases = ["gchat", "google-chat"]
+    resolve_reply_to_mode = "off"
+
+    def __init__(self):
+        self._config = GoogleChatConfigAdapter()
+        self._gateway = GoogleChatGatewayAdapter()
+        self._outbound = GoogleChatOutboundAdapter()
+        self._monitor = GoogleChatMonitor()
+        self._status = GoogleChatStatusAdapter()
+        self._security = GoogleChatSecurityAdapter()
+        self._session = GoogleChatSessionAdapter()
+        self._pairing = GoogleChatPairingAdapter()
+        self._dedupe = GoogleChatDedupeStore()
+
+    @property
+    def capabilities(self) -> ChannelCapabilities:
+        return ChannelCapabilities(
+            chat_types=["direct", "group", "thread"],
+            message_types=["text", "image", "file", "video"],
+            interactions=[],
+            reactions=True,
+            typing_indicator=True,
+            threads=True,
+            edit=True,
+            unsend=True,
+            reply=True,
+            media=True,
+            native_commands=False,
+            polls=False,
+            streaming=False,
+            streaming_mode=None,
+            block_streaming=True,
+            block_streaming_chunk_min_chars=4000,
+            block_streaming_chunk_max_chars=4000,
+            block_streaming_coalesce_min_chars=1500,
+            block_streaming_coalesce_max_chars=4000,
+        )
+
+    # ── ConfigProtocol ───────────────────────────────────
+
+    def list_account_ids(self, config: dict) -> list[str]:
+        return self._config.list_account_ids(config)
+
+    async def resolve_account(self, account_id: str) -> dict:
+        return await self._config.resolve_account(account_id)
+
+    def is_configured(self, account: dict) -> bool:
+        return self._config.is_configured(account)
+
+    def is_enabled(self, account: dict) -> bool:
+        return self._config.is_enabled(account)
+
+    def disabled_reason(self, account: dict) -> str:
+        return self._config.disabled_reason(account)
+
+    async def resolve_allow_from(self, config: dict, account_id: str) -> list[str] | None:
+        return await self._config.resolve_allow_from(config, account_id)
+
+    def describe_account(self, account: dict) -> dict:
+        return self._config.describe_account(account)
+
+    # ── GatewayProtocol ───────────────────────────────────
+
+    async def start(self, ctx) -> object:
+        return await self._gateway.start(ctx)
+
+    async def stop(self, ctx) -> None:
+        await self._gateway.stop(ctx)
+
+    # ── OutboundProtocol ─────────────────────────────────
+
+    async def send_text(
+        self,
+        target_id: str,
+        content: str,
+        *,
+        reply_to_id: str | None = None,
+        thread_id: str | None = None,
+        account_id: str | None = None,
+    ) -> None:
+        await self._outbound.send_text(
+            target_id, content,
+            reply_to_id=reply_to_id, thread_id=thread_id, account_id=account_id,
+        )
+
+    async def send_media(
+        self,
+        target_id: str,
+        media_url: str,
+        media_type: str,
+        reply_to_id: str | None = None,
+        thread_id: str | None = None,
+        account_id: str | None = None,
+    ) -> None:
+        await self._outbound.send_media(
+            target_id, media_url, media_type,
+            reply_to_id=reply_to_id, thread_id=thread_id, account_id=account_id,
+        )
+
+    async def send_card(
+        self,
+        target_id: str,
+        card_content: dict,
+        *,
+        reply_to_id: str | None = None,
+        thread_id: str | None = None,
+        account_id: str | None = None,
+    ) -> str | None:
+        return await self._outbound.send_card(
+            target_id, card_content,
+            reply_to_id=reply_to_id, thread_id=thread_id, account_id=account_id,
+        )
+
+    # ── StatusProtocol ────────────────────────────────────
+
+    async def probe(self, account: dict) -> bool:
+        return await self._status.probe(account)
+
+    def build_summary(self, snapshot: object) -> dict:
+        return self._status.build_summary(snapshot)
+
+    # ── SecurityProtocol ─────────────────────────────────
+
+    async def check_allowlist(self, peer_id: str, channel_type: str) -> bool:
+        from yuxi.channel.runtime.manager import gateway
+
+        config = gateway.global_config if gateway is not None else {}
+        account_ids = self._config.list_account_ids(config)
+        account_id = account_ids[0] if account_ids else "default"
+        account = await self._config.resolve_account(account_id)
+
+        account_dict = account.get("resolved")
+        if not account_dict:
+            return True
+
+        is_dm = channel_type == "direct"
+        space_name = peer_id if not is_dm else ""
+        space_type = "GROUP_CHAT" if not is_dm else "DM"
+
+        result = apply_inbound_access_policy(
+            space_name=space_name,
+            space_type=space_type,
+            sender_name=peer_id,
+            is_dm=is_dm,
+            account=account,
+        )
+        return result.get("allowed", False)
+
+    def resolve_dm_policy(self) -> dict:
+        return self._security.resolve_dm_policy()
+
+    def collect_warnings(self, config: dict, account_id: str | None = None) -> list[str]:
+        return self._security.collect_warnings(config, account_id)
+
+    # ── PairingProtocol ───────────────────────────────────
+
+    async def generate_code(self, peer_id: str) -> str:
+        return await self._pairing.generate_code(peer_id)
+
+    async def verify_code(self, peer_id: str, code: str) -> bool:
+        return await self._pairing.verify_code(peer_id, code)
+
+    # ── ThreadingProtocol ─────────────────────────────────
+
+    def extract_thread_id(self, msg: object) -> str | None:
+        metadata = getattr(msg, "metadata", {}) or {}
+        thread_name = metadata.get("thread_name", "")
+        if thread_name:
+            return thread_name
+        return getattr(msg, "message_thread_id", None)
+
+    def resolve_reply_transport(self, msg: object, thread_id: str | None):
+        from yuxi.channel.protocols import ReplyTransport
+
+        return ReplyTransport(
+            thread_id=thread_id,
+            reply_mode=getattr(self, "resolve_reply_to_mode", "off"),
+        )
+
+    # ── MessagingProtocol ─────────────────────────────────
+
+    def resolve_session(self, msg):
+        return self._session.resolve_session(msg)
+
+    # ── ReactionProtocol ──────────────────────────────────
+
+    async def send_reaction(
+        self,
+        target_id: str,
+        message_id: str,
+        emoji: str,
+        *,
+        account_id: str | None = None,
+    ) -> None:
+        from yuxi.channel.extensions.googlechat.reactions import add_reaction
+
+        account = await self._config.resolve_account(account_id or "default")
+        resolved = account.get("resolved")
+        if resolved:
+            await add_reaction(resolved, message_id, emoji)
+
+    async def remove_reaction(
+        self,
+        target_id: str,
+        message_id: str,
+        emoji: str,
+        *,
+        account_id: str | None = None,
+    ) -> None:
+        from yuxi.channel.extensions.googlechat.reactions import remove_reaction
+
+        account = await self._config.resolve_account(account_id or "default")
+        resolved = account.get("resolved")
+        if resolved:
+            bot_user = resolved.bot_user if hasattr(resolved, "bot_user") else None
+            await remove_reaction(resolved, message_id, emoji, bot_user)
+
+    async def fetch_reactions(
+        self,
+        target_id: str,
+        message_id: str,
+        *,
+        account_id: str | None = None,
+    ) -> list[dict]:
+        from yuxi.channel.extensions.googlechat.reactions import get_reactions
+
+        account = await self._config.resolve_account(account_id or "default")
+        resolved = account.get("resolved")
+        if resolved:
+            return await get_reactions(resolved, message_id)
+        return []
+
+    # ── FormatProtocol ────────────────────────────────────
+
+    def markdown_to_native(self, md_text: str) -> dict | str:
+        from yuxi.channel.extensions.googlechat.formatting import sanitize_text
+
+        return {"text": sanitize_text(md_text)}
+
+    def native_to_markdown(self, native_content: dict | str) -> str:
+        if isinstance(native_content, dict):
+            return native_content.get("text", "")
+        return str(native_content)
+
+    # ── DedupeProtocol ────────────────────────────────────
+
+    def is_duplicate(self, key: str) -> bool:
+        return self._dedupe.is_duplicate(key)
+
+    def mark_seen(self, key: str) -> None:
+        self._dedupe.mark_seen(key)
+
+    @property
+    def ttl_seconds(self) -> int:
+        return 300
+
+    @property
+    def max_entries(self) -> int:
+        return 10000
+
+    # ── ErrorHandlingProtocol ─────────────────────────────
+
+    def classify_error(self, error: BaseException) -> object:
+        import httpx
+        from yuxi.channel.errors import classify_error
+        from yuxi.channel.protocols import ClassifiedError, ErrorSeverity
+
+        if isinstance(error, httpx.HTTPStatusError):
+            status = error.response.status_code
+            if status == 429:
+                return ClassifiedError(severity=ErrorSeverity.RATE_LIMITED)
+            if status in (401, 403):
+                return ClassifiedError(severity=ErrorSeverity.FORBIDDEN)
+            if 500 <= status < 600:
+                return ClassifiedError(severity=ErrorSeverity.NETWORK)
+            return ClassifiedError(severity=ErrorSeverity.RETRYABLE)
+        if isinstance(error, httpx.TimeoutException):
+            return ClassifiedError(severity=ErrorSeverity.NETWORK)
+        return classify_error(error)
+
+    def is_retryable(self, error: BaseException) -> bool:
+        import httpx
+        from yuxi.channel.errors import is_retryable
+
+        if isinstance(error, httpx.HTTPStatusError):
+            status = error.response.status_code
+            if status == 429:
+                return True
+            if status in (401, 403):
+                return False
+            if 500 <= status < 600:
+                return True
+            return True
+        if isinstance(error, httpx.TimeoutException):
+            return True
+        return is_retryable(error)
+
+    # ── WebhookProtocol ───────────────────────────────────
+
+    def verify_signature(self, body: bytes, signature: str, secret: str) -> bool:
+        return False
+
+    # ── ConfigSchemaProtocol ──────────────────────────────
+
+    def config_schema(self) -> dict:
+        return self._config.config_schema()
+
+    # ── AgentPromptProtocol ──────────────────────────────
+
+    @property
+    def channel_format_instructions(self) -> str | None:
+        return (
+            "You are responding via Google Chat. Messages support basic Markdown (bold, italic, "
+            "links, code blocks, lists). "
+            "Messages are limited to 4000 characters and will be automatically chunked if longer. "
+            "You can use reactions (emoji) and thread replies. "
+            "Be concise and avoid excessive formatting."
+        )
+
+
+googlechat_plugin = GoogleChatPlugin()
+ChannelPluginRegistry.register(googlechat_plugin)
\ No newline at end of file
diff --git a/backend/package/yuxi/channel/extensions/googlechat/access_policy.py b/backend/package/yuxi/channel/extensions/googlechat/access_policy.py
new file mode 100644
index 00000000..205b56ae
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/googlechat/access_policy.py
@@ -0,0 +1,145 @@
+from __future__ import annotations
+
+import re
+import logging
+
+logger = logging.getLogger(__name__)
+
+_GROUP_KEY_PATTERN = re.compile(r"^spaces/.+$")
+
+
+def apply_inbound_access_policy(
+    space_name: str,
+    space_type: str,
+    sender_name: str,
+    is_dm: bool,
+    account: dict,
+) -> dict:
+    """
+    Returns {"allowed": bool, "reason": str, "requires_pairing": bool}
+    """
+    if is_dm:
+        return _check_dm_policy(sender_name, account)
+    else:
+        return _check_group_policy(space_name, space_type, account)
+
+
+def _check_dm_policy(sender_name: str, account: dict) -> dict:
+    dm_policy = account.get("dm_policy", "pairing")
+    dm_enabled = account.get("dm_enabled", True)
+
+    if not dm_enabled:
+        return {"allowed": False, "reason": "dm_disabled"}
+
+    if dm_policy == "open":
+        return {"allowed": True, "reason": "open_dm"}
+
+    if dm_policy == "allowlist":
+        allow_from = account.get("dm_allow_from", [])
+        if "*" in allow_from:
+            return {"allowed": True, "reason": "wildcard_allow"}
+        if sender_name in allow_from:
+            return {"allowed": True, "reason": "in_allowlist"}
+        normalized = _normalize_sender_for_allowlist_check(sender_name, allow_from)
+        if normalized:
+            return {"allowed": True, "reason": "in_allowlist"}
+        return {"allowed": False, "reason": "not_in_allowlist"}
+
+    if dm_policy == "pairing":
+        allow_from = account.get("dm_allow_from", [])
+        resolved = account.get("resolved")
+        if resolved and hasattr(resolved, "dm_allow_from"):
+            allow_from = resolved.dm_allow_from
+        if sender_name in allow_from:
+            return {"allowed": True, "reason": "already_paired"}
+        normalized = _normalize_sender_for_allowlist_check(sender_name, allow_from)
+        if normalized:
+            return {"allowed": True, "reason": "already_paired"}
+        return {"allowed": False, "reason": "pairing_required", "requires_pairing": True}
+
+    return {"allowed": False, "reason": f"unknown_policy: {dm_policy}"}
+
+
+def _check_group_policy(space_name: str, space_type: str, account: dict) -> dict:
+    group_policy = account.get("group_policy", "allowlist")
+    groups = account.get("groups", {})
+
+    if group_policy == "disabled":
+        return {"allowed": False, "reason": "groups_disabled"}
+
+    if group_policy == "open":
+        return {"allowed": True, "reason": "open_group"}
+
+    if group_policy == "allowlist":
+        if "*" in groups:
+            return {"allowed": True, "reason": "wildcard_group"}
+        if space_name in groups:
+            entry = groups[space_name]
+            if entry.get("enabled") is False:
+                return {"allowed": False, "reason": "group_disabled"}
+            return {"allowed": True, "reason": "in_group_list"}
+        if not _GROUP_KEY_PATTERN.match(space_name):
+            return {"allowed": False, "reason": "deprecated_group_key"}
+        return {"allowed": False, "reason": "not_in_group_list"}
+
+    return {"allowed": False, "reason": f"unknown_policy: {group_policy}"}
+
+
+def _normalize_sender_for_allowlist_check(sender_name: str, allow_from: list[str]) -> str | None:
+    for entry in allow_from:
+        if entry.startswith("googlechat:"):
+            if entry[len("googlechat:"):] == sender_name:
+                return sender_name
+        if entry.startswith("users/"):
+            if entry == sender_name:
+                return sender_name
+    return None
+
+
+def is_control_command_message(text: str) -> bool:
+    text = text.strip()
+    return text.startswith("/config") or text.startswith("/status") or text.startswith("/admin")
+
+
+def should_require_command_authorization(
+    is_group: bool, text: str, command_authorized: bool
+) -> bool:
+    if not is_group:
+        return False
+    if not is_control_command_message(text):
+        return False
+    return not command_authorized
+
+
+def detect_mention_for_access(annotations: list[dict]) -> bool:
+    if not annotations:
+        return False
+    for ann in annotations:
+        if ann.get("type") == "USER_MENTION":
+            um = ann.get("userMention", {})
+            if um.get("user", {}).get("name") == "users/app":
+                return True
+    return False
+
+
+def resolve_group_require_mention(space_name: str, account: dict) -> bool:
+    groups = account.get("groups", {})
+    default_require = account.get("require_mention", True)
+
+    if "*" in groups:
+        entry = groups["*"]
+        if "requireMention" in entry:
+            return entry["requireMention"]
+
+    if space_name in groups:
+        entry = groups[space_name]
+        if "requireMention" in entry:
+            return entry["requireMention"]
+
+    return default_require
+
+
+def should_bypass_mention_for_text_commands(text: str, account: dict) -> bool:
+    if not account.get("allow_text_commands", False):
+        return False
+    return is_control_command_message(text)
diff --git a/backend/package/yuxi/channel/extensions/googlechat/accounts.py b/backend/package/yuxi/channel/extensions/googlechat/accounts.py
new file mode 100644
index 00000000..b7209ec3
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/googlechat/accounts.py
@@ -0,0 +1,30 @@
+from __future__ import annotations
+
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+def resolve_merged_config(cfg: dict, account_id: str) -> dict:
+    gc_config = cfg.get("channels", {}).get("googlechat", {}) if cfg else {}
+
+    defaults = gc_config.get("accounts", {}).get("default", {})
+    account_cfg = gc_config.get("accounts", {}).get(account_id, {})
+
+    merged = {**gc_config, **defaults, **account_cfg}
+
+    clear_base = {
+        "serviceAccount",
+        "serviceAccountFile",
+        "audienceType",
+        "audience",
+        "webhookPath",
+        "webhookUrl",
+        "botUser",
+        "name",
+    }
+    for field in clear_base:
+        if field not in account_cfg and field not in defaults:
+            merged.pop(field, None)
+
+    return merged
\ No newline at end of file
diff --git a/backend/package/yuxi/channel/extensions/googlechat/actions.py b/backend/package/yuxi/channel/extensions/googlechat/actions.py
new file mode 100644
index 00000000..c4b74e38
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/googlechat/actions.py
@@ -0,0 +1,42 @@
+from __future__ import annotations
+
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+async def execute_send_action(account, space_name: str, text: str, thread_name: str | None = None) -> dict:
+    from yuxi.channel.extensions.googlechat.api import send_message
+
+    result = await send_message(account, space_name, text, thread_name=thread_name)
+    return {"success": True, "action": "send", "result": result}
+
+
+async def execute_upload_file_action(
+    account, space_name: str, file_path: str, content_type: str | None = None
+) -> dict:
+    from yuxi.channel.extensions.googlechat.media import upload_media_file
+
+    result = await upload_media_file(account, space_name, file_path, content_type)
+    token = result.get("attachmentUploadToken")
+    if token:
+        from yuxi.channel.extensions.googlechat.api import send_message_with_attachments
+
+        msg = await send_message_with_attachments(
+            account, space_name, "", [{"attachmentUploadToken": token}]
+        )
+        return {"success": True, "action": "upload-file", "result": msg}
+    return {"success": True, "action": "upload-file", "result": result}
+
+
+async def execute_react_action(account, message_name: str, emoji: str, remove: bool = False) -> dict:
+    from yuxi.channel.extensions.googlechat.reactions import execute_react_action as _execute
+
+    return await _execute(account, message_name, emoji, remove)
+
+
+async def execute_reactions_action(account, message_name: str) -> dict:
+    from yuxi.channel.extensions.googlechat.reactions import get_reactions
+
+    reactions = await get_reactions(account, message_name)
+    return {"success": True, "action": "reactions", "result": reactions}
\ No newline at end of file
diff --git a/backend/package/yuxi/channel/extensions/googlechat/api.py b/backend/package/yuxi/channel/extensions/googlechat/api.py
new file mode 100644
index 00000000..bdb591a6
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/googlechat/api.py
@@ -0,0 +1,454 @@
+from __future__ import annotations
+
+import logging
+
+import httpx
+
+from yuxi.channel.extensions.googlechat.auth import get_access_token
+from yuxi.channel.extensions.googlechat.types import ResolvedGoogleChatAccount
+
+logger = logging.getLogger(__name__)
+
+CHAT_API_BASE = "https://chat.googleapis.com/v1"
+CHAT_UPLOAD_BASE = "https://chat.googleapis.com/upload/v1"
+CHAT_MEDIA_BASE = "https://chat.googleapis.com/v1/media"
+
+
+async def _get_http_for_account(
+    account: ResolvedGoogleChatAccount,
+) -> tuple[httpx.AsyncClient, str]:
+    token = await get_access_token(account)
+    if not token:
+        raise RuntimeError("Failed to obtain Google Chat access token")
+
+    client = httpx.AsyncClient(
+        base_url=CHAT_API_BASE,
+        timeout=httpx.Timeout(30.0),
+        headers={"Authorization": f"Bearer {token}"},
+    )
+    return client, token
+
+
+async def send_message(
+    account: ResolvedGoogleChatAccount,
+    space_name: str,
+    text: str,
+    *,
+    thread_name: str | None = None,
+    quoted_message_name: str | None = None,
+    private_message_viewer: str | None = None,
+    accessory_widgets: list[dict] | None = None,
+    client_message_id: str | None = None,
+    message_reply_option: str = "REPLY_MESSAGE_FALLBACK_TO_NEW_THREAD",
+) -> dict:
+    url = f"{CHAT_API_BASE}/{space_name}/messages"
+    token = await get_access_token(account)
+    if not token:
+        raise RuntimeError("Failed to obtain access token")
+
+    body: dict = {"text": text}
+    if thread_name:
+        body["messageReplyOption"] = message_reply_option
+        body["thread"] = {"name": thread_name}
+    if quoted_message_name:
+        body["quotedMessageMetadata"] = {"name": quoted_message_name}
+    if private_message_viewer:
+        body["privateMessageViewer"] = {"name": private_message_viewer}
+    if accessory_widgets:
+        body["accessoryWidgets"] = accessory_widgets
+
+    params: dict = {}
+    if client_message_id:
+        params["messageId"] = client_message_id
+
+    async with httpx.AsyncClient(timeout=httpx.Timeout(30.0)) as client:
+        resp = await client.post(
+            url,
+            json=body,
+            params=params,
+            headers={
+                "Authorization": f"Bearer {token}",
+                "Content-Type": "application/json",
+            },
+        )
+        resp.raise_for_status()
+        return resp.json()
+
+
+async def send_message_with_attachments(
+    account: ResolvedGoogleChatAccount,
+    space_name: str,
+    text: str,
+    attachments: list[dict],
+    *,
+    thread_name: str | None = None,
+    quoted_message_name: str | None = None,
+    private_message_viewer: str | None = None,
+    client_message_id: str | None = None,
+    message_reply_option: str = "REPLY_MESSAGE_FALLBACK_TO_NEW_THREAD",
+) -> dict:
+    url = f"{CHAT_API_BASE}/{space_name}/messages"
+    token = await get_access_token(account)
+    if not token:
+        raise RuntimeError("Failed to obtain access token")
+
+    body: dict = {"text": text, "attachment": attachments}
+    if thread_name:
+        body["messageReplyOption"] = message_reply_option
+        body["thread"] = {"name": thread_name}
+    if quoted_message_name:
+        body["quotedMessageMetadata"] = {"name": quoted_message_name}
+    if private_message_viewer:
+        body["privateMessageViewer"] = {"name": private_message_viewer}
+
+    params: dict = {}
+    if client_message_id:
+        params["messageId"] = client_message_id
+
+    async with httpx.AsyncClient(timeout=httpx.Timeout(30.0)) as client:
+        resp = await client.post(
+            url,
+            json=body,
+            params=params,
+            headers={
+                "Authorization": f"Bearer {token}",
+                "Content-Type": "application/json",
+            },
+        )
+        resp.raise_for_status()
+        return resp.json()
+
+
+async def send_card_message(
+    account: ResolvedGoogleChatAccount,
+    space_name: str,
+    cards_v2: list[dict],
+    *,
+    thread_name: str | None = None,
+    fallback_text: str = "",
+    message_reply_option: str = "REPLY_MESSAGE_FALLBACK_TO_NEW_THREAD",
+) -> dict:
+    url = f"{CHAT_API_BASE}/{space_name}/messages"
+    token = await get_access_token(account)
+    if not token:
+        raise RuntimeError("Failed to obtain access token")
+
+    body: dict = {
+        "cardsV2": cards_v2,
+        "fallbackText": fallback_text,
+    }
+    if thread_name:
+        body["messageReplyOption"] = message_reply_option
+        body["thread"] = {"name": thread_name}
+
+    async with httpx.AsyncClient(timeout=httpx.Timeout(30.0)) as client:
+        resp = await client.post(
+            url,
+            json=body,
+            headers={
+                "Authorization": f"Bearer {token}",
+                "Content-Type": "application/json",
+            },
+        )
+        resp.raise_for_status()
+        return resp.json()
+
+
+async def update_message(
+    account: ResolvedGoogleChatAccount,
+    message_name: str,
+    text: str,
+) -> dict:
+    url = f"{CHAT_API_BASE}/{message_name}?updateMask=text"
+    token = await get_access_token(account)
+    if not token:
+        raise RuntimeError("Failed to obtain access token")
+
+    body = {"text": text}
+
+    async with httpx.AsyncClient(timeout=httpx.Timeout(30.0)) as client:
+        resp = await client.patch(
+            url,
+            json=body,
+            headers={
+                "Authorization": f"Bearer {token}",
+                "Content-Type": "application/json",
+            },
+        )
+        resp.raise_for_status()
+        return resp.json()
+
+
+async def delete_message(account: ResolvedGoogleChatAccount, message_name: str) -> None:
+    url = f"{CHAT_API_BASE}/{message_name}"
+    token = await get_access_token(account)
+    if not token:
+        raise RuntimeError("Failed to obtain access token")
+
+    async with httpx.AsyncClient(timeout=httpx.Timeout(30.0)) as client:
+        resp = await client.delete(
+            url,
+            headers={"Authorization": f"Bearer {token}"},
+        )
+        resp.raise_for_status()
+
+
+async def get_message(
+    account: ResolvedGoogleChatAccount,
+    message_name: str,
+) -> dict | None:
+    url = f"{CHAT_API_BASE}/{message_name}"
+    token = await get_access_token(account)
+    if not token:
+        raise RuntimeError("Failed to obtain access token")
+
+    async with httpx.AsyncClient(timeout=httpx.Timeout(15.0)) as client:
+        resp = await client.get(
+            url,
+            headers={"Authorization": f"Bearer {token}"},
+        )
+        if resp.status_code == 404:
+            return None
+        resp.raise_for_status()
+        return resp.json()
+
+
+async def upload_attachment(
+    account: ResolvedGoogleChatAccount,
+    space_name: str,
+    filename: str,
+    content: bytes,
+    content_type: str,
+) -> dict:
+    url = f"{CHAT_UPLOAD_BASE}/{space_name}/attachments:upload"
+    token = await get_access_token(account)
+    if not token:
+        raise RuntimeError("Failed to obtain access token")
+
+    async with httpx.AsyncClient(timeout=httpx.Timeout(60.0)) as client:
+        resp = await client.post(
+            url,
+            data={"filename": filename},
+            files={"data": (filename, content, content_type)},
+            headers={"Authorization": f"Bearer {token}"},
+        )
+        resp.raise_for_status()
+        return resp.json()
+
+
+async def download_media_content(
+    account: ResolvedGoogleChatAccount,
+    resource_name: str,
+    max_bytes: int = 20 * 1024 * 1024,
+) -> bytes:
+    url = f"{CHAT_MEDIA_BASE}/{resource_name}"
+    token = await get_access_token(account)
+    if not token:
+        raise RuntimeError("Failed to obtain access token")
+
+    async with httpx.AsyncClient(timeout=httpx.Timeout(60.0)) as client:
+        resp = await client.get(
+            url,
+            headers={"Authorization": f"Bearer {token}"},
+            params={"alt": "media"},
+        )
+        resp.raise_for_status()
+        content = resp.content
+        if len(content) > max_bytes:
+            raise ValueError(f"Media size {len(content)} exceeds limit {max_bytes}")
+        return content
+
+
+async def create_reaction(
+    account: ResolvedGoogleChatAccount,
+    message_name: str,
+    emoji: str,
+    custom_emoji_url: str | None = None,
+) -> dict:
+    url = f"{CHAT_API_BASE}/{message_name}/reactions"
+    token = await get_access_token(account)
+    if not token:
+        raise RuntimeError("Failed to obtain access token")
+
+    body: dict = {"emoji": {"unicode": emoji}}
+    if custom_emoji_url:
+        body["emoji"]["customEmojiUrl"] = custom_emoji_url
+
+    async with httpx.AsyncClient(timeout=httpx.Timeout(30.0)) as client:
+        resp = await client.post(
+            url,
+            json=body,
+            headers={
+                "Authorization": f"Bearer {token}",
+                "Content-Type": "application/json",
+            },
+        )
+        resp.raise_for_status()
+        return resp.json()
+
+
+async def list_reactions(
+    account: ResolvedGoogleChatAccount,
+    message_name: str,
+) -> list[dict]:
+    url = f"{CHAT_API_BASE}/{message_name}/reactions"
+    token = await get_access_token(account)
+    if not token:
+        raise RuntimeError("Failed to obtain access token")
+
+    async with httpx.AsyncClient(timeout=httpx.Timeout(30.0)) as client:
+        resp = await client.get(
+            url,
+            headers={"Authorization": f"Bearer {token}"},
+        )
+        resp.raise_for_status()
+        data = resp.json()
+        return data.get("reactions", [])
+
+
+async def delete_reaction(
+    account: ResolvedGoogleChatAccount,
+    reaction_name: str,
+) -> None:
+    url = f"{CHAT_API_BASE}/{reaction_name}"
+    token = await get_access_token(account)
+    if not token:
+        raise RuntimeError("Failed to obtain access token")
+
+    async with httpx.AsyncClient(timeout=httpx.Timeout(30.0)) as client:
+        resp = await client.delete(
+            url,
+            headers={"Authorization": f"Bearer {token}"},
+        )
+        resp.raise_for_status()
+
+
+async def list_thread_messages(
+    account: ResolvedGoogleChatAccount,
+    space_name: str,
+    thread_name: str,
+    *,
+    page_size: int = 20,
+    order_by: str = "createTime desc",
+) -> list[dict]:
+    url = f"{CHAT_API_BASE}/{space_name}/messages"
+    token = await get_access_token(account)
+    if not token:
+        raise RuntimeError("Failed to obtain access token")
+
+    params = {
+        "thread.name": thread_name,
+        "pageSize": page_size,
+        "orderBy": order_by,
+    }
+
+    async with httpx.AsyncClient(timeout=httpx.Timeout(10.0)) as client:
+        resp = await client.get(
+            url,
+            params=params,
+            headers={"Authorization": f"Bearer {token}"},
+        )
+        resp.raise_for_status()
+        data = resp.json()
+        return data.get("messages", [])
+
+
+async def list_messages(
+    account: ResolvedGoogleChatAccount,
+    space_name: str,
+    *,
+    page_size: int = 50,
+    page_token: str | None = None,
+    filter_str: str | None = None,
+    order_by: str = "createTime desc",
+) -> dict:
+    url = f"{CHAT_API_BASE}/{space_name}/messages"
+    token = await get_access_token(account)
+    if not token:
+        raise RuntimeError("Failed to obtain access token")
+
+    params: dict = {"pageSize": page_size, "orderBy": order_by}
+    if page_token:
+        params["pageToken"] = page_token
+    if filter_str:
+        params["filter"] = filter_str
+
+    async with httpx.AsyncClient(timeout=httpx.Timeout(10.0)) as client:
+        resp = await client.get(
+            url, params=params,
+            headers={"Authorization": f"Bearer {token}"},
+        )
+        resp.raise_for_status()
+        return resp.json()
+
+
+async def list_spaces(
+    account: ResolvedGoogleChatAccount,
+    *,
+    page_size: int = 100,
+    filter_str: str | None = None,
+) -> list[dict]:
+    url = f"{CHAT_API_BASE}/spaces"
+    token = await get_access_token(account)
+    if not token:
+        raise RuntimeError("Failed to obtain access token")
+
+    params: dict = {"pageSize": page_size}
+    if filter_str:
+        params["filter"] = filter_str
+
+    async with httpx.AsyncClient(timeout=httpx.Timeout(15.0)) as client:
+        resp = await client.get(
+            url, params=params,
+            headers={"Authorization": f"Bearer {token}"},
+        )
+        resp.raise_for_status()
+        data = resp.json()
+        return data.get("spaces", [])
+
+
+async def find_direct_message(
+    account: ResolvedGoogleChatAccount,
+    user_name: str,
+) -> str | None:
+    url = f"{CHAT_API_BASE}/spaces:findDirectMessage"
+    token = await get_access_token(account)
+    if not token:
+        raise RuntimeError("Failed to obtain access token")
+
+    async with httpx.AsyncClient(timeout=httpx.Timeout(30.0)) as client:
+        resp = await client.get(
+            url,
+            params={"name": user_name},
+            headers={"Authorization": f"Bearer {token}"},
+        )
+        if resp.status_code == 404:
+            return None
+        resp.raise_for_status()
+        data = resp.json()
+        return data.get("name")
+
+
+async def list_members(
+    account: ResolvedGoogleChatAccount,
+    space_name: str,
+    *,
+    page_size: int = 100,
+    page_token: str | None = None,
+) -> dict:
+    url = f"{CHAT_API_BASE}/{space_name}/members"
+    token = await get_access_token(account)
+    if not token:
+        raise RuntimeError("Failed to obtain access token")
+
+    params: dict = {"pageSize": page_size}
+    if page_token:
+        params["pageToken"] = page_token
+
+    async with httpx.AsyncClient(timeout=httpx.Timeout(10.0)) as client:
+        resp = await client.get(
+            url, params=params,
+            headers={"Authorization": f"Bearer {token}"},
+        )
+        resp.raise_for_status()
+        return resp.json()
diff --git a/backend/package/yuxi/channel/extensions/googlechat/approval.py b/backend/package/yuxi/channel/extensions/googlechat/approval.py
new file mode 100644
index 00000000..932d1e59
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/googlechat/approval.py
@@ -0,0 +1,36 @@
+from __future__ import annotations
+
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+class GoogleChatApprovalAuth:
+    channel_label = "Google Chat"
+
+    def resolve_approvers(self, cfg: dict, account_id: str) -> list[str]:
+        gc_config = cfg.get("channels", {}).get("googlechat", {}) if cfg else {}
+        allow_from = gc_config.get("dm", {}).get("allowFrom", [])
+
+        if not allow_from:
+            account_cfg = gc_config.get("accounts", {}).get(account_id, {})
+            allow_from = account_cfg.get("dm", {}).get("allowFrom", [])
+        else:
+            account_cfg = gc_config.get("accounts", {}).get(account_id, {})
+            account_allow_from = account_cfg.get("dm", {}).get("allowFrom", [])
+            if account_allow_from:
+                allow_from = account_allow_from
+
+        return [a for a in (self._normalize_approver_id(entry) for entry in allow_from) if a]
+
+    def _normalize_approver_id(self, value) -> str | None:
+        if not value:
+            return None
+        for prefix in ("googlechat:",):
+            if value.startswith(prefix):
+                value = value[len(prefix):]
+        if value.startswith("users/") and "@" in value:
+            return None
+        if value.startswith("users/"):
+            return value.lower()
+        return f"users/{value.lower()}"
\ No newline at end of file
diff --git a/backend/package/yuxi/channel/extensions/googlechat/auth.py b/backend/package/yuxi/channel/extensions/googlechat/auth.py
new file mode 100644
index 00000000..3fd2f330
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/googlechat/auth.py
@@ -0,0 +1,195 @@
+from __future__ import annotations
+
+import json
+import logging
+import re
+import time
+import hashlib
+
+from google.oauth2 import service_account
+from google.oauth2 import id_token as google_id_token
+from google.auth.transport import requests as google_requests
+
+import httpx
+import jwt
+
+from yuxi.channel.extensions.googlechat.types import ResolvedGoogleChatAccount
+
+logger = logging.getLogger(__name__)
+
+CHAT_API_SCOPE = "https://www.googleapis.com/auth/chat.bot"
+CERTS_URL = "https://www.googleapis.com/service_accounts/v1/metadata/x509/chat@system.gserviceaccount.com"
+CERTS_TTL = 600
+
+_STANDARD_CHAT_ISSUER = "chat@system.gserviceaccount.com"
+_ADDON_ISSUER_PATTERN = re.compile(r"^service-\d+@gcp-sa-gsuiteaddons\.iam\.gserviceaccount\.com$")
+
+_auth_cache: dict[str, tuple[str, service_account.Credentials]] = {}
+_MAX_AUTH_CACHE = 32
+
+_cert_cache: tuple[float, dict[str, str]] | None = None
+_http_client: httpx.AsyncClient | None = None
+
+
+async def _get_http() -> httpx.AsyncClient:
+    global _http_client
+    if _http_client is None:
+        _http_client = httpx.AsyncClient(timeout=httpx.Timeout(30.0))
+    return _http_client
+
+
+def _normalize_private_key(key: str) -> str:
+    return key.replace("\\n", "\n")
+
+
+def _build_auth_key(sa_info: dict, sa_file: str | None) -> str:
+    if sa_file:
+        return f"file:{sa_file}"
+    if sa_info:
+        key_hash = hashlib.sha256(
+            json.dumps({"client_email": sa_info.get("client_email"), "project_id": sa_info.get("project_id")}).encode()
+        ).hexdigest()[:16]
+        return f"inline:{key_hash}"
+    return "none"
+
+
+def get_or_create_credentials(
+    account_id: str, sa_info: dict, sa_file: str | None
+) -> service_account.Credentials:
+    key = _build_auth_key(sa_info, sa_file)
+    cached = _auth_cache.get(account_id)
+    if cached and cached[0] == key:
+        return cached[1]
+
+    private_key = _normalize_private_key(sa_info.get("private_key", ""))
+
+    creds = service_account.Credentials.from_service_account_info(
+        {
+            **sa_info,
+            "private_key": private_key,
+        },
+        scopes=[CHAT_API_SCOPE],
+    )
+
+    if len(_auth_cache) >= _MAX_AUTH_CACHE:
+        oldest = next(iter(_auth_cache))
+        _auth_cache.pop(oldest)
+
+    _auth_cache[account_id] = (key, creds)
+    return creds
+
+
+async def get_access_token(account: ResolvedGoogleChatAccount) -> str | None:
+    if not account.service_account_info:
+        return None
+    creds = get_or_create_credentials(
+        account.account_id,
+        account.service_account_info,
+        account.service_account_file,
+    )
+    request = google_requests.Request()
+    creds.refresh(request)
+    return creds.token
+
+
+def decode_unverified_jwt_payload(token: str) -> dict:
+    import base64
+
+    parts = token.split(".")
+    if len(parts) < 2:
+        return {}
+    try:
+        padded = parts[1] + "=" * (4 - len(parts[1]) % 4)
+        decoded = base64.b64decode(padded.replace("-", "+").replace("_", "/"), validate=False)
+        return json.loads(decoded)
+    except Exception:
+        return {}
+
+
+async def _google_verify_id_token(token: str, audience: str) -> dict:
+    request = google_requests.Request()
+    return google_id_token.verify_oauth2_token(token, request, audience)
+
+
+async def _fetch_chat_certs() -> dict[str, str]:
+    global _cert_cache
+    now = time.monotonic()
+    if _cert_cache and now - _cert_cache[0] < CERTS_TTL:
+        return _cert_cache[1]
+
+    http = await _get_http()
+    resp = await http.get(CERTS_URL)
+    resp.raise_for_status()
+    certs = resp.json()
+    _cert_cache = (now, certs)
+    return certs
+
+
+async def _verify_signed_jwt_with_certs(token: str, audience: str, certs: dict[str, str]) -> dict:
+    header = jwt.get_unverified_header(token)
+    kid = header.get("kid", "")
+
+    cert = certs.get(kid)
+    if not cert:
+        raise jwt.InvalidTokenError(f"No matching cert for kid: {kid}")
+
+    return jwt.decode(token, cert, algorithms=["RS256"], audience=audience)
+
+
+async def verify_webhook_token(
+    token: str, audience_type: str, audience: str, app_principal: str | None
+) -> tuple[bool, dict, str | None]:
+    payload = decode_unverified_jwt_payload(token)
+    iss = payload.get("iss", "")
+
+    if iss == _STANDARD_CHAT_ISSUER:
+        try:
+            if audience_type == "app-url":
+                verified = await _google_verify_id_token(token, audience)
+            elif audience_type == "project-number":
+                certs = await _fetch_chat_certs()
+                verified = await _verify_signed_jwt_with_certs(token, audience, certs)
+            else:
+                return False, payload, f"unsupported audience_type: {audience_type}"
+            return True, verified, None
+        except Exception as e:
+            return False, payload, f"token verification failed: {e}"
+
+    if _ADDON_ISSUER_PATTERN.match(iss):
+        if app_principal and payload.get("sub") != app_principal:
+            return False, payload, "token sub does not match appPrincipal"
+        try:
+            certs = await _fetch_chat_certs()
+            verified = await _verify_signed_jwt_with_certs(token, audience, certs)
+            return True, verified, None
+        except Exception as e:
+            return False, payload, f"add-on token verification failed: {e}"
+
+    return False, payload, f"invalid issuer: {iss}"
+
+
+async def probe_account(account: ResolvedGoogleChatAccount) -> dict:
+    token = await get_access_token(account)
+    if not token:
+        return {"ok": False, "status": "no_token", "error": "Failed to obtain access token"}
+
+    http = await _get_http()
+    try:
+        resp = await http.get(
+            "https://chat.googleapis.com/v1/spaces?pageSize=1",
+            headers={"Authorization": f"Bearer {token}"},
+            timeout=10.0,
+        )
+        resp.raise_for_status()
+        return {"ok": True, "status": "connected"}
+    except httpx.HTTPStatusError as e:
+        return {"ok": False, "status": f"http_{e.response.status_code}", "error": str(e)}
+    except Exception as e:
+        return {"ok": False, "status": "error", "error": str(e)}
+
+
+async def close_auth_http() -> None:
+    global _http_client
+    if _http_client:
+        await _http_client.aclose()
+        _http_client = None
\ No newline at end of file
diff --git a/backend/package/yuxi/channel/extensions/googlechat/config.py b/backend/package/yuxi/channel/extensions/googlechat/config.py
new file mode 100644
index 00000000..b7a97e3b
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/googlechat/config.py
@@ -0,0 +1,313 @@
+from __future__ import annotations
+
+import json
+import logging
+import os
+import re
+from pathlib import Path
+
+from yuxi.channel.extensions.googlechat.types import ResolvedGoogleChatAccount
+
+logger = logging.getLogger(__name__)
+
+_AUDIENCE_VARIANTS: dict[str, set[str]] = {
+    "app-url": {"app-url", "app_url", "app"},
+    "project-number": {"project-number", "project_number", "project"},
+}
+
+_DEFAULT_WEBHOOK_PATH = "/api/webhook/googlechat"
+
+_REQUIRED_SERVICE_ACCOUNT_FIELDS = {"client_email", "private_key"}
+
+
+class GoogleChatConfigAdapter:
+    def list_account_ids(self, config: dict) -> list[str]:
+        gc_config = config.get("channels", {}).get("googlechat", {})
+        accounts = gc_config.get("accounts", {})
+        default_account = gc_config.get("defaultAccount", "default")
+        if not accounts:
+            return [default_account]
+        return list(accounts.keys())
+
+    async def resolve_account(self, account_id: str) -> dict:
+        from yuxi.channel.runtime.manager import gateway
+
+        config = gateway.global_config if gateway is not None else {}
+        gc_config = config.get("channels", {}).get("googlechat", {}) if config else {}
+
+        defaults = gc_config.get("accounts", {}).get("default", {})
+        account_cfg = gc_config.get("accounts", {}).get(account_id, {})
+        merged = {**gc_config, **defaults, **account_cfg}
+
+        credential_source, sa_info, sa_file = self._resolve_credential(merged, account_id, account_id == "default")
+
+        resolved = ResolvedGoogleChatAccount(
+            account_id=account_id,
+            name=merged.get("name", account_id),
+            credential_source=credential_source,
+            service_account_info=sa_info,
+            service_account_file=sa_file,
+            audience_type=self.normalize_audience_type(merged.get("audienceType")),
+            audience=merged.get("audience"),
+            webhook_path=merged.get("webhookPath", _DEFAULT_WEBHOOK_PATH),
+            webhook_url=merged.get("webhookUrl"),
+            bot_user=merged.get("botUser"),
+            app_principal=merged.get("appPrincipal"),
+            group_policy=merged.get("groupPolicy", "allowlist"),
+            require_mention=merged.get("requireMention", True),
+            allow_text_commands=merged.get("allowTextCommands", False),
+            reply_to_mode=merged.get("replyToMode", "off"),
+            text_chunk_limit=merged.get("textChunkLimit", 4000),
+            media_max_mb=merged.get("mediaMaxMb", 20),
+            typing_indicator=merged.get("typingIndicator", "message"),
+            allow_bots=merged.get("allowBots", False),
+            dangerously_allow_name_matching=merged.get("dangerouslyAllowNameMatching", False),
+            actions_reactions=merged.get("actions", {}).get("reactions", True),
+            dm_policy=merged.get("dm", {}).get("policy", "pairing"),
+            dm_enabled=merged.get("dm", {}).get("enabled", True),
+            dm_allow_from=merged.get("dm", {}).get("allowFrom", []),
+            groups=merged.get("groups", {}),
+            block_streaming_coalesce_min_chars=merged.get("blockStreamingCoalesce", {}).get("minChars", 1500),
+            block_streaming_coalesce_idle_ms=merged.get("blockStreamingCoalesce", {}).get("idleMs", 1000),
+            enabled=merged.get("enabled", True),
+            config=merged,
+        )
+
+        return {
+            "account_id": account_id,
+            "name": resolved.name,
+            "credential_source": resolved.credential_source,
+            "service_account_info": resolved.service_account_info,
+            "service_account_file": resolved.service_account_file,
+            "audience_type": resolved.audience_type,
+            "audience": resolved.audience,
+            "webhook_path": resolved.webhook_path,
+            "webhook_url": resolved.webhook_url,
+            "bot_user": resolved.bot_user,
+            "app_principal": resolved.app_principal,
+            "group_policy": resolved.group_policy,
+            "require_mention": resolved.require_mention,
+            "allow_text_commands": resolved.allow_text_commands,
+            "reply_to_mode": resolved.reply_to_mode,
+            "text_chunk_limit": resolved.text_chunk_limit,
+            "media_max_mb": resolved.media_max_mb,
+            "typing_indicator": resolved.typing_indicator,
+            "allow_bots": resolved.allow_bots,
+            "dangerously_allow_name_matching": resolved.dangerously_allow_name_matching,
+            "actions_reactions": resolved.actions_reactions,
+            "dm_policy": resolved.dm_policy,
+            "dm_enabled": resolved.dm_enabled,
+            "dm_allow_from": resolved.dm_allow_from,
+            "groups": resolved.groups,
+            "block_streaming_coalesce_min_chars": resolved.block_streaming_coalesce_min_chars,
+            "block_streaming_coalesce_idle_ms": resolved.block_streaming_coalesce_idle_ms,
+            "enabled": resolved.enabled,
+            "config": resolved.config,
+            "resolved": resolved,
+        }
+
+    def is_configured(self, account: dict) -> bool:
+        resolved = account.get("resolved")
+        if isinstance(resolved, ResolvedGoogleChatAccount):
+            return resolved.is_configured
+        return bool(account.get("credential_source") != "none")
+
+    def is_enabled(self, account: dict) -> bool:
+        return bool(account.get("enabled", True))
+
+    def disabled_reason(self, account: dict) -> str:
+        if not self.is_configured(account):
+            return "Service Account credentials are required"
+        return ""
+
+    async def resolve_allow_from(self, config: dict, account_id: str) -> list[str] | None:
+        account = await self.resolve_account(account_id)
+        return account.get("dm_allow_from", [])
+
+    def describe_account(self, account: dict) -> dict:
+        return {
+            "account_id": account.get("account_id", ""),
+            "name": account.get("name", ""),
+            "configured": self.is_configured(account),
+            "credential_source": account.get("credential_source", "none"),
+        }
+
+    def config_schema(self) -> dict:
+        return {
+            "type": "object",
+            "properties": {
+                "enabled": {"type": "boolean", "default": True},
+                "name": {"type": "string"},
+                "serviceAccount": {
+                    "type": "string",
+                    "description": "Google Cloud Service Account JSON (inline string)",
+                },
+                "serviceAccountFile": {
+                    "type": "string",
+                    "description": "Path to Service Account JSON file",
+                },
+                "audienceType": {
+                    "type": "string",
+                    "enum": ["app-url", "project-number"],
+                },
+                "audience": {"type": "string"},
+                "webhookPath": {"type": "string", "default": "/api/webhook/googlechat"},
+                "webhookUrl": {"type": "string"},
+                "botUser": {"type": "string"},
+                "appPrincipal": {"type": "string"},
+                "groupPolicy": {
+                    "type": "string",
+                    "enum": ["open", "allowlist", "disabled"],
+                    "default": "allowlist",
+                },
+                "requireMention": {"type": "boolean", "default": True},
+                "allowTextCommands": {"type": "boolean", "default": False},
+                "replyToMode": {
+                    "type": "string",
+                    "enum": ["off", "first", "all", "batched"],
+                    "default": "off",
+                },
+                "textChunkLimit": {"type": "integer", "default": 4000},
+                "mediaMaxMb": {"type": "integer", "default": 20},
+                "typingIndicator": {
+                    "type": "string",
+                    "enum": ["message", "reaction", "off"],
+                    "default": "message",
+                },
+                "allowBots": {"type": "boolean", "default": False},
+                "dangerouslyAllowNameMatching": {"type": "boolean", "default": False},
+                "dm": {
+                    "type": "object",
+                    "properties": {
+                        "policy": {
+                            "type": "string",
+                            "enum": ["open", "pairing", "allowlist"],
+                            "default": "pairing",
+                        },
+                        "allowFrom": {"type": "array", "items": {"type": "string"}},
+                        "enabled": {"type": "boolean", "default": True},
+                    },
+                },
+                "groups": {
+                    "type": "object",
+                    "additionalProperties": {
+                        "type": "object",
+                        "properties": {
+                            "requireMention": {"type": "boolean"},
+                            "enabled": {"type": "boolean"},
+                            "users": {"type": "array", "items": {"type": "string"}},
+                            "systemPrompt": {"type": "string"},
+                        },
+                    },
+                },
+                "actions": {
+                    "type": "object",
+                    "properties": {
+                        "reactions": {"type": "boolean", "default": True},
+                    },
+                },
+                "accounts": {
+                    "type": "object",
+                    "additionalProperties": {
+                        "type": "object",
+                        "properties": {
+                            "name": {"type": "string"},
+                            "serviceAccount": {"type": "string"},
+                            "serviceAccountFile": {"type": "string"},
+                            "audienceType": {"type": "string", "enum": ["app-url", "project-number"]},
+                            "audience": {"type": "string"},
+                            "webhookPath": {"type": "string"},
+                            "webhookUrl": {"type": "string"},
+                            "botUser": {"type": "string"},
+                            "appPrincipal": {"type": "string"},
+                            "dm": {
+                                "type": "object",
+                                "properties": {
+                                    "policy": {"type": "string", "enum": ["open", "pairing", "allowlist"]},
+                                    "allowFrom": {"type": "array", "items": {"type": "string"}},
+                                    "enabled": {"type": "boolean"},
+                                },
+                            },
+                        },
+                    },
+                },
+                "defaultAccount": {"type": "string", "default": "default"},
+            },
+        }
+
+    @staticmethod
+    def normalize_audience_type(value: str | None) -> str | None:
+        if not value:
+            return None
+        normalized = value.strip().lower()
+        for canonical, variants in _AUDIENCE_VARIANTS.items():
+            if normalized in variants:
+                return canonical
+        return None
+
+    @staticmethod
+    def _resolve_credential(
+        merged: dict, account_id: str, is_default: bool
+    ) -> tuple[str, dict | None, str | None]:
+        sa_file = merged.get("serviceAccountFile")
+        if sa_file:
+            try:
+                with open(sa_file, encoding="utf-8") as f:
+                    sa_info = json.load(f)
+                return "file", sa_info, sa_file
+            except Exception:
+                logger.warning("Failed to read serviceAccountFile: %s", sa_file)
+
+        sa_inline = merged.get("serviceAccount")
+        if sa_inline:
+            try:
+                sa_info = json.loads(sa_inline) if isinstance(sa_inline, str) else sa_inline
+                return "inline", sa_info, None
+            except Exception:
+                logger.warning("Failed to parse inline serviceAccount")
+
+        if is_default:
+            sa_env = os.getenv("GOOGLE_CHAT_SERVICE_ACCOUNT")
+            if sa_env:
+                try:
+                    sa_info = json.loads(sa_env)
+                    return "env", sa_info, None
+                except Exception:
+                    logger.warning("Failed to parse GOOGLE_CHAT_SERVICE_ACCOUNT env var")
+
+            sa_file_env = os.getenv("GOOGLE_CHAT_SERVICE_ACCOUNT_FILE")
+            if sa_file_env:
+                try:
+                    with open(sa_file_env, encoding="utf-8") as f:
+                        sa_info = json.load(f)
+                    return "env_file", sa_info, sa_file_env
+                except Exception:
+                    logger.warning("Failed to read GOOGLE_CHAT_SERVICE_ACCOUNT_FILE: %s", sa_file_env)
+
+        return "none", None, None
+
+    @staticmethod
+    def validate_service_account(sa_info: dict) -> list[str]:
+        errors = []
+        if sa_info.get("type") and sa_info["type"] != "service_account":
+            errors.append(f"invalid type: {sa_info['type']}, expected service_account")
+        if not sa_info.get("client_email"):
+            errors.append("missing client_email")
+        if not sa_info.get("private_key"):
+            errors.append("missing private_key")
+        universe = sa_info.get("universe_domain")
+        if universe and universe != "googleapis.com":
+            errors.append(f"unexpected universe_domain: {universe}, expected googleapis.com")
+        auth_uri = sa_info.get("auth_uri")
+        if auth_uri and auth_uri != "https://accounts.google.com/o/oauth2/auth":
+            errors.append(f"unexpected auth_uri: {auth_uri}")
+        token_uri = sa_info.get("token_uri")
+        if token_uri and token_uri != "https://oauth2.googleapis.com/token":
+            errors.append(f"unexpected token_uri: {token_uri}")
+        cert_url = sa_info.get("auth_provider_x509_cert_url")
+        if cert_url and cert_url != "https://www.googleapis.com/oauth2/v1/certs":
+            errors.append(f"unexpected auth_provider_x509_cert_url: {cert_url}")
+        client_cert = sa_info.get("client_x509_cert_url")
+        if client_cert and not client_cert.startswith("https://www.googleapis.com/robot/v1/metadata/x509/"):
+            errors.append(f"unexpected client_x509_cert_url: {client_cert}")
+        return errors
\ No newline at end of file
diff --git a/backend/package/yuxi/channel/extensions/googlechat/dedupe.py b/backend/package/yuxi/channel/extensions/googlechat/dedupe.py
new file mode 100644
index 00000000..6bb11460
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/googlechat/dedupe.py
@@ -0,0 +1,39 @@
+from __future__ import annotations
+
+import time
+from collections import OrderedDict
+
+
+class GoogleChatDedupeStore:
+    def __init__(self, ttl_seconds: int = 300, max_entries: int = 10000):
+        self._store: OrderedDict[str, float] = OrderedDict()
+        self._ttl_seconds = ttl_seconds
+        self._max_entries = max_entries
+
+    def is_duplicate(self, key: str) -> bool:
+        self._evict_expired()
+        return key in self._store
+
+    def mark_seen(self, key: str) -> None:
+        self._evict_expired()
+        self._store[key] = time.monotonic()
+        if len(self._store) > self._max_entries:
+            self._store.popitem(last=False)
+
+    def reset(self) -> None:
+        self._store.clear()
+
+    @property
+    def ttl_seconds(self) -> int:
+        return self._ttl_seconds
+
+    @property
+    def max_entries(self) -> int:
+        return self._max_entries
+
+    def _evict_expired(self) -> None:
+        now = time.monotonic()
+        cutoff = now - self._ttl_seconds
+        stale = [k for k, v in self._store.items() if v < cutoff]
+        for k in stale:
+            del self._store[k]
\ No newline at end of file
diff --git a/backend/package/yuxi/channel/extensions/googlechat/formatting.py b/backend/package/yuxi/channel/extensions/googlechat/formatting.py
new file mode 100644
index 00000000..2ea8ed06
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/googlechat/formatting.py
@@ -0,0 +1,96 @@
+from __future__ import annotations
+
+import re
+
+_CONTROL_CHAR_PATTERN = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]")
+_MULTI_NEWLINE_PATTERN = re.compile(r"\n{3,}")
+
+
+def sanitize_text(text: str) -> str:
+    cleaned = _CONTROL_CHAR_PATTERN.sub("", text)
+    cleaned = _MULTI_NEWLINE_PATTERN.sub("\n\n", cleaned)
+    return cleaned.strip()
+
+
+def chunk_markdown_text(text: str, limit: int = 4000) -> list[str]:
+    if len(text) <= limit:
+        return [text]
+
+    chunks = []
+    lines = text.split("\n")
+    current = ""
+
+    for line in lines:
+        if len(current) + len(line) + 1 > limit:
+            if current:
+                chunks.append(current.strip())
+                current = ""
+            if len(line) > limit:
+                while len(line) > limit:
+                    split_at = _find_code_block_split(line, limit)
+                    chunks.append(line[:split_at].strip())
+                    line = line[split_at:].strip()
+                if line:
+                    current = line
+            else:
+                current = line
+        else:
+            if current:
+                current += "\n" + line
+            else:
+                current = line
+
+    if current.strip():
+        chunks.append(current.strip())
+
+    return chunks
+
+
+def _find_code_block_split(line: str, limit: int) -> int:
+    in_block = False
+    chars = 0
+    for i, ch in enumerate(line):
+        if ch == "`":
+            if line[i:i+3] == "```":
+                in_block = not in_block
+        if not in_block and chars >= limit:
+            return i
+        chars += 1
+    return limit
+
+
+def format_conversation_label(user_name: str, space_name: str) -> str:
+    if space_name:
+        return f"googlechat::{space_name}"
+    return f"googlechat::{user_name}"
+
+
+def build_typing_indicator_text(bot_name: str) -> str:
+    return f"_{bot_name} is typing..._"
+
+
+def resolve_bot_display_name(account: dict) -> str:
+    resolved = account.get("resolved")
+    if resolved and resolved.name:
+        return resolved.name
+    return account.get("name", "Bot")
+
+
+def format_thread_context(recent_messages: list[dict]) -> str:
+    if not recent_messages:
+        return ""
+
+    lines = [
+        "[Thread conversation - previous replies. You are participating in this thread.]",
+        "",
+        "[Previous messages]",
+    ]
+    for msg in recent_messages[-10:]:
+        author = msg.get("author", "Unknown")
+        content = msg.get("content", "")
+        if content:
+            lines.append(f"{author}: {content}")
+
+    lines.append("")
+    lines.append("[Current message]")
+    return "\n".join(lines)
\ No newline at end of file
diff --git a/backend/package/yuxi/channel/extensions/googlechat/gateway.py b/backend/package/yuxi/channel/extensions/googlechat/gateway.py
new file mode 100644
index 00000000..22faa637
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/googlechat/gateway.py
@@ -0,0 +1,109 @@
+from __future__ import annotations
+
+import asyncio
+import logging
+import time
+
+from yuxi.channel.extensions.googlechat.auth import probe_account
+from yuxi.channel.extensions.googlechat.config import GoogleChatConfigAdapter
+from yuxi.channel.extensions.googlechat.types import ResolvedGoogleChatAccount
+from yuxi.channel.extensions.googlechat.webhook import GoogleChatWebhookHandler
+from yuxi.channel.gateway.routes import webhook_registry
+
+logger = logging.getLogger(__name__)
+
+_STATUS_OK = "ok"
+_STATUS_ERROR = "error"
+
+
+class GoogleChatGatewayAdapter:
+    def __init__(self):
+        self._config_adapter = GoogleChatConfigAdapter()
+        self._webhook_handler = GoogleChatWebhookHandler()
+        self._running: dict[str, bool] = {}
+        self._start_times: dict[str, float] = {}
+        self._status_snapshots: dict[str, dict] = {}
+
+    async def start(self, ctx) -> object:
+        account_dict = getattr(ctx, "account", {}) or {}
+        account = account_dict.get("resolved")
+        if not isinstance(account, ResolvedGoogleChatAccount):
+            account_id = account_dict.get("account_id", "default")
+            resolved_dict = await self._config_adapter.resolve_account(account_id)
+            account = resolved_dict.get("resolved")
+
+        if not account or not account.is_configured:
+            logger.warning("Google Chat account not configured: %s", account.account_id if account else "unknown")
+            return asyncio.Queue()
+
+        account_id = account.account_id
+        webhook_path = account.webhook_path
+
+        self._webhook_handler.register_target(webhook_path, account)
+
+        webhook_registry.register(
+            "googlechat",
+            self._webhook_handler.handle_webhook,
+            guard_config=None,
+        )
+
+        self._running[account_id] = True
+        self._start_times[account_id] = time.time()
+        self._status_snapshots[account_id] = {
+            "running": True,
+            "webhook_path": webhook_path,
+            "audience_type": account.audience_type,
+            "audience": account.audience,
+            "credential_source": account.credential_source,
+            "start_time": time.time(),
+        }
+
+        self._warn_app_principal_misconfiguration(account)
+
+        logger.info("Google Chat gateway started for account: %s, webhook: %s", account_id, webhook_path)
+
+        queue = getattr(ctx, "queue", asyncio.Queue())
+        return queue
+
+    async def stop(self, ctx) -> None:
+        account_dict = getattr(ctx, "account", {}) or {}
+        account_id = account_dict.get("account_id", "default")
+
+        account = account_dict.get("resolved")
+        if isinstance(account, ResolvedGoogleChatAccount):
+            self._webhook_handler.unregister_target(account.webhook_path)
+
+        self._running.pop(account_id, None)
+        self._status_snapshots[account_id] = {
+            "running": False,
+            "stop_time": time.time(),
+        }
+
+        logger.info("Google Chat gateway stopped for account: %s", account_id)
+
+    def get_status(self, account_id: str) -> dict:
+        return self._status_snapshots.get(account_id, {})
+
+    async def probe(self, account_dict: dict) -> dict:
+        account = account_dict.get("resolved")
+        if not isinstance(account, ResolvedGoogleChatAccount):
+            return {"ok": False, "status": "no_account"}
+
+        return await probe_account(account)
+
+    @staticmethod
+    def _warn_app_principal_misconfiguration(account: ResolvedGoogleChatAccount) -> None:
+        if account.audience_type != "app-url":
+            return
+        if not account.app_principal:
+            logger.error(
+                'appPrincipal is missing for audienceType "app-url"; '
+                "add-on token verification will fail. "
+                "Set appPrincipal to the numeric OAuth 2.0 client ID (uniqueId, 21 digits), not an email."
+            )
+        elif "@" in account.app_principal:
+            logger.error(
+                'appPrincipal "%s" looks like an email address. '
+                "Set appPrincipal to the numeric OAuth 2.0 client ID (uniqueId, 21 digits), not an email.",
+                account.app_principal,
+            )
\ No newline at end of file
diff --git a/backend/package/yuxi/channel/extensions/googlechat/media.py b/backend/package/yuxi/channel/extensions/googlechat/media.py
new file mode 100644
index 00000000..b34d022f
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/googlechat/media.py
@@ -0,0 +1,122 @@
+from __future__ import annotations
+
+import logging
+import os
+import tempfile
+
+import httpx
+
+from yuxi.channel.extensions.googlechat.api import (
+    download_media_content,
+    upload_attachment,
+)
+from yuxi.channel.extensions.googlechat.types import ResolvedGoogleChatAccount
+
+logger = logging.getLogger(__name__)
+
+_DEFAULT_MEDIA_MAX_MB = 20
+
+
+async def download_attachment(
+    account: ResolvedGoogleChatAccount,
+    resource_name: str,
+    max_mb: int = _DEFAULT_MEDIA_MAX_MB,
+) -> tuple[bytes, str, str]:
+    max_bytes = max_mb * 1024 * 1024
+    content = await download_media_content(account, resource_name, max_bytes)
+
+    content_type = "application/octet-stream"
+    try:
+        import magic
+        content_type = magic.Magic(mime=True).from_buffer(content)
+    except ImportError:
+        if content[:4] == b"\xff\xd8\xff":
+            content_type = "image/jpeg"
+        elif content[:4] == b"\x89PNG":
+            content_type = "image/png"
+        elif content[:6] in (b"GIF87a", b"GIF89a"):
+            content_type = "image/gif"
+
+    ext = _mime_to_ext(content_type)
+    filename = f"attachment{ext}"
+
+    return content, filename, content_type
+
+
+async def upload_media_file(
+    account: ResolvedGoogleChatAccount,
+    space_name: str,
+    file_path: str,
+    content_type: str | None = None,
+) -> dict:
+    if not content_type:
+        try:
+            import magic
+            content_type = magic.Magic(mime=True).from_file(file_path)
+        except ImportError:
+            content_type = "application/octet-stream"
+
+    filename = os.path.basename(file_path)
+    with open(file_path, "rb") as f:
+        content = f.read()
+
+    return await upload_attachment(account, space_name, filename, content, content_type)
+
+
+async def fetch_remote_media(url: str, max_mb: int = _DEFAULT_MEDIA_MAX_MB) -> tuple[bytes, str, str]:
+    max_bytes = max_mb * 1024 * 1024
+
+    async with httpx.AsyncClient(timeout=httpx.Timeout(60.0)) as client:
+        resp = await client.get(url)
+        resp.raise_for_status()
+        content = resp.content
+
+    if len(content) > max_bytes:
+        raise ValueError(f"Media size {len(content)} exceeds limit {max_bytes}")
+
+    content_type = resp.headers.get("content-type", "application/octet-stream")
+    ext = _mime_to_ext(content_type)
+    filename = f"downloaded{ext}"
+
+    return content, filename, content_type
+
+
+def _mime_to_ext(mime: str) -> str:
+    ext_map = {
+        "image/jpeg": ".jpg",
+        "image/png": ".png",
+        "image/gif": ".gif",
+        "image/webp": ".webp",
+        "image/svg+xml": ".svg",
+        "video/mp4": ".mp4",
+        "video/webm": ".webm",
+        "audio/mpeg": ".mp3",
+        "audio/wav": ".wav",
+        "application/pdf": ".pdf",
+        "text/plain": ".txt",
+        "text/html": ".html",
+    }
+    return ext_map.get(mime, ".bin")
+
+
+def build_attachment_placeholder(attachments: list[dict]) -> str:
+    if not attachments:
+        return ""
+
+    types: dict[str, int] = {}
+    for att in attachments:
+        ct = att.get("contentType", "unknown")
+        main_type = ct.split("/")[0]
+        types[main_type] = types.get(main_type, 0) + 1
+
+    parts = []
+    for t, count in types.items():
+        parts.append(f"{count} {t}")
+
+    if len(attachments) == 1:
+        ct = attachments[0].get("contentType", "")
+        main = ct.split("/")[0]
+        return f"<media:{main}>"
+
+    joined = " + ".join(parts)
+    return f"[{joined} attached]"
\ No newline at end of file
diff --git a/backend/package/yuxi/channel/extensions/googlechat/monitor.py b/backend/package/yuxi/channel/extensions/googlechat/monitor.py
new file mode 100644
index 00000000..71f5817d
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/googlechat/monitor.py
@@ -0,0 +1,169 @@
+from __future__ import annotations
+
+import logging
+from datetime import datetime, timezone
+
+from yuxi.channel.extensions.googlechat.formatting import sanitize_text
+from yuxi.channel.extensions.googlechat.media import build_attachment_placeholder
+from yuxi.channel.extensions.googlechat.types import GoogleChatEvent, ResolvedGoogleChatAccount
+from yuxi.channel.message.models import (
+    GroupContext,
+    MessageType,
+    PeerInfo,
+    UnifiedMessage,
+)
+from yuxi.channel.routing.models import PeerKind
+
+logger = logging.getLogger(__name__)
+
+
+class GoogleChatMonitor:
+    def __init__(self):
+        self._last_event_time: float = 0
+
+    def event_to_unified_message(
+        self,
+        raw_payload: dict,
+        event: GoogleChatEvent,
+        text: str | None = None,
+        account: ResolvedGoogleChatAccount | None = None,
+    ) -> UnifiedMessage | None:
+        if not event.message:
+            return None
+
+        msg = event.message
+        sender = msg.sender
+        space_name = event.effective_space_name
+        space_type = event.space_type
+
+        safe_text = text if text is not None else sanitize_text(msg.argument_text or msg.text)
+        if not safe_text:
+            return None
+
+        is_dm = space_type == "DM" or event.effective_space.single_user_bot_dm if event.effective_space else False
+        is_group = space_type in ("GROUP_CHAT", "SPACE") and not is_dm
+
+        sender_kind = PeerKind.DIRECT if is_dm else PeerKind.GROUP
+
+        sender_info = PeerInfo(
+            kind=sender_kind,
+            id=sender.name or "unknown",
+            display_name=sender.display_name,
+            username=sender.email,
+            is_bot=sender.type == "BOT",
+        )
+
+        group = None
+        if is_group:
+            group = GroupContext(
+                id=space_name,
+                name=event.effective_space.display_name if event.effective_space else None,
+                kind="group",
+            )
+
+        thread_id = None
+        if msg.thread:
+            thread_id = msg.thread.get("name")
+
+        timestamp = None
+        if event.event_time:
+            try:
+                from datetime import datetime as dt
+
+                timestamp = dt.fromisoformat(event.event_time.replace("Z", "+00:00"))
+            except (ValueError, TypeError):
+                pass
+
+        media_urls = []
+        media_types = []
+        for att in msg.attachment:
+            dr = att.get("driveDataRef") or att.get("attachmentDataRef")
+            if dr:
+                media_urls.append(dr.get("resourceName", ""))
+                media_types.append(att.get("contentType", ""))
+
+        if msg.attachment:
+            placeholder = build_attachment_placeholder(msg.attachment)
+            if placeholder:
+                safe_text = f"{placeholder}\n{safe_text}"
+
+        was_mentioned = _detect_mention(msg.annotations)
+        slash_cmd = _detect_slash_command(msg.annotations)
+
+        metadata = {
+            "ChatType": "direct" if is_dm else "group",
+            "WasMentioned": was_mentioned,
+            "CommandAuthorized": True,
+            "space_type": space_type,
+            "space_name": space_name,
+        }
+        if slash_cmd:
+            metadata["SlashCommand"] = slash_cmd
+        if is_group and event.effective_space:
+            metadata["GroupSpace"] = event.effective_space.display_name
+
+        return UnifiedMessage(
+            msg_id=msg.name or f"gc:{int(datetime.now(tz=timezone.utc).timestamp())}",
+            channel_type="googlechat",
+            account_id=account.account_id if account else "default",
+            content=safe_text,
+            sender=sender_info,
+            message_type=MessageType.TEXT,
+            media_urls=media_urls,
+            media_types=media_types,
+            group=group,
+            timestamp=timestamp,
+            raw_payload=raw_payload,
+            message_thread_id=thread_id,
+            metadata=metadata,
+            was_mentioned=was_mentioned,
+        )
+
+    async def inject_thread_context(
+        self,
+        account: ResolvedGoogleChatAccount,
+        space_name: str,
+        thread_name: str,
+    ) -> str | None:
+        try:
+            from yuxi.channel.extensions.googlechat.api import list_thread_messages
+            from yuxi.channel.extensions.googlechat.formatting import format_thread_context
+
+            messages = await list_thread_messages(account, space_name, thread_name, page_size=20)
+            if not messages:
+                return None
+
+            history = []
+            for msg in messages:
+                sender = msg.get("sender", {})
+                history.append({
+                    "author": sender.get("displayName", sender.get("name", "Unknown")),
+                    "content": msg.get("text", msg.get("argumentText", "")),
+                })
+
+            return format_thread_context(history)
+        except Exception:
+            logger.debug("Failed to fetch thread history for %s in %s", thread_name, space_name)
+            return None
+
+
+def _detect_mention(annotations: list[dict]) -> bool:
+    if not annotations:
+        return False
+    for ann in annotations:
+        if ann.get("type") == "USER_MENTION":
+            um = ann.get("userMention", {})
+            mention_user = um.get("user", {})
+            if mention_user.get("name") == "users/app":
+                return True
+    return False
+
+
+def _detect_slash_command(annotations: list[dict]) -> str | None:
+    if not annotations:
+        return None
+    for ann in annotations:
+        if ann.get("type") == "SLASH_COMMAND":
+            sc = ann.get("slashCommand", {})
+            return sc.get("commandId")
+    return None
diff --git a/backend/package/yuxi/channel/extensions/googlechat/outbound.py b/backend/package/yuxi/channel/extensions/googlechat/outbound.py
new file mode 100644
index 00000000..ee6186a7
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/googlechat/outbound.py
@@ -0,0 +1,272 @@
+from __future__ import annotations
+
+import logging
+
+from yuxi.channel.extensions.googlechat.api import (
+    delete_message,
+    send_message,
+    send_message_with_attachments,
+    update_message,
+)
+from yuxi.channel.extensions.googlechat.formatting import chunk_markdown_text, sanitize_text
+from yuxi.channel.extensions.googlechat.types import ResolvedGoogleChatAccount
+from yuxi.channel.protocols import (
+    OutboundDeliveryCapabilities,
+    OutboundDeliveryMode,
+    OutboundPresentationCapabilities,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class GoogleChatOutboundAdapter:
+    delivery_mode = OutboundDeliveryMode.DIRECT
+    chunker_mode = "markdown"
+    text_chunk_limit: int = 4000
+    poll_max_options = None
+    supports_poll_duration_seconds = False
+    supports_anonymous_polls = False
+    extract_markdown_images = False
+    presentation_capabilities = OutboundPresentationCapabilities(
+        supported=False,
+    )
+    delivery_capabilities = OutboundDeliveryCapabilities(
+        durable_final_text=True,
+        durable_final_media=False,
+        durable_final_thread=True,
+    )
+
+    async def _get_account(self, account_id: str | None) -> ResolvedGoogleChatAccount | None:
+        if account_id:
+            from yuxi.channel.plugins.registry import ChannelPluginRegistry
+
+            plugin = ChannelPluginRegistry.get("googlechat")
+            if plugin and hasattr(plugin, "_config"):
+                account_dict = await plugin._config.resolve_account(account_id)
+                return account_dict.get("resolved")
+        return None
+
+    async def send_text(
+        self,
+        target_id: str,
+        content: str,
+        *,
+        reply_to_id: str | None = None,
+        thread_id: str | None = None,
+        account_id: str | None = None,
+    ) -> None:
+        account = await self._get_account(account_id)
+        if not account:
+            raise RuntimeError(f"Google Chat account not found: {account_id}")
+
+        safe = sanitize_text(content)
+        if not safe:
+            return
+
+        reply_mode = getattr(account, "reply_to_mode", "off")
+        effective_thread = thread_id if reply_mode != "off" else None
+
+        chunks = chunk_markdown_text(safe, self.text_chunk_limit)
+        for i, chunk in enumerate(chunks):
+            chunk_thread = effective_thread if reply_mode == "all" or (reply_mode == "first" and i == 0) else None
+            quoted = reply_to_id if i == 0 else None
+            await send_message(
+                account,
+                target_id,
+                chunk,
+                thread_name=chunk_thread,
+                quoted_message_name=quoted,
+            )
+
+    async def send_media(
+        self,
+        target_id: str,
+        media_url: str,
+        media_type: str,
+        reply_to_id: str | None = None,
+        thread_id: str | None = None,
+        account_id: str | None = None,
+    ) -> None:
+        import httpx
+
+        account = await self._get_account(account_id)
+        if not account:
+            raise RuntimeError(f"Google Chat account not found: {account_id}")
+
+        async with httpx.AsyncClient(timeout=httpx.Timeout(60.0)) as client:
+            resp = await client.get(media_url)
+            resp.raise_for_status()
+            media_data = resp.content
+
+        from yuxi.channel.extensions.googlechat.api import upload_attachment
+
+        upload_result = await upload_attachment(
+            account, target_id, "attachment", media_data, media_type
+        )
+        attachment_token = upload_result.get("attachmentUploadToken")
+
+        if attachment_token:
+            attachment = {"attachmentUploadToken": attachment_token}
+            await send_message_with_attachments(
+                account, target_id, "", [attachment], thread_name=thread_id,
+                quoted_message_name=reply_to_id,
+            )
+
+    async def edit_message(
+        self,
+        target_id: str,
+        message_id: str,
+        content: str,
+        *,
+        thread_id: str | None = None,
+        account_id: str | None = None,
+    ) -> str | None:
+        account = await self._get_account(account_id)
+        if not account:
+            return None
+
+        safe = sanitize_text(content)
+        await update_message(account, message_id, safe)
+        return message_id
+
+    async def send_card(
+        self,
+        target_id: str,
+        card_content: dict,
+        *,
+        reply_to_id: str | None = None,
+        thread_id: str | None = None,
+        account_id: str | None = None,
+    ) -> str | None:
+        account = await self._get_account(account_id)
+        if not account:
+            return None
+
+        cards_v2 = card_content.get("cardsV2", card_content.get("cards", []))
+        if isinstance(cards_v2, dict):
+            cards_v2 = [cards_v2]
+
+        if not cards_v2:
+            return None
+
+        fallback = card_content.get("fallbackText", card_content.get("text", ""))
+
+        from yuxi.channel.extensions.googlechat.api import send_card_message
+
+        result = await send_card_message(
+            account, target_id, cards_v2,
+            thread_name=thread_id, fallback_text=fallback,
+        )
+        return result.get("name")
+
+    async def edit_card(
+        self,
+        target_id: str,
+        message_id: str,
+        card_content: dict,
+        *,
+        thread_id: str | None = None,
+        account_id: str | None = None,
+    ) -> str | None:
+        account = await self._get_account(account_id)
+        if not account:
+            return None
+
+        cards_v2 = card_content.get("cardsV2", card_content.get("cards", []))
+        if isinstance(cards_v2, dict):
+            cards_v2 = [cards_v2]
+
+        if not cards_v2:
+            return None
+
+        import httpx
+
+        from yuxi.channel.extensions.googlechat.api import (
+            CHAT_API_BASE,
+            get_access_token,
+        )
+
+        url = f"{CHAT_API_BASE}/{message_id}?updateMask=cardsV2"
+        token = await get_access_token(account)
+        if not token:
+            return None
+
+        body = {"cardsV2": cards_v2}
+        async with httpx.AsyncClient(timeout=httpx.Timeout(30.0)) as client:
+            resp = await client.patch(
+                url, json=body,
+                headers={
+                    "Authorization": f"Bearer {token}",
+                    "Content-Type": "application/json",
+                },
+            )
+            resp.raise_for_status()
+            return message_id
+
+    async def send_payload(self, ctx: object) -> object:
+        return None
+
+    async def send_poll(self, ctx: object) -> object:
+        return None
+
+    async def render_presentation(
+        self, payload: object, presentation: object, ctx: object
+    ) -> object | None:
+        return None
+
+    async def pin_delivered_message(
+        self, config: dict, target_ref: object, message_id: str, pin: object
+    ) -> None:
+        pass
+
+    async def before_deliver_payload(
+        self, config: dict, target_ref: object, payload: object, hint: object | None = None
+    ) -> None:
+        pass
+
+    async def after_deliver_payload(
+        self, config: dict, target_ref: object, payload: object, results: list
+    ) -> None:
+        pass
+
+    def resolve_target(
+        self,
+        to: str | None = None,
+        *,
+        config: dict | None = None,
+        allow_from: list[str] | None = None,
+        account_id: str | None = None,
+        mode: str | None = None,
+    ) -> tuple[bool, str]:
+        if not to:
+            return False, "Target is required"
+        from yuxi.channel.extensions.googlechat.targets import normalize_googlechat_target
+
+        normalized = normalize_googlechat_target(to)
+        if not normalized:
+            return False, f"Invalid target: {to}"
+        return True, normalized
+
+    def chunker(self, text: str, limit: int, ctx: object | None = None) -> list[str]:
+        return chunk_markdown_text(text, limit)
+
+    def sanitize_text(self, text: str, payload: object) -> str:
+        return sanitize_text(text)
+
+    def should_skip_plain_text_sanitization(self, payload: object) -> bool:
+        return False
+
+    def normalize_payload(
+        self, payload: object, config: dict, account_id: str | None = None
+    ) -> object | None:
+        return payload
+
+    def resolve_effective_text_chunk_limit(
+        self, config: dict, account_id: str | None = None, fallback_limit: int | None = None
+    ) -> int | None:
+        return self.text_chunk_limit
+
+    def should_treat_delivered_text_as_visible(
+        self, kind: str, text: str | None = None
+    ) -> bool:
+        return True
\ No newline at end of file
diff --git a/backend/package/yuxi/channel/extensions/googlechat/pairing.py b/backend/package/yuxi/channel/extensions/googlechat/pairing.py
new file mode 100644
index 00000000..0cee0071
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/googlechat/pairing.py
@@ -0,0 +1,62 @@
+from __future__ import annotations
+
+import logging
+
+from yuxi.channel.extensions.googlechat.api import send_message
+
+logger = logging.getLogger(__name__)
+
+_PARING_CODE_LENGTH = 6
+
+_pairing_store: dict[str, dict] = {}
+
+
+class GoogleChatPairingAdapter:
+    id_label = "Google Chat User ID"
+
+    def normalize_allow_entry(self, entry: str) -> str:
+        for prefix in ("googlechat:", "google-chat:", "gchat:"):
+            if entry.lower().startswith(prefix):
+                entry = entry[len(prefix):]
+                break
+        return entry.lower()
+
+    async def generate_code(self, peer_id: str) -> str:
+        import random
+        import string
+
+        code = "".join(random.choices(string.ascii_uppercase + string.digits, k=_PARING_CODE_LENGTH))
+        _pairing_store[peer_id] = {
+            "code": code,
+            "status": "pending",
+        }
+        return code
+
+    async def verify_code(self, peer_id: str, code: str) -> bool:
+        stored = _pairing_store.get(peer_id)
+        if stored and stored.get("code") == code:
+            stored["status"] = "verified"
+            return True
+        return False
+
+    async def notify_approval(
+        self, config: dict, peer_id: str, account_id: str | None = None
+    ) -> None:
+        logger.info("Pairing approved for peer: %s (account: %s)", peer_id, account_id)
+
+    async def send_pairing_challenge(
+        self, account, space_name: str, peer_id: str
+    ) -> str:
+        code = await self.generate_code(peer_id)
+        text = (
+            f"🔒 *Access Request*\n\n"
+            f"Your user ID: `{peer_id}`\n"
+            f"Pairing code: `{code}`\n\n"
+            f"An administrator must approve this pairing request before you can interact with this bot."
+        )
+        result = await send_message(account, space_name, text)
+        return result.get("name", "")
+
+    def is_paired(self, peer_id: str) -> bool:
+        stored = _pairing_store.get(peer_id)
+        return stored is not None and stored.get("status") == "verified"
\ No newline at end of file
diff --git a/backend/package/yuxi/channel/extensions/googlechat/plugin.json b/backend/package/yuxi/channel/extensions/googlechat/plugin.json
new file mode 100644
index 00000000..c7032da5
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/googlechat/plugin.json
@@ -0,0 +1,183 @@
+{
+  "id": "googlechat",
+  "name": "Google Chat",
+  "version": "1.0.0",
+  "description": "Google Chat 渠道插件，支持通过 Google Chat API 收发消息、卡片交互和线程对话。",
+  "author": "Yuxi Team",
+  "license": "MIT",
+  "entry_point": "yuxi.channel.extensions.googlechat",
+  "capabilities": {
+    "chat_types": ["direct", "group", "thread"],
+    "message_types": ["text", "image", "file", "video"],
+    "interactions": [],
+    "reactions": true,
+    "typing_indicator": true,
+    "threads": true,
+    "edit": true,
+    "unsend": true,
+    "reply": true,
+    "media": true,
+    "native_commands": false,
+    "polls": false,
+    "group_management": false,
+    "streaming": false,
+    "streaming_mode": null,
+    "block_streaming": true,
+    "block_streaming_chunk_min_chars": 4000,
+    "block_streaming_chunk_max_chars": 4000,
+    "block_streaming_coalesce_min_chars": 1500,
+    "block_streaming_coalesce_max_chars": 4000,
+    "adaptive_cards": true,
+    "sse_support": false
+  },
+  "config_schema": {
+    "type": "object",
+    "properties": {
+      "service_account_json": {
+        "type": "string",
+        "title": "Service Account JSON",
+        "description": "Google Cloud Service Account 凭证 JSON 字符串或文件路径",
+        "format": "textarea"
+      },
+      "space_allowlist": {
+        "type": "array",
+        "title": "Space Allowlist",
+        "description": "允许访问的 Google Chat Space ID 列表",
+        "items": {
+          "type": "string"
+        }
+      },
+      "dm_policy": {
+        "type": "string",
+        "title": "DM Policy",
+        "description": "私信访问控制策略",
+        "enum": ["open", "pairing", "closed"],
+        "default": "pairing"
+      },
+      "group_policy": {
+        "type": "string",
+        "title": "Group Policy",
+        "description": "群组访问控制策略",
+        "enum": ["open", "allowlist", "closed"],
+        "default": "allowlist"
+      },
+      "webhook_secret": {
+        "type": "string",
+        "title": "Webhook Secret",
+        "description": "Google Chat webhook 验证密钥",
+        "format": "password"
+      },
+      "bot_name": {
+        "type": "string",
+        "title": "Bot Name",
+        "description": "机器人在 Google Chat 中显示的名称",
+        "default": "Yuxi Bot"
+      },
+      "avatar_url": {
+        "type": "string",
+        "title": "Avatar URL",
+        "description": "机器人头像 URL",
+        "format": "uri"
+      },
+      "auto_join_spaces": {
+        "type": "boolean",
+        "title": "Auto Join Spaces",
+        "description": "是否自动加入邀请的 Space",
+        "default": true
+      },
+      "card_response_enabled": {
+        "type": "boolean",
+        "title": "Card Response Enabled",
+        "description": "是否启用卡片消息响应",
+        "default": true
+      },
+      "threading_enabled": {
+        "type": "boolean",
+        "title": "Threading Enabled",
+        "description": "是否启用线程对话支持",
+        "default": true
+      },
+      "mention_required_in_groups": {
+        "type": "boolean",
+        "title": "Mention Required in Groups",
+        "description": "群组中是否需要 @提及 机器人才响应",
+        "default": true
+      },
+      "max_message_length": {
+        "type": "integer",
+        "title": "Max Message Length",
+        "description": "单条消息最大字符数",
+        "default": 4096,
+        "minimum": 100,
+        "maximum": 4096
+      },
+      "rate_limit_per_minute": {
+        "type": "integer",
+        "title": "Rate Limit Per Minute",
+        "description": "每分钟最大请求数",
+        "default": 60,
+        "minimum": 10,
+        "maximum": 300
+      }
+    },
+    "required": ["service_account_json"]
+  },
+  "webhook": {
+    "enabled": true,
+    "path": "/webhooks/googlechat",
+    "methods": ["POST"],
+    "auth_type": "hmac",
+    "secret_header": "X-Goog-Signature"
+  },
+  "dependencies": ["httpx", "google-auth"],
+  "min_api_version": "1.0.0",
+  "documentation_url": "https://developers.google.com/chat",
+  "icon": "googlechat",
+  "color": "#00832d",
+  "tags": ["google", "chat", "workspace", "enterprise"],
+  "features": {
+    "rich_text": false,
+    "markdown_support": true,
+    "html_support": true,
+    "emoji_support": true,
+    "file_sharing": true,
+    "voice_messages": false,
+    "video_messages": false,
+    "location_sharing": false,
+    "contact_cards": false,
+    "message_search": true,
+    "threaded_replies": true,
+    "message_editing": true,
+    "message_deletion": true,
+    "typing_indicators": true,
+    "read_receipts": false,
+    "delivery_receipts": true,
+    "bot_commands": true,
+    "slash_commands": true,
+    "interactive_cards": true,
+    "buttons": true,
+    "quick_replies": false,
+    "menus": false,
+    "forms": true,
+    "polls": false,
+    "scheduling": false,
+    "reminders": false,
+    "notifications": true,
+    "mentions": true,
+    "hashtags": false,
+    "custom_status": false,
+    "presence_indicators": false,
+    "user_profiles": true,
+    "group_profiles": true,
+    "channel_directory": false,
+    "guest_access": true,
+    "single_sign_on": true,
+    "two_factor_auth": true,
+    "end_to_end_encryption": false,
+    "data_residency": true,
+    "gdpr_compliant": true,
+    "hipaa_compliant": false,
+    "soc2_compliant": true,
+    "iso27001_compliant": true
+  }
+}
\ No newline at end of file
diff --git a/backend/package/yuxi/channel/extensions/googlechat/reactions.py b/backend/package/yuxi/channel/extensions/googlechat/reactions.py
new file mode 100644
index 00000000..3b4652eb
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/googlechat/reactions.py
@@ -0,0 +1,66 @@
+from __future__ import annotations
+
+import logging
+
+from yuxi.channel.extensions.googlechat.api import (
+    create_reaction,
+    delete_reaction,
+    list_reactions,
+)
+from yuxi.channel.extensions.googlechat.types import ResolvedGoogleChatAccount
+
+logger = logging.getLogger(__name__)
+
+
+async def add_reaction(
+    account: ResolvedGoogleChatAccount,
+    message_name: str,
+    emoji: str,
+) -> dict:
+    return await create_reaction(account, message_name, emoji)
+
+
+async def remove_reaction(
+    account: ResolvedGoogleChatAccount,
+    message_name: str,
+    emoji: str,
+    bot_user: str | None = None,
+) -> bool:
+    reactions = await list_reactions(account, message_name)
+
+    target = None
+    for r in reactions:
+        user = r.get("user", {})
+        user_name = user.get("name", "")
+        if user_name == "users/app" or (bot_user and user_name == bot_user):
+            if r.get("emoji", {}).get("unicode") == emoji:
+                target = r
+                break
+
+    if not target:
+        return False
+
+    await delete_reaction(account, target["name"])
+    return True
+
+
+async def get_reactions(
+    account: ResolvedGoogleChatAccount,
+    message_name: str,
+) -> list[dict]:
+    return await list_reactions(account, message_name)
+
+
+async def execute_react_action(
+    account: ResolvedGoogleChatAccount,
+    message_name: str,
+    emoji: str,
+    remove: bool = False,
+    bot_user: str | None = None,
+) -> dict:
+    if remove:
+        removed = await remove_reaction(account, message_name, emoji, bot_user)
+        return {"success": removed, "action": "remove", "emoji": emoji}
+    else:
+        result = await add_reaction(account, message_name, emoji)
+        return {"success": True, "action": "add", "emoji": emoji, "result": result}
\ No newline at end of file
diff --git a/backend/package/yuxi/channel/extensions/googlechat/reply_delivery.py b/backend/package/yuxi/channel/extensions/googlechat/reply_delivery.py
new file mode 100644
index 00000000..6c59bc0d
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/googlechat/reply_delivery.py
@@ -0,0 +1,97 @@
+from __future__ import annotations
+
+import logging
+
+from yuxi.channel.extensions.googlechat.api import (
+    delete_message,
+    send_message,
+    update_message,
+)
+from yuxi.channel.extensions.googlechat.formatting import (
+    build_typing_indicator_text,
+    chunk_markdown_text,
+    resolve_bot_display_name,
+)
+from yuxi.channel.extensions.googlechat.types import ResolvedGoogleChatAccount
+
+logger = logging.getLogger(__name__)
+
+
+class TypingPlaceholder:
+    def __init__(self, message_name: str, space_name: str, consumed: bool = False):
+        self.message_name = message_name
+        self.space_name = space_name
+        self.consumed = consumed
+
+
+async def send_typing_placeholder(
+    account: ResolvedGoogleChatAccount,
+    space_name: str,
+    bot_name: str | None = None,
+) -> TypingPlaceholder:
+    name = bot_name or resolve_bot_display_name(account.config if hasattr(account, "config") else {})
+    text = build_typing_indicator_text(name)
+    result = await send_message(account, space_name, text)
+    message_name = result.get("name", "")
+    return TypingPlaceholder(message_name=message_name, space_name=space_name)
+
+
+async def deliver_text_chunk(
+    account: ResolvedGoogleChatAccount,
+    space_name: str,
+    text: str,
+    placeholder: TypingPlaceholder | None = None,
+    thread_name: str | None = None,
+) -> str:
+    if placeholder and not placeholder.consumed:
+        await update_message(account, placeholder.message_name, text)
+        placeholder.consumed = True
+        return placeholder.message_name
+
+    if placeholder and placeholder.consumed:
+        result = await send_message(account, space_name, text, thread_name=thread_name)
+        return result.get("name", "")
+
+    result = await send_message(account, space_name, text, thread_name=thread_name)
+    return result.get("name", "")
+
+
+async def deliver_text_with_streaming(
+    account: ResolvedGoogleChatAccount,
+    space_name: str,
+    full_text: str,
+    bot_name: str | None = None,
+    thread_name: str | None = None,
+    text_chunk_limit: int = 4000,
+) -> list[str]:
+    chunks = chunk_markdown_text(full_text, text_chunk_limit)
+    if not chunks:
+        return []
+
+    placeholder = await send_typing_placeholder(account, space_name, bot_name)
+    message_names = []
+
+    for i, chunk in enumerate(chunks):
+        try:
+            name = await deliver_text_chunk(
+                account, space_name, chunk, placeholder if i == 0 else None, thread_name
+            )
+            if name:
+                message_names.append(name)
+        except Exception as e:
+            logger.warning("Failed to deliver chunk %d: %s", i, e)
+
+    return message_names
+
+
+async def clear_typing_placeholder(
+    account: ResolvedGoogleChatAccount,
+    placeholder: TypingPlaceholder,
+) -> None:
+    try:
+        await delete_message(account, placeholder.message_name)
+    except Exception:
+        try:
+            await update_message(account, placeholder.message_name, "")
+        except Exception:
+            pass
\ No newline at end of file
diff --git a/backend/package/yuxi/channel/extensions/googlechat/security.py b/backend/package/yuxi/channel/extensions/googlechat/security.py
new file mode 100644
index 00000000..b2a59d02
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/googlechat/security.py
@@ -0,0 +1,87 @@
+from __future__ import annotations
+
+import logging
+import re
+
+logger = logging.getLogger(__name__)
+
+_EMAIL_PATTERN = re.compile(r".+@.+\..+")
+
+
+class GoogleChatSecurityAdapter:
+    def __init__(self):
+        self._dm_policy: str = "pairing"
+
+    def resolve_dm_policy(self) -> dict:
+        return {"mode": self._dm_policy, "allow_from": []}
+
+    def check_allow_entry(self, entry: str) -> bool:
+        if not entry:
+            return False
+        if entry.startswith("googlechat:"):
+            entry = entry[len("googlechat:"):]
+        if entry.startswith("users/") and _EMAIL_PATTERN.match(entry[6:]):
+            return False
+        return bool(entry)
+
+    def normalize_allow_entry(self, entry: str) -> str:
+        if not entry:
+            return entry
+        for prefix in ("googlechat:", "google-chat:", "gchat:"):
+            if entry.lower().startswith(prefix):
+                entry = entry[len(prefix):]
+                break
+        if entry.startswith("user:users/"):
+            entry = entry[5:]
+        elif entry.startswith("space:spaces/"):
+            entry = entry[6:]
+        return entry.lower()
+
+    def collect_warnings(self, config: dict, account_id: str | None = None) -> list[str]:
+        warnings = []
+        gc_config = config.get("channels", {}).get("googlechat", {})
+
+        group_policy = gc_config.get("groupPolicy", "allowlist")
+        if group_policy == "open":
+            warnings.append(
+                "Group policy is 'open' - any Google Chat Space can trigger the bot. Consider using 'allowlist' for production."
+            )
+
+        dm_config = gc_config.get("dm", {})
+        dm_policy = dm_config.get("policy", "pairing")
+        if dm_policy == "open":
+            warnings.append(
+                "DM policy is 'open' - anyone can DM the bot. Consider using 'pairing' or 'allowlist' for production."
+            )
+
+        allow_from = dm_config.get("allowFrom", [])
+        for entry in allow_from:
+            normalized = self.normalize_allow_entry(entry)
+            if normalized.startswith("users/") and "@" in normalized:
+                warnings.append(
+                    f"DM allowFrom entry '{entry}' uses email format. Email addresses are mutable; use users/<numeric_id> instead."
+                )
+
+        groups = gc_config.get("groups", {})
+        for group_key in groups:
+            if not group_key.startswith("spaces/") and group_key != "*":
+                warnings.append(
+                    f"Group key '{group_key}' is not a stable spaces/<id> name. This may break routing."
+                )
+
+        return warnings
+
+    def check_dm_allowlist(
+        self, sender_name: str, allow_from: list[str], dangerously_allow_name_matching: bool = False
+    ) -> bool:
+        if "*" in allow_from:
+            return True
+
+        for entry in allow_from:
+            if "@" in entry and not dangerously_allow_name_matching:
+                continue
+            normalized = self.normalize_allow_entry(entry)
+            if normalized == self.normalize_allow_entry(sender_name):
+                return True
+
+        return False
\ No newline at end of file
diff --git a/backend/package/yuxi/channel/extensions/googlechat/session.py b/backend/package/yuxi/channel/extensions/googlechat/session.py
new file mode 100644
index 00000000..5b3bff90
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/googlechat/session.py
@@ -0,0 +1,48 @@
+from __future__ import annotations
+
+from yuxi.channel.protocols import SessionResolution
+
+
+class GoogleChatSessionAdapter:
+    def resolve_session(self, msg) -> SessionResolution:
+        from yuxi.channel.routing.models import PeerKind
+
+        if hasattr(msg, "sender") and hasattr(msg.sender, "kind"):
+            if msg.sender.kind == PeerKind.DIRECT:
+                space_id = _extract_space_id(msg)
+                return SessionResolution(
+                    kind="direct",
+                    conversation_id=space_id or msg.sender.id,
+                )
+
+        space_id = _extract_space_id(msg)
+        if space_id:
+            return SessionResolution(kind="group", conversation_id=space_id)
+
+        if msg.group and msg.group.id:
+            return SessionResolution(kind="group", conversation_id=msg.group.id)
+
+        return SessionResolution(kind="group", conversation_id="unknown")
+
+    @staticmethod
+    def build_dm_session_key(agent_id: str, account_id: str, space_id: str) -> str:
+        return f"agent:{agent_id}:googlechat:direct:{space_id}"
+
+    @staticmethod
+    def build_group_session_key(agent_id: str, account_id: str, space_id: str) -> str:
+        return f"agent:{agent_id}:googlechat:group:{space_id}"
+
+
+def _extract_space_id(msg) -> str | None:
+    metadata = getattr(msg, "metadata", {}) or {}
+    space_name = metadata.get("space_name", "")
+    if space_name:
+        return space_name
+
+    raw = getattr(msg, "raw_payload", {}) or {}
+    event = raw.get("event", raw)
+    space = event.get("space", {})
+    if space:
+        return space.get("name")
+
+    return None
\ No newline at end of file
diff --git a/backend/package/yuxi/channel/extensions/googlechat/status.py b/backend/package/yuxi/channel/extensions/googlechat/status.py
new file mode 100644
index 00000000..e1116d1a
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/googlechat/status.py
@@ -0,0 +1,82 @@
+from __future__ import annotations
+
+import logging
+
+from yuxi.channel.extensions.googlechat.auth import probe_account
+from yuxi.channel.extensions.googlechat.types import ResolvedGoogleChatAccount
+
+logger = logging.getLogger(__name__)
+
+
+class GoogleChatStatusAdapter:
+    async def probe(self, account_dict: dict) -> bool:
+        account = account_dict.get("resolved")
+        if not isinstance(account, ResolvedGoogleChatAccount):
+            return False
+        result = await probe_account(account)
+        return result.get("ok", False)
+
+    def build_summary(self, snapshot) -> dict:
+        return {
+            "channel": "googlechat",
+            "mode": "webhook",
+        }
+
+    def build_account_snapshot(self, account_dict: dict) -> dict:
+        return {
+            "account_id": account_dict.get("account_id", "default"),
+            "name": account_dict.get("name", ""),
+            "configured": account_dict.get("credential_source", "none") != "none",
+            "enabled": account_dict.get("enabled", True),
+        }
+
+    def collect_status_issues(self, accounts: list[dict]) -> list[dict]:
+        issues = []
+        for account in accounts:
+            if not account:
+                continue
+
+            if not account.get("audience"):
+                issues.append({
+                    "channel": "googlechat",
+                    "account_id": account.get("account_id", "default"),
+                    "kind": "audience_missing",
+                    "message": "Audience is not configured. Webhook token verification will fail.",
+                    "fix": "Set channels.googlechat.audience and channels.googlechat.audienceType.",
+                })
+
+            if not account.get("audience_type"):
+                issues.append({
+                    "channel": "googlechat",
+                    "account_id": account.get("account_id", "default"),
+                    "kind": "audience_type_missing",
+                    "message": "audienceType is not configured.",
+                    "fix": "Set channels.googlechat.audienceType to 'app-url' or 'project-number'.",
+                })
+
+            resolved = account.get("resolved")
+            if isinstance(resolved, ResolvedGoogleChatAccount):
+                if resolved.audience_type == "app-url" and not resolved.app_principal:
+                    issues.append({
+                        "channel": "googlechat",
+                        "account_id": account.get("account_id", "default"),
+                        "kind": "app_principal_missing",
+                        "message": (
+                            'appPrincipal is missing for audienceType "app-url"; '
+                            "add-on token verification will fail. "
+                            "Set appPrincipal to the numeric OAuth 2.0 client ID (uniqueId, 21 digits), not an email."
+                        ),
+                    })
+
+                if resolved.app_principal and "@" in resolved.app_principal:
+                    issues.append({
+                        "channel": "googlechat",
+                        "account_id": account.get("account_id", "default"),
+                        "kind": "app_principal_is_email",
+                        "message": (
+                            f'appPrincipal "{resolved.app_principal}" looks like an email address. '
+                            "Set appPrincipal to the numeric OAuth 2.0 client ID (uniqueId, 21 digits), not an email."
+                        ),
+                    })
+
+        return issues
\ No newline at end of file
diff --git a/backend/package/yuxi/channel/extensions/googlechat/streaming.py b/backend/package/yuxi/channel/extensions/googlechat/streaming.py
new file mode 100644
index 00000000..5131cdab
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/googlechat/streaming.py
@@ -0,0 +1,38 @@
+from __future__ import annotations
+
+import logging
+
+from yuxi.channel.extensions.googlechat.api import send_message
+from yuxi.channel.extensions.googlechat.formatting import chunk_markdown_text
+from yuxi.channel.extensions.googlechat.types import ResolvedGoogleChatAccount
+
+logger = logging.getLogger(__name__)
+
+
+async def stream_text_by_chunks(
+    account: ResolvedGoogleChatAccount,
+    space_name: str,
+    full_text: str,
+    *,
+    thread_name: str | None = None,
+    chunk_limit: int = 4000,
+) -> list[str]:
+    chunks = chunk_markdown_text(full_text, chunk_limit)
+    message_names: list[str] = []
+    for chunk in chunks:
+        result = await send_message(account, space_name, chunk, thread_name=thread_name)
+        name = result.get("name", "")
+        if name:
+            message_names.append(name)
+    return message_names
+
+
+async def send_coalesced_text(
+    account: ResolvedGoogleChatAccount,
+    space_name: str,
+    text: str,
+    *,
+    thread_name: str | None = None,
+) -> str:
+    result = await send_message(account, space_name, text, thread_name=thread_name)
+    return result.get("name", "")
\ No newline at end of file
diff --git a/backend/package/yuxi/channel/extensions/googlechat/targets.py b/backend/package/yuxi/channel/extensions/googlechat/targets.py
new file mode 100644
index 00000000..2a1e47fb
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/googlechat/targets.py
@@ -0,0 +1,46 @@
+from __future__ import annotations
+
+
+def normalize_googlechat_target(raw: str) -> str | None:
+    if not raw:
+        return None
+    t = raw.strip()
+    for prefix in ("googlechat:", "google-chat:", "gchat:"):
+        if t.lower().startswith(prefix):
+            t = t[len(prefix):]
+            t = t.strip()
+            break
+    if t.lower().startswith("user:users/"):
+        t = t[5:]
+    elif t.lower().startswith("space:spaces/"):
+        t = t[6:]
+    if "@" in t and not t.startswith("users/") and not t.startswith("spaces/"):
+        t = f"users/{t.lower()}"
+    return t
+
+
+def strip_message_suffix(target: str) -> str:
+    idx = target.find("/messages/")
+    return target[:idx] if idx != -1 else target
+
+
+async def resolve_outbound_space(account, target: str) -> str | None:
+    from yuxi.channel.extensions.googlechat.api import find_direct_message
+
+    normalized = normalize_googlechat_target(target)
+    if not normalized:
+        return None
+    base = strip_message_suffix(normalized)
+    if base.startswith("spaces/"):
+        return base
+    if base.startswith("users/"):
+        return await find_direct_message(account, base)
+    return None
+
+
+def is_space_target(target: str) -> bool:
+    return target.startswith("spaces/")
+
+
+def is_user_target(target: str) -> bool:
+    return target.startswith("users/")
\ No newline at end of file
diff --git a/backend/package/yuxi/channel/extensions/googlechat/types.py b/backend/package/yuxi/channel/extensions/googlechat/types.py
new file mode 100644
index 00000000..bfab5f2c
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/googlechat/types.py
@@ -0,0 +1,189 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+
+
+@dataclass
+class GoogleChatUser:
+    name: str = ""
+    display_name: str | None = None
+    email: str | None = None
+    type: str = "HUMAN"
+    avatar_url: str | None = None
+
+
+@dataclass
+class GoogleChatSpace:
+    name: str = ""
+    type: str = "GROUP_CHAT"
+    display_name: str | None = None
+    single_user_bot_dm: bool = False
+
+
+@dataclass
+class GoogleChatAttachment:
+    name: str | None = None
+    content_name: str | None = None
+    content_type: str | None = None
+    attachment_upload_token: str | None = None
+    drive_data_ref: dict | None = None
+    thumbnail_uri: str | None = None
+    download_uri: str | None = None
+    source: str = "UPLOAD_CONTENT"
+
+
+@dataclass
+class GoogleChatAnnotation:
+    type: str = ""
+    user_mention: dict | None = None
+    slash_command: dict | None = None
+
+
+@dataclass
+class GoogleChatMessage:
+    name: str = ""
+    sender: GoogleChatUser = field(default_factory=GoogleChatUser)
+    space: GoogleChatSpace = field(default_factory=GoogleChatSpace)
+    text: str = ""
+    argument_text: str = ""
+    thread: dict | None = None
+    attachment: list[dict] = field(default_factory=list)
+    annotations: list[dict] = field(default_factory=list)
+    create_time: str = ""
+    fallback_text: str = ""
+    formatted_text: str = ""
+    last_update_time: str = ""
+
+
+@dataclass
+class GoogleChatEvent:
+    type: str = ""
+    event_time: str = ""
+    message: GoogleChatMessage | None = None
+    space: GoogleChatSpace | None = None
+    user: GoogleChatUser | None = None
+    thread: dict | None = None
+    reaction: dict | None = None
+    config_complete_redirect_url: str | None = None
+    is_dialog_event: bool = False
+    dialog_action_type: str | None = None
+    common: dict | None = None
+
+    @classmethod
+    def from_payload(cls, payload: dict) -> GoogleChatEvent:
+        event_data = payload.get("event", payload)
+        event = cls(
+            type=event_data.get("type", ""),
+            event_time=event_data.get("eventTime", ""),
+            config_complete_redirect_url=payload.get("configCompleteRedirectUrl"),
+            is_dialog_event=payload.get("isDialogEvent", False),
+            dialog_action_type=payload.get("dialogActionType"),
+            common=payload.get("common"),
+        )
+
+        space_data = event_data.get("space") or payload.get("space")
+        if space_data:
+            event.space = GoogleChatSpace(
+                name=space_data.get("name", ""),
+                type=space_data.get("type", "GROUP_CHAT"),
+                display_name=space_data.get("displayName"),
+                single_user_bot_dm=space_data.get("singleUserBotDm", False),
+            )
+
+        user_data = event_data.get("user") or payload.get("user")
+        if user_data:
+            event.user = GoogleChatUser(
+                name=user_data.get("name", ""),
+                display_name=user_data.get("displayName"),
+                email=user_data.get("email"),
+                type=user_data.get("type", "HUMAN"),
+                avatar_url=user_data.get("avatarUrl"),
+            )
+
+        msg_data = event_data.get("message")
+        if msg_data:
+            sender_data = msg_data.get("sender", {})
+            event.message = GoogleChatMessage(
+                name=msg_data.get("name", ""),
+                sender=GoogleChatUser(
+                    name=sender_data.get("name", ""),
+                    display_name=sender_data.get("displayName"),
+                    email=sender_data.get("email"),
+                    type=sender_data.get("type", "HUMAN"),
+                    avatar_url=sender_data.get("avatarUrl"),
+                ),
+                text=msg_data.get("text", ""),
+                argument_text=msg_data.get("argumentText", ""),
+                thread=msg_data.get("thread"),
+                attachment=msg_data.get("attachment", []),
+                annotations=msg_data.get("annotations", []),
+                create_time=msg_data.get("createTime", ""),
+                fallback_text=msg_data.get("fallbackText", ""),
+                formatted_text=msg_data.get("formattedText", ""),
+                last_update_time=msg_data.get("lastUpdateTime", ""),
+            )
+            if msg_data.get("space"):
+                msg_space = msg_data["space"]
+                event.message.space = GoogleChatSpace(
+                    name=msg_space.get("name", ""),
+                    type=msg_space.get("type", "GROUP_CHAT"),
+                    display_name=msg_space.get("displayName"),
+                    single_user_bot_dm=msg_space.get("singleUserBotDm", False),
+                )
+
+        return event
+
+    @property
+    def effective_space(self) -> GoogleChatSpace | None:
+        if self.space:
+            return self.space
+        if self.message and self.message.space:
+            return self.message.space
+        return None
+
+    @property
+    def effective_space_name(self) -> str:
+        s = self.effective_space
+        return s.name if s else ""
+
+    @property
+    def space_type(self) -> str:
+        s = self.effective_space
+        return s.type if s else "GROUP_CHAT"
+
+
+@dataclass
+class ResolvedGoogleChatAccount:
+    account_id: str
+    name: str = ""
+    credential_source: str = "none"
+    service_account_info: dict | None = None
+    service_account_file: str | None = None
+    audience_type: str | None = None
+    audience: str | None = None
+    webhook_path: str = "/api/webhook/googlechat"
+    webhook_url: str | None = None
+    bot_user: str | None = None
+    app_principal: str | None = None
+    group_policy: str = "allowlist"
+    require_mention: bool = True
+    allow_text_commands: bool = False
+    reply_to_mode: str = "off"
+    text_chunk_limit: int = 4000
+    media_max_mb: int = 20
+    typing_indicator: str = "message"
+    allow_bots: bool = False
+    dangerously_allow_name_matching: bool = False
+    actions_reactions: bool = True
+    dm_policy: str = "pairing"
+    dm_enabled: bool = True
+    dm_allow_from: list[str] = field(default_factory=list)
+    groups: dict = field(default_factory=dict)
+    block_streaming_coalesce_min_chars: int = 1500
+    block_streaming_coalesce_idle_ms: int = 1000
+    enabled: bool = True
+    config: dict = field(default_factory=dict)
+
+    @property
+    def is_configured(self) -> bool:
+        return self.credential_source != "none"
diff --git a/backend/package/yuxi/channel/extensions/googlechat/webhook.py b/backend/package/yuxi/channel/extensions/googlechat/webhook.py
new file mode 100644
index 00000000..52ae350e
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/googlechat/webhook.py
@@ -0,0 +1,308 @@
+from __future__ import annotations
+
+import asyncio
+import logging
+
+from yuxi.channel.extensions.googlechat.auth import verify_webhook_token
+from yuxi.channel.extensions.googlechat.formatting import sanitize_text
+from yuxi.channel.extensions.googlechat.monitor import GoogleChatMonitor
+from yuxi.channel.extensions.googlechat.types import GoogleChatEvent, ResolvedGoogleChatAccount
+
+logger = logging.getLogger(__name__)
+
+_PRE_AUTH_BODY_LIMIT = 16 * 1024
+_PRE_AUTH_TIMEOUT = 3.0
+
+
+class GoogleChatWebhookHandler:
+    def __init__(self, monitor: GoogleChatMonitor | None = None):
+        self._monitor = monitor or GoogleChatMonitor()
+        self._targets: dict[str, ResolvedGoogleChatAccount] = {}
+
+    def register_target(self, webhook_path: str, account: ResolvedGoogleChatAccount) -> None:
+        self._targets[webhook_path] = account
+
+    def unregister_target(self, webhook_path: str) -> None:
+        self._targets.pop(webhook_path, None)
+
+    def _resolve_account(
+        self, token: str = "", webhook_path: str = ""
+    ) -> ResolvedGoogleChatAccount | None:
+        if webhook_path and webhook_path in self._targets:
+            return self._targets[webhook_path]
+
+        if token and self._targets:
+            from yuxi.channel.extensions.googlechat.auth import decode_unverified_jwt_payload
+
+            payload = decode_unverified_jwt_payload(token)
+            target_audience = payload.get("aud", "")
+            for path, account in self._targets.items():
+                if account.audience and account.audience == target_audience:
+                    return account
+
+        for account in self._targets.values():
+            return account
+        return None
+
+    async def handle_webhook(self, payload: dict) -> dict:
+        token = payload.get("_bearer_token", "")
+        raw_body = payload.get("_raw_body")
+
+        account = self._resolve_account(token=token)
+        audience_type = account.audience_type or "app-url" if account else "app-url"
+        audience = account.audience or "" if account else ""
+        app_principal = account.app_principal if account else None
+
+        if token:
+            ok, verified_payload, error = await verify_webhook_token(
+                token, audience_type, audience, app_principal
+            )
+            if not ok:
+                logger.warning("Webhook auth failed: %s", error)
+                return {"error": "unauthorized", "detail": error}
+
+        body = raw_body or payload
+
+        if not token:
+            if isinstance(body, bytes) and len(body) > _PRE_AUTH_BODY_LIMIT:
+                return {"error": "payload too large"}
+
+        event = GoogleChatEvent.from_payload(body)
+        return await self._process_event(body, event, account)
+
+    async def _process_event(
+        self, raw_payload: dict, event: GoogleChatEvent, account: ResolvedGoogleChatAccount | None = None
+    ) -> dict:
+        if event.type == "ADDED_TO_SPACE":
+            return await self._handle_added_to_space(event, account)
+
+        if event.type == "REMOVED_FROM_SPACE":
+            return await self._handle_removed_from_space(event, account)
+
+        if event.type == "CARD_CLICKED":
+            return await self._handle_card_clicked(event, account)
+
+        if event.type == "APP_HOME":
+            return await self._handle_app_home(event, account)
+
+        if event.type != "MESSAGE":
+            return {"status": "ignored", "reason": f"event type: {event.type}"}
+
+        if not event.message:
+            return {"status": "ignored", "reason": "no message in event"}
+
+        msg = event.message
+        sender = msg.sender
+
+        allow_bots = account.allow_bots if account else False
+        if sender.type == "BOT" and not allow_bots:
+            return {"status": "ignored", "reason": "bot message"}
+
+        if sender.name == "users/app":
+            return {"status": "ignored", "reason": "self message"}
+
+        text = msg.argument_text or msg.text
+        if not text:
+            return {"status": "ignored", "reason": "empty text"}
+
+        safe_text = sanitize_text(text)
+
+        if msg.thread and account:
+            thread_name = msg.thread.get("name", "")
+            if thread_name:
+                try:
+                    thread_ctx = await asyncio.wait_for(
+                        self._monitor.inject_thread_context(
+                            account, event.effective_space_name, thread_name
+                        ),
+                        timeout=5.0,
+                    )
+                    if thread_ctx:
+                        safe_text = f"{thread_ctx}\n{safe_text}"
+                except (asyncio.TimeoutError, Exception):
+                    pass
+
+        try:
+            unified_msg = self._monitor.event_to_unified_message(
+                raw_payload, event, safe_text, account=account
+            )
+            if unified_msg is None:
+                return {"status": "filtered"}
+
+            from yuxi.channel.runtime.manager import gateway
+
+            processor = getattr(gateway, "_processor", None) or getattr(gateway, "processor", None)
+            if processor:
+                await asyncio.wait_for(processor.process(unified_msg), timeout=120.0)
+
+            return {"status": "processed"}
+        except asyncio.TimeoutError:
+            logger.error("Agent response timeout for Google Chat message")
+            return {"status": "timeout"}
+        except Exception:
+            logger.exception("Failed to process Google Chat message")
+            return {"status": "error"}
+
+    async def _handle_added_to_space(
+        self, event: GoogleChatEvent, account: ResolvedGoogleChatAccount | None = None
+    ) -> dict:
+        space_name = event.effective_space_name
+
+        if not space_name or not account:
+            return {"status": "skipped", "reason": "no space or account"}
+
+        space_display = (
+            event.effective_space.display_name if event.effective_space and event.effective_space.display_name
+            else "this space"
+        )
+
+        welcome_text = (
+            f"Hi! I'm a bot powered by ForcePilot. "
+            f"I've been added to *{space_display}*.\n\n"
+            f"• Mention me with @ to ask questions\n"
+            f"• I can participate in threads\n"
+            f"• Use /help to see available commands"
+        )
+
+        try:
+            from yuxi.channel.extensions.googlechat.api import send_message
+
+            await send_message(account, space_name, welcome_text)
+            logger.info("Sent welcome message to space: %s", space_name)
+            return {"status": "processed", "action": "welcome"}
+        except Exception:
+            logger.exception("Failed to send welcome message to space: %s", space_name)
+            return {"status": "error", "action": "welcome"}
+
+    async def _handle_removed_from_space(
+        self, event: GoogleChatEvent, account: ResolvedGoogleChatAccount | None = None
+    ) -> dict:
+        space_name = event.effective_space_name
+
+        if not space_name:
+            return {"status": "skipped", "reason": "no space"}
+
+        logger.info(
+            "Bot removed from space: %s, user: %s",
+            space_name,
+            event.user.name if event.user else "unknown",
+        )
+
+        if account:
+            from yuxi.channel.runtime.manager import gateway
+
+            channel_ctx = getattr(gateway, "_channel_contexts", {}).get(account.account_id)
+            if channel_ctx and hasattr(channel_ctx, "webhook_handler"):
+                channel_ctx.webhook_handler.unregister_target(account.webhook_path)
+
+        return {"status": "processed", "action": "cleanup"}
+
+    async def _handle_card_clicked(
+        self, event: GoogleChatEvent, account: ResolvedGoogleChatAccount | None = None
+    ) -> dict:
+        import time
+
+        common = event.common or {}
+        action = common.get("invokedFunction", "")
+        parameters = common.get("parameters", {})
+
+        user = event.user
+        user_id = user.name if user else "unknown"
+        space_name = event.effective_space_name
+
+        logger.info("CARD_CLICKED: action=%s user=%s space=%s", action, user_id, space_name)
+
+        from yuxi.channel.message.models import (
+            MessageType,
+            PeerInfo,
+            UnifiedMessage,
+        )
+        from yuxi.channel.routing.models import PeerKind
+
+        unified_msg = UnifiedMessage(
+            msg_id=f"card_clicked:{user_id}:{int(time.time())}",
+            channel_type="googlechat",
+            account_id=account.account_id if account else "default",
+            content=f"[card_action:{action}]",
+            sender=PeerInfo(
+                kind=PeerKind.GROUP if space_name else PeerKind.DIRECT,
+                id=user_id,
+                display_name=user.display_name if user else None,
+            ),
+            message_type=MessageType.COMMAND,
+            metadata={
+                "ChatType": "direct" if not space_name else "group",
+                "InteractionType": "card_clicked",
+                "InvokedFunction": action,
+                "Parameters": parameters,
+                "space_name": space_name,
+            },
+            raw_payload=event.__dict__ if hasattr(event, "__dict__") else {},
+        )
+
+        try:
+            from yuxi.channel.runtime.manager import gateway
+
+            processor = getattr(gateway, "_processor", None) or getattr(gateway, "processor", None)
+            if processor:
+                await asyncio.wait_for(processor.process(unified_msg), timeout=120.0)
+            return {"status": "processed", "action": "card_clicked"}
+        except asyncio.TimeoutError:
+            logger.error("Agent response timeout for CARD_CLICKED event")
+            return {"status": "timeout", "action": "card_clicked"}
+        except Exception:
+            logger.exception("Failed to process CARD_CLICKED event")
+            return {"status": "error", "action": "card_clicked"}
+
+    async def _handle_app_home(
+        self, event: GoogleChatEvent, account: ResolvedGoogleChatAccount | None = None
+    ) -> dict:
+        space_name = event.effective_space_name
+        if not space_name or not account:
+            return {"status": "skipped", "reason": "no space or account"}
+
+        cards_v2 = [{
+            "cardId": "app_home",
+            "card": {
+                "header": {"title": "ForcePilot AI Assistant"},
+                "sections": [{
+                    "widgets": [
+                        {
+                            "textParagraph": {
+                                "text": "Ask me anything! I'm here to help you with tasks, research, and more."
+                            }
+                        },
+                        {
+                            "buttonList": {
+                                "buttons": [
+                                    {
+                                        "text": "Start a Conversation",
+                                        "onClick": {"action": {"function": "start_conversation"}},
+                                    },
+                                    {
+                                        "text": "Help",
+                                        "onClick": {"action": {"function": "show_help"}},
+                                    },
+                                ]
+                            }
+                        },
+                    ],
+                }],
+            },
+        }]
+
+        try:
+            from yuxi.channel.extensions.googlechat.api import send_card_message
+
+            await send_card_message(
+                account, space_name, cards_v2, fallback_text="ForcePilot AI Assistant"
+            )
+            return {"status": "processed", "action": "app_home"}
+        except Exception:
+            logger.exception("Failed to render APP_HOME")
+            return {"status": "error", "action": "app_home"}
+
+    async def handle_webhook_request(self, raw_body: dict) -> dict:
+        if not raw_body or not raw_body.get("event"):
+            return {"error": "invalid request", "detail": "No valid event in payload"}
+        return await self.handle_webhook(raw_body)