feat(confluence): 新增Confluence渠道插件完整实现

新增了完整的Confluence集成插件，包含以下核心功能：
1.  基础认证与配置管理，支持API Token和OAuth2两种认证方式
2.  评论去重、权限控制与提及解析
3.  知识库检索与页面内容处理
4.  评论收发、编辑删除与流式回复支持
5.  附件与页面标签管理
6.  内容属性存储与AI元数据管理
7.  Webhook事件接收与处理
8.  完整的插件配置与状态检查

backend/package/yuxi/channel/extensions/confluence/__init__.py

@@ -0,0 +1,359 @@
+import logging
+
+from yuxi.channel.capabilities import ChannelCapabilities
+from yuxi.channel.context import ChannelContext
+from yuxi.channel.extensions.base import BaseChannelPlugin
+from yuxi.channel.extensions.confluence.attachments import ConfluenceAttachmentManager
+from yuxi.channel.extensions.confluence.config import list_confluence_account_ids, resolve_confluence_account
+from yuxi.channel.extensions.confluence.content_properties import ContentPropertyStore
+from yuxi.channel.extensions.confluence.gateway import ConfluenceGateway
+from yuxi.channel.extensions.confluence.knowledge import ConfluenceKnowledge
+from yuxi.channel.extensions.confluence.labels import ConfluenceLabelManager
+from yuxi.channel.extensions.confluence.mentions import ConfluenceMentions
+from yuxi.channel.extensions.confluence.outbound import ConfluenceOutbound
+from yuxi.channel.extensions.confluence.page_writer import ConfluencePageWriter
+from yuxi.channel.extensions.confluence.security import ConfluenceCommentPolicy
+from yuxi.channel.extensions.confluence.status import ConfluenceStatus
+from yuxi.channel.extensions.confluence.streaming import ConfluenceStreaming
+from yuxi.channel.extensions.confluence.types import InboundConfluenceComment, PageContext
+from yuxi.channel.extensions.confluence.webhook import ConfluenceWebhookReceiver
+from yuxi.channel.plugins.registry import ChannelPluginRegistry
+
+logger = logging.getLogger(__name__)
+
+
+class ConfluencePlugin(BaseChannelPlugin):
+    id = "confluence"
+    name = "Confluence"
+    order = 90
+    label = "Confluence"
+    aliases = ["confluence-cloud", "atlassian-confluence"]
+    resolve_reply_to_mode = "off"
+
+    def __init__(self):
+        super().__init__()
+        self._gateway: ConfluenceGateway | None = None
+        self._outbound: ConfluenceOutbound | None = None
+        self._status: ConfluenceStatus | None = None
+        self._security: ConfluenceCommentPolicy | None = None
+        self._streaming = ConfluenceStreaming()
+        self._page_writer: ConfluencePageWriter | None = None
+        self._knowledge: ConfluenceKnowledge | None = None
+        self._webhook: ConfluenceWebhookReceiver | None = None
+        self._attachment_manager: ConfluenceAttachmentManager | None = None
+        self._label_manager: ConfluenceLabelManager | None = None
+        self._content_property_store: ContentPropertyStore | None = None
+
+    def config_schema(self) -> dict:
+        return {
+            "type": "object",
+            "properties": {
+                "url": {
+                    "type": "string",
+                    "title": "Confluence 站点 URL",
+                    "description": "例如 https://your-domain.atlassian.net",
+                },
+                "email": {
+                    "type": "string",
+                    "title": "Bot 账号邮箱",
+                },
+                "api_token": {
+                    "type": "string",
+                    "title": "API Token",
+                    "format": "password",
+                },
+                "auth_mode": {
+                    "type": "string",
+                    "title": "认证方式",
+                    "enum": ["api_token", "pat", "oauth"],
+                    "default": "api_token",
+                },
+                "space_keys": {
+                    "type": "string",
+                    "title": "空间 Key（逗号分隔）",
+                    "description": "留空则监听所有空间",
+                },
+                "comment_poll_enabled": {
+                    "type": "boolean",
+                    "title": "启用评论轮询",
+                    "default": True,
+                },
+                "poll_interval": {
+                    "type": "number",
+                    "title": "轮询间隔（秒）",
+                    "default": 30,
+                },
+                "comment_policy": {
+                    "type": "string",
+                    "title": "评论策略",
+                    "enum": ["open", "allowlist", "pairing", "disabled"],
+                    "default": "open",
+                },
+                "allow_from": {
+                    "type": "string",
+                    "title": "允许评论的用户 ID（逗号分隔）",
+                    "description": "仅 allowlist 模式下生效",
+                },
+                "enable_page_writer": {
+                    "type": "boolean",
+                    "title": "启用 AI 页面编写",
+                    "default": False,
+                },
+                "enable_knowledge_retrieval": {
+                    "type": "boolean",
+                    "title": "启用知识库检索",
+                    "default": False,
+                },
+                "enable_webhook": {
+                    "type": "boolean",
+                    "title": "启用 Webhook 接收",
+                    "default": False,
+                },
+                "name": {
+                    "type": "string",
+                    "title": "Bot 显示名称",
+                    "default": "Confluence Bot",
+                },
+            },
+            "required": ["url", "email", "api_token"],
+        }
+
+    @property
+    def capabilities(self) -> ChannelCapabilities:
+        return ChannelCapabilities(
+            chat_types=["comment"],
+            message_types=["text"],
+            interactions=[],
+            reactions=False,
+            typing_indicator=False,
+            threads=True,
+            edit=True,
+            unsend=True,
+            reply=True,
+            media=False,
+            effects=False,
+            native_commands=False,
+            polls=False,
+            group_management=False,
+            streaming=True,
+            streaming_mode="block",
+            block_streaming=True,
+            block_streaming_chunk_min_chars=self._streaming.MIN_CHARS,
+            block_streaming_chunk_max_chars=self._streaming.MAX_CHARS,
+            block_streaming_chunk_break_preference=self._streaming.CHUNK_BREAK,
+            block_streaming_coalesce_min_chars=self._streaming.COALESCE_MIN,
+            block_streaming_coalesce_max_chars=self._streaming.COALESCE_MAX,
+            block_streaming_coalesce_idle_ms=self._streaming.COALESCE_IDLE_MS,
+            sse_support=False,
+            sse_max_reasoning_chars=0,
+            polling_fallback=True,
+            tts=None,
+        )
+
+    def list_account_ids(self, config: dict | None = None) -> list[str]:
+        return list_confluence_account_ids(config or {})
+
+    def resolve_account(self, account_id: str = "default", config: dict | None = None) -> dict:
+        account = resolve_confluence_account(config or {}, account_id)
+        return {
+            "account_id": account.account_id,
+            "url": account.url,
+            "email": account.email,
+            "space_keys": account.space_keys,
+            "comment_policy": account.comment_policy,
+            "allow_from": account.allow_from,
+            "comment_poll_enabled": account.comment_poll_enabled,
+            "poll_interval": account.poll_interval,
+            "auth_mode": account.auth_mode,
+            "deployment_type": account.deployment_type,
+            "enable_page_writer": account.enable_page_writer,
+            "enable_knowledge_retrieval": account.enable_knowledge_retrieval,
+            "enable_webhook": account.enable_webhook,
+        }
+
+    def is_configured(self, account: dict | None = None) -> bool:
+        if account is None:
+            account = self.resolve_account()
+        return bool(account.get("url") and account.get("email") and account.get("api_token"))
+
+    async def start(self, ctx) -> object:
+        config = getattr(ctx, "config", ctx) if hasattr(ctx, "config") else (ctx if isinstance(ctx, dict) else {})
+        account_id = getattr(ctx, "account_id", "default") if hasattr(ctx, "account_id") else "default"
+        account = resolve_confluence_account(config, account_id)
+
+        self._gateway = ConfluenceGateway()
+        self._status = ConfluenceStatus()
+        self._security = ConfluenceCommentPolicy(
+            policy=account.comment_policy,
+            allow_from=account.allow_from,
+        )
+
+        async def _authorize_sender(author_id: str, _raw: dict) -> bool:
+            return self._security.is_allowed(author_id) if self._security else True
+
+        async def _on_comment(comment: InboundConfluenceComment) -> None:
+            logger.info(
+                "Confluence comment received: %s from %s on page %s",
+                comment.comment_id, comment.author_name,
+                comment.page_context.page_title if comment.page_context else "?"
+            )
+
+        client = await self._gateway.start(
+            account_id=account_id,
+            account=account,
+            on_comment=_on_comment,
+            authorize_sender=_authorize_sender,
+        )
+        self._outbound = ConfluenceOutbound(self._gateway)
+
+        if account.enable_page_writer:
+            self._page_writer = ConfluencePageWriter(self._gateway)
+
+        if account.enable_knowledge_retrieval:
+            self._knowledge = ConfluenceKnowledge(self._gateway)
+
+        if account.enable_webhook:
+            self._webhook = ConfluenceWebhookReceiver(
+                on_page_changed=None,
+                on_comment=_on_comment,
+            )
+            await self._auto_register_webhooks(client, account)
+
+        self._attachment_manager = ConfluenceAttachmentManager(self._gateway)
+        self._label_manager = ConfluenceLabelManager(self._gateway)
+        self._content_property_store = ContentPropertyStore(self._gateway)
+
+        return client
+
+    async def _auto_register_webhooks(self, client, account) -> None:
+        try:
+            existing = await client._request("GET", "/rest/webhooks/1.0/webhook")
+            hooks = existing if isinstance(existing, list) else existing.get("results", [])
+            target_url = f"{account.url.rstrip('/')}/webhooks/confluence"
+            for hook in hooks:
+                if hook.get("url") == target_url:
+                    logger.info("Webhook already registered: %s", hook.get("id"))
+                    return
+            result = await client._request(
+                "POST",
+                "/rest/webhooks/1.0/webhook",
+                json={
+                    "name": "ForcePilot Confluence Integration",
+                    "url": target_url,
+                    "events": list({
+                        "page_created", "page_updated", "page_removed",
+                        "blog_created", "blog_updated",
+                        "comment_created", "comment_updated", "comment_removed",
+                    }),
+                },
+            )
+            logger.info("Webhook registered: %s", result.get("id", "unknown"))
+        except Exception as e:
+            logger.warning("Failed to auto-register webhook: %s", e)
+
+    async def stop(self, ctx) -> None:
+        if self._gateway:
+            await self._gateway.stop_all()
+            self._gateway = None
+        self._outbound = None
+        self._status = None
+        self._security = None
+        self._page_writer = None
+        self._knowledge = None
+        self._webhook = None
+        self._attachment_manager = None
+        self._label_manager = None
+        self._content_property_store = None
+
+    async def on_config_changed(self, prev_cfg: dict, next_cfg: dict, account_id: str) -> None:
+        if prev_cfg != next_cfg:
+            logger.info("Confluence config changed, reloading...")
+            ctx = ChannelContext(channel_type="confluence", account_id=account_id, config=next_cfg)
+            await self.stop(ctx)
+            await self.start(ctx)
+
+    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,
+    ) -> dict:
+        if not content or self._outbound is None:
+            return {"success": False, "error": "Not ready"}
+
+        result = await self._outbound.send_text(
+            target_id=target_id,
+            content=content,
+            reply_to_id=reply_to_id,
+            thread_id=thread_id,
+            account_id=account_id or "default",
+        )
+        return {"success": result.success, "comment_id": result.comment_id, "error": result.error}
+
+    async def edit_message(
+        self,
+        message_id: str,
+        content: str,
+        *,
+        account_id: str | None = None,
+    ) -> dict:
+        if not content or self._outbound is None:
+            return {"success": False, "error": "Not ready"}
+        result = await self._outbound.edit_comment(
+            comment_id=message_id,
+            text=content,
+            account_id=account_id or "default",
+        )
+        return {"success": result.success, "comment_id": result.comment_id, "error": result.error}
+
+    async def delete_message(
+        self,
+        message_id: str,
+        *,
+        account_id: str | None = None,
+    ) -> dict:
+        if self._outbound is None:
+            return {"success": False, "error": "Not ready"}
+        result = await self._outbound.delete_comment(
+            comment_id=message_id,
+            account_id=account_id or "default",
+        )
+        return {"success": result.success, "comment_id": result.comment_id, "error": result.error}
+
+    async def probe(self, account: dict | None = None) -> bool:
+        if self._gateway is None or self._status is None:
+            return False
+        cfg = account or {}
+        aid = cfg.get("account_id", "default")
+        confluence_account = resolve_confluence_account(cfg, aid)
+        client = self._gateway.get_client(aid)
+        if client is None:
+            return False
+        result = await self._status.probe(confluence_account, client)
+        return result.get("success", False)
+
+    def resolve_dm_policy(self) -> dict:
+        if self._security:
+            return {"mode": self._security.policy, "allow_from": list(self._security.allowlist)}
+        return {"mode": "open", "allow_from": []}
+
+    async def check_allowlist(self, peer_id: str, channel_type: str) -> bool:
+        if self._security:
+            return self._security.is_allowed(peer_id)
+        return True
+
+    def extract_mentions(self, raw_message: dict) -> list[dict]:
+        return ConfluenceMentions.extract_mentions(raw_message)
+
+    def build_context_note(self, context) -> str:
+        if context is None:
+            return ""
+        if isinstance(context, PageContext):
+            return context.context_label
+        return str(context)
+
+
+ChannelPluginRegistry.register(ConfluencePlugin())

backend/package/yuxi/channel/extensions/confluence/attachments.py

@@ -0,0 +1,106 @@
+import logging
+from pathlib import Path
+
+from yuxi.channel.extensions.confluence.gateway import ConfluenceGateway
+from yuxi.channel.extensions.confluence.types import OutboundResult
+
+logger = logging.getLogger(__name__)
+
+
+class ConfluenceAttachmentManager:
+    def __init__(self, gateway: ConfluenceGateway):
+        self._gateway = gateway
+
+    async def list_attachments(
+        self, page_id: str, account_id: str = "default",
+    ) -> list[dict]:
+        client = self._gateway.get_client(account_id)
+        if client is None:
+            return []
+        try:
+            result = await client._request(
+                "GET", f"/wiki/api/v2/pages/{page_id}/attachments",
+            )
+            return result.get("results", [])
+        except Exception as e:
+            logger.error("Failed to list attachments for page %s: %s", page_id, e)
+            return []
+
+    async def get_attachment(
+        self, attachment_id: str, account_id: str = "default",
+    ) -> dict | None:
+        client = self._gateway.get_client(account_id)
+        if client is None:
+            return None
+        try:
+            return await client._request(
+                "GET", f"/wiki/api/v2/attachments/{attachment_id}",
+            )
+        except Exception as e:
+            logger.error("Failed to get attachment %s: %s", attachment_id, e)
+            return None
+
+    async def upload_attachment(
+        self,
+        page_id: str,
+        file_path: str,
+        filename: str | None = None,
+        account_id: str = "default",
+    ) -> OutboundResult:
+        client = self._gateway.get_client(account_id)
+        if client is None:
+            return OutboundResult(success=False, error="Client not found")
+
+        path = Path(file_path)
+        if not path.exists():
+            return OutboundResult(success=False, error=f"File not found: {file_path}")
+
+        fname = filename or path.name
+        try:
+            result = await client._request(
+                "POST",
+                f"/wiki/api/v2/pages/{page_id}/attachments",
+                files={"file": (fname, path.read_bytes(), "application/octet-stream")},
+            )
+            return OutboundResult(success=True, comment_id=result.get("id", ""))
+        except Exception as e:
+            logger.error("Failed to upload attachment to page %s: %s", page_id, e)
+            return OutboundResult(success=False, error=str(e))
+
+    async def download_attachment(
+        self,
+        page_id: str,
+        filename: str,
+        save_path: str,
+        account_id: str = "default",
+    ) -> str | None:
+        client = self._gateway.get_client(account_id)
+        if client is None:
+            return None
+
+        try:
+            url = f"/download/attachments/{page_id}/{filename}"
+            if client._http is None:
+                raise RuntimeError("Client not initialized")
+            resp = await client._http.get(url)
+            resp.raise_for_status()
+            Path(save_path).write_bytes(resp.content)
+            return save_path
+        except Exception as e:
+            logger.error("Failed to download attachment %s: %s", filename, e)
+            return None
+
+    async def delete_attachment(
+        self, attachment_id: str, account_id: str = "default",
+    ) -> OutboundResult:
+        client = self._gateway.get_client(account_id)
+        if client is None:
+            return OutboundResult(success=False, error="Client not found")
+        try:
+            await client._request(
+                "DELETE", f"/wiki/api/v2/attachments/{attachment_id}",
+            )
+            return OutboundResult(success=True, comment_id=attachment_id)
+        except Exception as e:
+            logger.error("Failed to delete attachment %s: %s", attachment_id, e)
+            return OutboundResult(success=False, error=str(e))

backend/package/yuxi/channel/extensions/confluence/auth.py

@@ -0,0 +1,49 @@
+import logging
+
+import httpx
+
+logger = logging.getLogger(__name__)
+
+ATLASSIAN_TOKEN_URL = "https://auth.atlassian.com/oauth/token"
+
+
+class ConfluenceOAuth2Flow:
+    @staticmethod
+    async def get_access_token(
+        client_id: str,
+        client_secret: str,
+        code: str,
+        redirect_uri: str,
+    ) -> dict:
+        async with httpx.AsyncClient() as c:
+            resp = await c.post(
+                ATLASSIAN_TOKEN_URL,
+                json={
+                    "grant_type": "authorization_code",
+                    "client_id": client_id,
+                    "client_secret": client_secret,
+                    "code": code,
+                    "redirect_uri": redirect_uri,
+                },
+            )
+            resp.raise_for_status()
+            return resp.json()
+
+    @staticmethod
+    async def refresh_token(
+        client_id: str,
+        client_secret: str,
+        refresh_token: str,
+    ) -> dict:
+        async with httpx.AsyncClient() as c:
+            resp = await c.post(
+                ATLASSIAN_TOKEN_URL,
+                json={
+                    "grant_type": "refresh_token",
+                    "client_id": client_id,
+                    "client_secret": client_secret,
+                    "refresh_token": refresh_token,
+                },
+            )
+            resp.raise_for_status()
+            return resp.json()

backend/package/yuxi/channel/extensions/confluence/config.py

@@ -0,0 +1,70 @@
+from __future__ import annotations
+
+import os
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from yuxi.channel.extensions.confluence.types import ConfluenceAccount
+
+ENV_MAP = {
+    "url": "CONFLUENCE_URL",
+    "email": "CONFLUENCE_EMAIL",
+    "api_token": "CONFLUENCE_API_TOKEN",
+    "oauth_client_id": "CONFLUENCE_OAUTH_CLIENT_ID",
+    "oauth_client_secret": "CONFLUENCE_OAUTH_CLIENT_SECRET",
+    "auth_mode": "CONFLUENCE_AUTH_MODE",
+    "deployment_type": "CONFLUENCE_DEPLOYMENT_TYPE",
+    "space_keys": "CONFLUENCE_SPACE_KEYS",
+    "comment_poll_enabled": "CONFLUENCE_COMMENT_POLL_ENABLED",
+    "poll_interval": "CONFLUENCE_POLL_INTERVAL",
+    "comment_policy": "CONFLUENCE_COMMENT_POLICY",
+    "allow_from": "CONFLUENCE_ALLOW_FROM",
+}
+
+
+def _env_or_config(key: str) -> str | None:
+    env_key = ENV_MAP.get(key, "")
+    if env_key:
+        return os.getenv(env_key)
+    return None
+
+
+def list_confluence_account_ids(config: dict) -> list[str]:
+    accounts = config.get("channels", {}).get("confluence", {}).get("accounts", {})
+    if accounts:
+        return list(accounts.keys())
+    return ["default"]
+
+
+def resolve_confluence_account(config: dict, account_id: str = "default") -> ConfluenceAccount:
+    from yuxi.channel.extensions.confluence.types import ConfluenceAccount
+
+    confluence_cfg = config.get("channels", {}).get("confluence", {})
+    account_cfg = confluence_cfg.get("accounts", {}).get(account_id, confluence_cfg)
+
+    def _get(key: str) -> str | None:
+        val = account_cfg.get(key)
+        if val is not None:
+            return str(val)
+        return _env_or_config(key)
+
+    space_keys_str = _get("space_keys") or ""
+    allow_from_str = _get("allow_from") or ""
+    poll_enabled_str = _get("comment_poll_enabled") or "true"
+
+    return ConfluenceAccount(
+        account_id=account_id,
+        url=_get("url") or "",
+        email=_get("email") or "",
+        api_token=_get("api_token") or "",
+        oauth_client_id=_get("oauth_client_id") or "",
+        oauth_client_secret=_get("oauth_client_secret") or "",
+        auth_mode=_get("auth_mode") or "api_token",
+        deployment_type=_get("deployment_type") or "cloud",
+        space_keys=[k.strip() for k in space_keys_str.split(",") if k.strip()],
+        comment_poll_enabled=poll_enabled_str.lower() != "false",
+        poll_interval=float(_get("poll_interval") or "30"),
+        comment_policy=_get("comment_policy") or "open",
+        allow_from=[a.strip() for a in allow_from_str.split(",") if a.strip()],
+        name=account_cfg.get("name", "Confluence Bot"),
+    )

backend/package/yuxi/channel/extensions/confluence/content_properties.py

@@ -0,0 +1,91 @@
+import logging
+from datetime import datetime, UTC
+
+from yuxi.channel.extensions.confluence.gateway import ConfluenceGateway
+
+logger = logging.getLogger(__name__)
+
+
+class ContentPropertyStore:
+    NAMESPACE = "forcepilot_ai"
+
+    def __init__(self, gateway: ConfluenceGateway):
+        self._gateway = gateway
+
+    async def _get_properties(self, page_id: str, account_id: str = "default") -> list[dict]:
+        client = self._gateway.get_client(account_id)
+        if client is None:
+            return []
+        try:
+            result = await client._request("GET", f"/wiki/api/v2/pages/{page_id}/properties")
+            return result.get("results", [])
+        except Exception as e:
+            logger.error("Failed to get properties for page %s: %s", page_id, e)
+            return []
+
+    async def _set(
+        self, page_id: str, key: str, value: dict, account_id: str = "default",
+    ) -> dict | None:
+        client = self._gateway.get_client(account_id)
+        if client is None:
+            return None
+        try:
+            return await client._request(
+                "POST",
+                f"/wiki/api/v2/pages/{page_id}/properties",
+                json={"key": key, "value": value},
+            )
+        except Exception as e:
+            logger.error("Failed to set property '%s' for page %s: %s", key, page_id, e)
+            return None
+
+    async def _delete(
+        self, page_id: str, property_id: str, account_id: str = "default",
+    ) -> bool:
+        client = self._gateway.get_client(account_id)
+        if client is None:
+            return False
+        try:
+            await client._request(
+                "DELETE", f"/wiki/api/v2/pages/{page_id}/properties/{property_id}",
+            )
+            return True
+        except Exception as e:
+            logger.error("Failed to delete property %s from page %s: %s", property_id, page_id, e)
+            return False
+
+    async def mark_processed(self, page_id: str, processor: str, account_id: str = "default") -> dict | None:
+        return await self._set(
+            page_id,
+            f"{self.NAMESPACE}.{processor}.processed_at",
+            {"timestamp": datetime.now(tz=UTC).isoformat()},
+            account_id,
+        )
+
+    async def store_summary(self, page_id: str, summary: str, account_id: str = "default") -> dict | None:
+        return await self._set(
+            page_id,
+            f"{self.NAMESPACE}.summary",
+            {"text": summary},
+            account_id,
+        )
+
+    async def get_ai_metadata(self, page_id: str, account_id: str = "default") -> dict:
+        properties = await self._get_properties(page_id, account_id)
+        metadata = {}
+        for prop in properties:
+            key = prop.get("key", "")
+            if key.startswith(self.NAMESPACE):
+                metadata[key] = prop.get("value", {})
+        return metadata
+
+    async def clear_ai_metadata(self, page_id: str, account_id: str = "default") -> bool:
+        properties = await self._get_properties(page_id, account_id)
+        success = True
+        for prop in properties:
+            key = prop.get("key", "")
+            prop_id = prop.get("id")
+            if key.startswith(self.NAMESPACE) and prop_id:
+                if not await self._delete(page_id, prop_id, account_id):
+                    success = False
+        return success

backend/package/yuxi/channel/extensions/confluence/dedupe.py

@@ -0,0 +1,34 @@
+import time
+
+
+class CommentDeduplicator:
+    def __init__(self, max_size: int = 50_000, ttl_seconds: int = 600):
+        self._max_size = max_size
+        self._ttl = ttl_seconds
+        self._cache: dict[str, float] = {}
+
+    def is_duplicate(self, comment_id: str) -> bool:
+        self._evict_expired()
+        return comment_id in self._cache
+
+    def mark(self, comment_id: str):
+        self._evict_expired()
+        self._cache[comment_id] = time.time()
+        self._evict_oldest_if_needed()
+
+    def _evict_expired(self):
+        now = time.time()
+        expired = [cid for cid, ts in self._cache.items() if now - ts > self._ttl]
+        for cid in expired:
+            del self._cache[cid]
+
+    def _evict_oldest_if_needed(self):
+        while len(self._cache) > self._max_size:
+            oldest = min(self._cache, key=self._cache.get)
+            del self._cache[oldest]
+
+    def clear(self):
+        self._cache.clear()
+
+    def __len__(self) -> int:
+        return len(self._cache)

backend/package/yuxi/channel/extensions/confluence/format.py

@@ -0,0 +1,426 @@
+import json
+import re
+from datetime import datetime
+
+
+class ADFBuilder:
+    PANEL_TYPES = {"info", "note", "success", "warning", "error", "tip"}
+    STATUS_COLORS = {"neutral", "purple", "blue", "red", "yellow", "green"}
+
+    @staticmethod
+    def doc(blocks: list[dict]) -> str:
+        return json.dumps(
+            {
+                "version": 1,
+                "type": "doc",
+                "content": blocks,
+            },
+            ensure_ascii=False,
+        )
+
+    @classmethod
+    def paragraph(cls, text: str) -> dict:
+        return {
+            "type": "paragraph",
+            "content": cls._parse_inline(text),
+        }
+
+    @classmethod
+    def heading(cls, text: str, level: int = 3) -> dict:
+        return {
+            "type": "heading",
+            "attrs": {"level": min(max(level, 1), 6)},
+            "content": cls._parse_inline(text),
+        }
+
+    @staticmethod
+    def code_block(code: str, language: str = "python") -> dict:
+        return {
+            "type": "codeBlock",
+            "attrs": {"language": language},
+            "content": [{"type": "text", "text": code}],
+        }
+
+    @classmethod
+    def bullet_list(cls, items: list[str]) -> dict:
+        return {
+            "type": "bulletList",
+            "content": [
+                {
+                    "type": "listItem",
+                    "content": [cls.paragraph(item)],
+                }
+                for item in items
+            ],
+        }
+
+    @classmethod
+    def ordered_list(cls, items: list[str]) -> dict:
+        return {
+            "type": "orderedList",
+            "content": [
+                {
+                    "type": "listItem",
+                    "content": [cls.paragraph(item)],
+                }
+                for item in items
+            ],
+        }
+
+    @staticmethod
+    def mention(account_id: str, display_name: str) -> dict:
+        return {
+            "type": "mention",
+            "attrs": {
+                "id": account_id,
+                "text": f"@{display_name}",
+                "accessLevel": "",
+                "userType": "DEFAULT",
+            },
+        }
+
+    @staticmethod
+    def link(url: str, text: str | None = None) -> dict:
+        return {
+            "type": "text",
+            "text": text or url,
+            "marks": [{"type": "link", "attrs": {"href": url}}],
+        }
+
+    @staticmethod
+    def text_with_marks(text: str, marks: list[dict] | None = None) -> dict:
+        node: dict = {"type": "text", "text": text}
+        if marks:
+            node["marks"] = marks
+        return node
+
+    @staticmethod
+    def table(headers: list[str], rows: list[list[str]]) -> dict:
+        return {
+            "type": "table",
+            "attrs": {"isNumberColumnEnabled": False, "layout": "default"},
+            "content": [
+                {
+                    "type": "tableRow",
+                    "content": [
+                        {
+                            "type": "tableHeader",
+                            "attrs": {},
+                            "content": [{"type": "paragraph", "content": [{"type": "text", "text": h}]}],
+                        }
+                        for h in headers
+                    ],
+                }
+            ]
+            + [
+                {
+                    "type": "tableRow",
+                    "content": [
+                        {
+                            "type": "tableCell",
+                            "attrs": {},
+                            "content": [{"type": "paragraph", "content": [{"type": "text", "text": cell}]}],
+                        }
+                        for cell in row
+                    ],
+                }
+                for row in rows
+            ],
+        }
+
+    @classmethod
+    def panel(cls, content: list[dict], panel_type: str = "info") -> dict:
+        return {
+            "type": "panel",
+            "attrs": {"panelType": panel_type if panel_type in cls.PANEL_TYPES else "info"},
+            "content": content,
+        }
+
+    @classmethod
+    def info_panel(cls, text: str) -> dict:
+        return cls.panel(
+            [{"type": "paragraph", "content": [{"type": "text", "text": text}]}],
+            "info",
+        )
+
+    @staticmethod
+    def expand(title: str, content: list[dict]) -> dict:
+        return {
+            "type": "expand",
+            "attrs": {"title": title},
+            "content": content,
+        }
+
+    @staticmethod
+    def task_list(tasks: list[dict]) -> dict:
+        return {
+            "type": "taskList",
+            "attrs": {},
+            "content": [
+                {
+                    "type": "taskItem",
+                    "attrs": {
+                        "localId": task.get("id", f"task-{i}"),
+                        "state": task.get("state", "TODO"),
+                    },
+                    "content": [
+                        {"type": "text", "text": task.get("text", "")},
+                    ],
+                }
+                for i, task in enumerate(tasks)
+            ],
+        }
+
+    @staticmethod
+    def blockquote(content: list[dict]) -> dict:
+        return {"type": "blockquote", "content": content}
+
+    @staticmethod
+    def rule() -> dict:
+        return {"type": "rule"}
+
+    @classmethod
+    def status(cls, text: str, color: str = "neutral") -> dict:
+        return {
+            "type": "status",
+            "attrs": {
+                "text": text,
+                "color": color if color in cls.STATUS_COLORS else "neutral",
+            },
+        }
+
+    @staticmethod
+    def date(iso_date: str) -> dict:
+        ts = int(datetime.fromisoformat(iso_date).timestamp() * 1000)
+        return {
+            "type": "date",
+            "attrs": {"timestamp": ts},
+        }
+
+    @staticmethod
+    def media(media_id: str, media_type: str = "file", collection: str = "") -> dict:
+        return {
+            "type": "mediaSingle",
+            "attrs": {"layout": "center"},
+            "content": [
+                {
+                    "type": "media",
+                    "attrs": {
+                        "type": media_type,
+                        "id": media_id,
+                        "collection": collection,
+                    },
+                }
+            ],
+        }
+
+    @staticmethod
+    def emoji(short_name: str, id: str | None = None, text: str | None = None) -> dict:
+        attrs: dict = {"shortName": short_name}
+        if id:
+            attrs["id"] = id
+        if text:
+            attrs["text"] = text
+        return {
+            "type": "emoji",
+            "attrs": attrs,
+        }
+
+    @staticmethod
+    def macro(name: str, params: dict | None = None) -> dict:
+        macro_params: dict = {}
+        if params:
+            for key, value in params.items():
+                macro_params[key] = {"value": value}
+        return {
+            "type": "extension",
+            "attrs": {
+                "extensionType": "com.atlassian.confluence.macro.core",
+                "extensionKey": name,
+                "parameters": {"macroParams": macro_params},
+            },
+        }
+
+    @staticmethod
+    def layout_section(columns: list[list[dict]], column_widths: list[int] | None = None) -> dict:
+        if column_widths is None:
+            column_widths = [100 // len(columns)] * len(columns)
+        return {
+            "type": "layoutSection",
+            "attrs": {"layoutType": "two_equal" if len(columns) == 2 else "default"},
+            "content": [
+                {
+                    "type": "layoutColumn",
+                    "attrs": {"width": column_widths[i]},
+                    "content": col,
+                }
+                for i, col in enumerate(columns)
+            ],
+        }
+
+    @staticmethod
+    def _parse_inline(text: str) -> list[dict]:
+        pattern = re.compile(
+            r'(\*\*(.+?)\*\*|\*(.+?)\*|`(.+?)`|\[(.+?)\]\((.+?)\))'
+        )
+        nodes = []
+        last_end = 0
+        for m in pattern.finditer(text):
+            if m.start() > last_end:
+                plain = text[last_end:m.start()]
+                if plain:
+                    nodes.append({"type": "text", "text": plain})
+            last_end = m.end()
+
+            if m.group(2):
+                nodes.append({"type": "text", "text": m.group(2), "marks": [{"type": "strong"}]})
+            elif m.group(3):
+                nodes.append({"type": "text", "text": m.group(3), "marks": [{"type": "em"}]})
+            elif m.group(4):
+                nodes.append({"type": "text", "text": m.group(4), "marks": [{"type": "code"}]})
+            elif m.group(5) and m.group(6):
+                nodes.append({
+                    "type": "text",
+                    "text": m.group(5),
+                    "marks": [{"type": "link", "attrs": {"href": m.group(6)}}],
+                })
+
+        if last_end < len(text):
+            nodes.append({"type": "text", "text": text[last_end:]})
+
+        return nodes or [{"type": "text", "text": text}]
+
+    @classmethod
+    def from_markdown(cls, md_text: str) -> str:
+        blocks = []
+        lines = md_text.split("\n")
+        i = 0
+
+        while i < len(lines):
+            stripped = lines[i].strip()
+
+            if not stripped:
+                i += 1
+                continue
+
+            heading_match = re.match(r"^(#{1,6})\s+(.+)", stripped)
+            if heading_match:
+                level = len(heading_match.group(1))
+                blocks.append(cls.heading(heading_match.group(2), level))
+                i += 1
+                continue
+
+            if stripped.startswith("```"):
+                lang = stripped[3:].strip() or "text"
+                code_lines = []
+                i += 1
+                while i < len(lines) and not lines[i].strip().startswith("```"):
+                    code_lines.append(lines[i])
+                    i += 1
+                i += 1
+                blocks.append(cls.code_block("\n".join(code_lines), lang))
+                continue
+
+            if stripped.startswith("- ") or stripped.startswith("* "):
+                items = []
+                while i < len(lines) and (lines[i].strip().startswith("- ") or lines[i].strip().startswith("* ")):
+                    items.append(lines[i].strip()[2:])
+                    i += 1
+                blocks.append(cls.bullet_list(items))
+                continue
+
+            ordered_match = re.match(r"^(\d+)\.\s+(.+)", stripped)
+            if ordered_match:
+                items = []
+                while i < len(lines):
+                    m = re.match(r"^(\d+)\.\s+(.+)", lines[i].strip())
+                    if not m:
+                        break
+                    items.append(m.group(2))
+                    i += 1
+                blocks.append(cls.ordered_list(items))
+                continue
+
+            if stripped.startswith("> "):
+                quote_lines = []
+                while i < len(lines) and lines[i].strip().startswith("> "):
+                    quote_lines.append(lines[i].strip()[2:])
+                    i += 1
+                blocks.append(cls.blockquote([cls.paragraph("\n".join(quote_lines))]))
+                continue
+
+            if stripped.startswith("---") or stripped.startswith("***"):
+                blocks.append(cls.rule())
+                i += 1
+                continue
+
+            table_match = re.match(r"^\|(.+)\|$", stripped)
+            if table_match and i + 1 < len(lines) and re.match(r"^\|[\s\-:|]+\|$", lines[i + 1].strip()):
+                headers = [h.strip() for h in table_match.group(1).split("|") if h.strip() or h.strip() == ""]
+                rows = []
+                i += 2
+                while i < len(lines):
+                    row_match = re.match(r"^\|(.+)\|$", lines[i].strip())
+                    if not row_match:
+                        break
+                    cells = [c.strip() for c in row_match.group(1).split("|")]
+                    rows.append(cells)
+                    i += 1
+                blocks.append(cls.table(headers, rows))
+                continue
+
+            blocks.append(cls.paragraph(stripped))
+            i += 1
+
+        return cls.doc(blocks)
+
+
+def extract_text_from_adf(adf_json: str) -> str:
+    try:
+        doc = json.loads(adf_json) if isinstance(adf_json, str) else adf_json
+    except (json.JSONDecodeError, TypeError):
+        return ""
+
+    parts = []
+
+    def _walk(node):
+        if isinstance(node, dict):
+            if node.get("type") == "text":
+                parts.append(node.get("text", ""))
+            elif node.get("type") == "mention":
+                parts.append(node.get("attrs", {}).get("text", "@unknown"))
+            elif node.get("type") == "hardBreak":
+                parts.append("\n")
+            for child in node.get("content", []):
+                _walk(child)
+        elif isinstance(node, list):
+            for item in node:
+                _walk(item)
+
+    _walk(doc)
+    return "".join(parts)
+
+
+ADF_REQUIRED_KEYS = {"version", "type", "content"}
+
+
+def validate_adf(adf_json: str) -> tuple[bool, str]:
+    try:
+        doc = json.loads(adf_json) if isinstance(adf_json, str) else adf_json
+    except (json.JSONDecodeError, TypeError) as e:
+        return False, f"Invalid JSON: {e}"
+
+    if not isinstance(doc, dict):
+        return False, "ADF root must be a dict"
+
+    missing = ADF_REQUIRED_KEYS - set(doc.keys())
+    if missing:
+        return False, f"Missing required keys: {missing}"
+
+    if doc.get("type") != "doc":
+        return False, f"Root type must be 'doc', got '{doc.get('type')}'"
+
+    if doc.get("version") != 1:
+        return False, f"ADF version must be 1, got {doc.get('version')}"
+
+    return True, "OK"

backend/package/yuxi/channel/extensions/confluence/gateway.py

@@ -0,0 +1,492 @@
+import asyncio
+import logging
+import re
+import time
+from collections.abc import Callable
+from datetime import datetime, UTC
+
+import httpx
+
+from yuxi.channel.extensions.confluence.dedupe import CommentDeduplicator
+from yuxi.channel.extensions.confluence.search import CQLBuilder
+from yuxi.channel.extensions.confluence.types import (
+    ConfluenceAccount,
+    InboundConfluenceComment,
+    PageContext,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class ConfluenceClient:
+    def __init__(self, account: ConfluenceAccount):
+        self._account = account
+        self._http: httpx.AsyncClient | None = None
+        self._username = account.email
+        self._rate_limit_remaining = 65_000
+
+    async def __aenter__(self):
+        auth_header = self._account.auth_header
+        self._http = httpx.AsyncClient(
+            base_url=self._account.base_url,
+            headers={
+                **auth_header,
+                "Accept": "application/json",
+                "Content-Type": "application/json",
+            },
+            timeout=httpx.Timeout(30.0),
+        )
+        return self
+
+    async def __aexit__(self, *args):
+        if self._http:
+            await self._http.aclose()
+
+    async def _request(self, method: str, path: str, **kwargs) -> dict:
+        if self._http is None:
+            raise RuntimeError("Client not initialized")
+
+        if self._rate_limit_remaining < 100:
+            await asyncio.sleep(0.5)
+        elif self._rate_limit_remaining < 500:
+            await asyncio.sleep(0.1)
+
+        resp = await self._http.request(method, path, **kwargs)
+        self._update_rate_limit(resp.headers)
+
+        if resp.status_code == 429:
+            retry_after = int(resp.headers.get("Retry-After", "5"))
+            logger.warning("Confluence rate limited, retry after %ds", retry_after)
+            await asyncio.sleep(retry_after)
+            resp = await self._http.request(method, path, **kwargs)
+            self._update_rate_limit(resp.headers)
+
+        resp.raise_for_status()
+        return resp.json()
+
+    def _update_rate_limit(self, headers: dict):
+        remaining = headers.get("X-RateLimit-Remaining")
+        if remaining is not None:
+            self._rate_limit_remaining = int(remaining)
+
+    async def cql_search(self, cql: str, limit: int = 50, expand: str | None = None) -> dict:
+        params = {"cql": cql, "limit": limit}
+        if expand:
+            params["expand"] = expand
+        return await self._request("GET", "/rest/api/search", params=params)
+
+    async def create_footer_comment(self, payload: dict) -> dict:
+        return await self._request("POST", "/wiki/api/v2/footer-comments", json=payload)
+
+    async def get_comment_children(self, comment_id: str, limit: int = 25) -> dict:
+        return await self._request(
+            "GET",
+            f"/wiki/api/v2/footer-comments/{comment_id}/children",
+            params={"limit": limit},
+        )
+
+    async def get_page(self, page_id: str, body_format: str = "atlas_doc_format") -> dict:
+        return await self._request(
+            "GET",
+            f"/wiki/api/v2/pages/{page_id}",
+            params={"body-format": body_format},
+        )
+
+    async def update_page(self, page_id: str, payload: dict) -> dict:
+        return await self._request("PUT", f"/wiki/api/v2/pages/{page_id}", json=payload)
+
+    async def get_space(self, space_id: str) -> dict:
+        return await self._request("GET", f"/wiki/api/v2/spaces/{space_id}")
+
+    async def create_page(self, payload: dict) -> dict:
+        return await self._request("POST", "/wiki/api/v2/pages", json=payload)
+
+    async def update_footer_comment(self, comment_id: str, adf_body: str) -> dict:
+        return await self._request(
+            "PUT",
+            f"/wiki/api/v2/footer-comments/{comment_id}",
+            json={
+                "body": {
+                    "representation": "atlas_doc_format",
+                    "value": adf_body,
+                },
+            },
+        )
+
+    async def delete_footer_comment(self, comment_id: str) -> None:
+        await self._request("DELETE", f"/wiki/api/v2/footer-comments/{comment_id}")
+
+    async def get_footer_comments(
+        self,
+        page_id: str,
+        cursor: str | None = None,
+        limit: int = 25,
+        sort: str = "created-date",
+    ) -> dict:
+        params = {"limit": limit, "sort": sort}
+        if cursor:
+            params["cursor"] = cursor
+        return await self._request(
+            "GET",
+            f"/wiki/api/v2/pages/{page_id}/footer-comments",
+            params=params,
+        )
+
+    async def get_comment(self, comment_id: str) -> dict:
+        return await self._request("GET", f"/wiki/api/v2/footer-comments/{comment_id}")
+
+    async def delete_page(self, page_id: str, purge: bool = False) -> None:
+        params = {"purge": str(purge).lower()}
+        await self._request("DELETE", f"/wiki/api/v2/pages/{page_id}", params=params)
+
+    async def get_space_pages(
+        self,
+        space_id: str,
+        cursor: str | None = None,
+        limit: int = 25,
+        depth: str = "all",
+    ) -> dict:
+        params = {"limit": limit, "depth": depth}
+        if cursor:
+            params["cursor"] = cursor
+        return await self._request(
+            "GET",
+            f"/wiki/api/v2/spaces/{space_id}/pages",
+            params=params,
+        )
+
+    async def get_spaces(self, limit: int = 10, cursor: str | None = None) -> dict:
+        params = {"limit": limit}
+        if cursor:
+            params["cursor"] = cursor
+        return await self._request("GET", "/wiki/api/v2/spaces", params=params)
+
+    async def get_users(
+        self, query: str | None = None, limit: int = 25,
+    ) -> dict:
+        params = {"limit": limit}
+        if query:
+            params["query"] = query
+        return await self._request("GET", "/wiki/api/v2/users", params=params)
+
+    async def create_space(self, key: str, name: str, description: str = "") -> dict:
+        payload = {
+            "key": key,
+            "name": name,
+        }
+        if description:
+            payload["description"] = {"plain": {"value": description}}
+        return await self._request("POST", "/wiki/api/v2/spaces", json=payload)
+
+    async def delete_space(self, space_id: str) -> None:
+        await self._request("DELETE", f"/wiki/api/v2/spaces/{space_id}")
+
+    async def get_page_children(self, page_id: str, limit: int = 25) -> dict:
+        return await self._request(
+            "GET",
+            f"/wiki/api/v2/pages/{page_id}/children",
+            params={"limit": limit},
+        )
+
+    async def get_page_ancestors(self, page_id: str) -> dict:
+        return await self._request("GET", f"/wiki/api/v2/pages/{page_id}/ancestors")
+
+    async def get_page_versions(self, page_id: str, limit: int = 25) -> dict:
+        return await self._request(
+            "GET",
+            f"/wiki/api/v2/pages/{page_id}/versions",
+            params={"limit": limit},
+        )
+
+    async def list_all_footer_comments(
+        self,
+        cursor: str | None = None,
+        limit: int = 25,
+        sort: str = "created-date",
+    ) -> dict:
+        params = {"limit": limit, "sort": sort}
+        if cursor:
+            params["cursor"] = cursor
+        return await self._request("GET", "/wiki/api/v2/footer-comments", params=params)
+
+    async def update_page_title(self, page_id: str, title: str) -> dict:
+        return await self._request(
+            "PUT",
+            f"/wiki/api/v2/pages/{page_id}/title",
+            json={"title": title},
+        )
+
+    async def update_space(self, space_id: str, name: str, description: str = "") -> dict:
+        payload: dict = {"name": name}
+        if description:
+            payload["description"] = {"plain": {"value": description}}
+        return await self._request(
+            "PUT", f"/wiki/api/v2/spaces/{space_id}", json=payload,
+        )
+
+    async def get_blogpost(self, blogpost_id: str, body_format: str = "atlas_doc_format") -> dict:
+        return await self._request(
+            "GET",
+            f"/wiki/api/v2/blogposts/{blogpost_id}",
+            params={"body-format": body_format},
+        )
+
+    async def create_blogpost(self, payload: dict) -> dict:
+        return await self._request("POST", "/wiki/api/v2/blogposts", json=payload)
+
+    async def update_blogpost(self, blogpost_id: str, payload: dict) -> dict:
+        return await self._request("PUT", f"/wiki/api/v2/blogposts/{blogpost_id}", json=payload)
+
+    async def delete_blogpost(self, blogpost_id: str) -> None:
+        await self._request("DELETE", f"/wiki/api/v2/blogposts/{blogpost_id}")
+
+    async def get_tasks(self, page_id: str, limit: int = 25) -> dict:
+        return await self._request(
+            "GET",
+            f"/wiki/api/v2/pages/{page_id}/tasks",
+            params={"limit": limit},
+        )
+
+    async def get_task(self, task_id: str) -> dict:
+        return await self._request("GET", f"/wiki/api/v2/tasks/{task_id}")
+
+    async def update_task(self, task_id: str, state: str = "COMPLETE") -> dict:
+        return await self._request(
+            "PUT",
+            f"/wiki/api/v2/tasks/{task_id}",
+            json={"state": state},
+        )
+
+    async def _paginated_request(
+        self,
+        method: str,
+        path: str,
+        params: dict | None = None,
+        max_results: int = 250,
+    ) -> list[dict]:
+        cursor = None
+        all_results: list[dict] = []
+        params = params or {}
+        params.setdefault("limit", 25)
+
+        while len(all_results) < max_results:
+            if cursor:
+                params["cursor"] = cursor
+
+            if self._http is None:
+                raise RuntimeError("Client not initialized")
+            resp = await self._http.request(method, path, params=params)
+            self._update_rate_limit(resp.headers)
+            resp.raise_for_status()
+
+            page = resp.json()
+            all_results.extend(page.get("results", []))
+
+            links = resp.headers.get("Link", "")
+            cursor = self._extract_next_cursor(links)
+            if not cursor:
+                break
+
+        return all_results[:max_results]
+
+    @staticmethod
+    def _extract_next_cursor(link_header: str) -> str | None:
+        match = re.search(r'<[^>]*cursor=([^>&]+)[^>]*>\s*;\s*rel="next"', link_header)
+        if match:
+            return match.group(1)
+        return None
+
+
+class ConfluenceGateway:
+    MAX_RETRY_BACKOFF = 300.0
+    MAX_SEEN_COMMENTS = 50_000
+
+    def __init__(self):
+        self._active_clients: dict[str, ConfluenceClient] = {}
+        self._poll_tasks: dict[str, asyncio.Task] = {}
+        self._last_poll_time: dict[str, float] = {}
+        self._deduplicator = CommentDeduplicator(max_size=50_000, ttl_seconds=600)
+        self._cancel_events: dict[str, asyncio.Event] = {}
+
+    async def start(
+        self,
+        account_id: str,
+        account: ConfluenceAccount,
+        on_comment: Callable[[InboundConfluenceComment], None],
+        authorize_sender: Callable[[str, dict], bool],
+    ) -> ConfluenceClient:
+        client = ConfluenceClient(account)
+        await client.__aenter__()
+        self._active_clients[account_id] = client
+
+        if account.comment_poll_enabled:
+            cancel = asyncio.Event()
+            self._cancel_events[account_id] = cancel
+            task = asyncio.create_task(
+                self._poll_comments_loop(account_id, account, client, on_comment, authorize_sender, cancel)
+            )
+            self._poll_tasks[account_id] = task
+
+        return client
+
+    async def stop(self, account_id: str):
+        cancel = self._cancel_events.pop(account_id, None)
+        if cancel:
+            cancel.set()
+        task = self._poll_tasks.pop(account_id, None)
+        if task:
+            task.cancel()
+            try:
+                await task
+            except asyncio.CancelledError:
+                pass
+        client = self._active_clients.pop(account_id, None)
+        if client:
+            await client.__aexit__(None, None, None)
+
+    async def stop_all(self):
+        for account_id in list(self._active_clients.keys()):
+            await self.stop(account_id)
+
+    def get_client(self, account_id: str = "default") -> ConfluenceClient | None:
+        return self._active_clients.get(account_id)
+
+    async def _poll_comments_loop(
+        self,
+        account_id: str,
+        account: ConfluenceAccount,
+        client: ConfluenceClient,
+        on_comment: Callable,
+        authorize_sender: Callable,
+        cancel: asyncio.Event,
+    ):
+        error_count = 0
+        last_poll = self._last_poll_time.get(account_id, time.time() - 3600)
+        seen_in_session: set[str] = set()
+
+        while not cancel.is_set():
+            try:
+                since_iso = datetime.fromtimestamp(last_poll, tz=UTC).strftime("%Y-%m-%dT%H:%M:%S.000Z")
+
+                cql = CQLBuilder.comment_poll(since_iso, account.space_keys)
+                results = await client.cql_search(cql, limit=50, expand="body.view,history.lastUpdated,container")
+                new_max_time = last_poll
+
+                for result in results.get("results", []):
+                    content = result.get("content", {})
+                    comment_id = content.get("id", "")
+
+                    if not comment_id:
+                        continue
+                    if comment_id in seen_in_session:
+                        continue
+                    if self._deduplicator.is_duplicate(comment_id):
+                        continue
+                    seen_in_session.add(comment_id)
+                    self._deduplicator.mark(comment_id)
+
+                    author_info = content.get("history", {}).get("createdBy", {})
+                    author_id = author_info.get("accountId", "")
+
+                    if self._is_self_comment(client, author_info):
+                        continue
+
+                    if not authorize_sender(author_id, result):
+                        continue
+
+                    page_ctx = self._extract_page_context(result)
+                    if page_ctx is None:
+                        continue
+
+                    comment_text = self._extract_comment_text(result)
+
+                    inbound = InboundConfluenceComment(
+                        comment_id=comment_id,
+                        author_id=author_id,
+                        author_name=author_info.get("displayName", ""),
+                        body_text=comment_text,
+                        container_id=page_ctx.page_id,
+                        parent_comment_id=self._extract_parent_comment_id(result),
+                        page_context=page_ctx,
+                        created_at=self._parse_datetime(content.get("history", {}).get("createdDate")),
+                    )
+
+                    on_comment(inbound)
+
+                    updated = content.get("history", {}).get("lastUpdated", {}).get("when", "")
+                    updated_ts = self._parse_timestamp(updated)
+                    if updated_ts and updated_ts > new_max_time:
+                        new_max_time = updated_ts
+
+                last_poll = new_max_time
+                self._last_poll_time[account_id] = new_max_time
+                error_count = 0
+
+            except asyncio.CancelledError:
+                break
+            except Exception as e:
+                error_count += 1
+                backoff = min(2.0 * (2**error_count), self.MAX_RETRY_BACKOFF)
+                logger.error(
+                    "Confluence poll error (attempt %d, retry in %.1fs): %s",
+                    error_count,
+                    backoff,
+                    e,
+                )
+
+            try:
+                await asyncio.wait_for(cancel.wait(), timeout=account.poll_interval)
+                break
+            except TimeoutError:
+                pass
+
+    def _is_self_comment(self, client: ConfluenceClient, author: dict) -> bool:
+        author_email = author.get("email", "")
+        return author_email and author_email.lower() == client._username.lower()
+
+    def _extract_page_context(self, result: dict) -> PageContext | None:
+        container = result.get("resultParentContainer")
+        if not container:
+            return None
+        return PageContext(
+            page_id=container.get("id", ""),
+            page_title=container.get("title", ""),
+            space_key=container.get("space", {}).get("key", ""),
+        )
+
+    def _extract_comment_text(self, result: dict) -> str:
+        body = result.get("content", {}).get("body", {})
+        view_value = body.get("view", {}).get("value", "")
+        return self._strip_html(view_value)
+
+    def _extract_parent_comment_id(self, result: dict) -> str | None:
+        ancestors = result.get("content", {}).get("ancestors", [])
+        for ancestor in reversed(ancestors):
+            if ancestor.get("type") == "comment":
+                return ancestor.get("id")
+        return None
+
+    @staticmethod
+    def _strip_html(html: str) -> str:
+        clean = re.sub(r"<[^>]+>", "", html)
+        clean = re.sub(r"&[a-z]+;", " ", clean)
+        return re.sub(r"\s+", " ", clean).strip()
+
+    @staticmethod
+    def _parse_timestamp(iso_str: str | None) -> float:
+        if not iso_str:
+            return 0.0
+        try:
+            dt = datetime.fromisoformat(iso_str.replace("Z", "+00:00"))
+            return dt.timestamp()
+        except (ValueError, TypeError):
+            return 0.0
+
+    @staticmethod
+    def _parse_datetime(iso_str: str | None) -> datetime | None:
+        if not iso_str:
+            return None
+        try:
+            return datetime.fromisoformat(iso_str.replace("Z", "+00:00"))
+        except (ValueError, TypeError):
+            return None

backend/package/yuxi/channel/extensions/confluence/knowledge.py

@@ -0,0 +1,68 @@
+import logging
+
+from yuxi.channel.extensions.confluence.format import extract_text_from_adf
+from yuxi.channel.extensions.confluence.gateway import ConfluenceGateway
+from yuxi.channel.extensions.confluence.search import CQLBuilder
+
+logger = logging.getLogger(__name__)
+
+
+class ConfluenceKnowledge:
+    def __init__(self, gateway: ConfluenceGateway):
+        self._gateway = gateway
+
+    async def retrieve_context(
+        self,
+        query: str,
+        account_id: str = "default",
+        space_keys: list[str] | None = None,
+        max_pages: int = 5,
+    ) -> list[dict]:
+        client = self._gateway.get_client(account_id)
+        if client is None:
+            return []
+
+        cql = CQLBuilder.search_pages(query, space_keys)
+
+        try:
+            results = await client.cql_search(cql, limit=max_pages)
+        except Exception as e:
+            logger.error("CQL search failed for context retrieval '%s': %s", query, e)
+            return []
+
+        contexts = []
+        for item in results.get("results", []):
+            content = item.get("content", {})
+            page_id = content.get("id")
+            title = content.get("title", "")
+            body = content.get("body", {})
+            adf_value = body.get("view", {}).get("value", "")
+            text = extract_text_from_adf(adf_value) if adf_value else ""
+
+            contexts.append({
+                "page_id": page_id,
+                "title": title,
+                "text": text[:2000],
+                "url": content.get("_links", {}).get("base", ""),
+            })
+
+        return contexts
+
+    async def build_context_note(
+        self,
+        page_context,
+        account_id: str = "default",
+    ) -> str:
+        if page_context is None:
+            return ""
+
+        client = self._gateway.get_client(account_id)
+        if client is None:
+            return ""
+
+        try:
+            page = await client.get_page(page_context.page_id)
+            title = page.get("title", "")
+            return f"当前页面: [{page_context.space_key}] {title}"
+        except Exception:
+            return f"当前页面: {page_context.context_label}"

backend/package/yuxi/channel/extensions/confluence/labels.py

@@ -0,0 +1,69 @@
+import logging
+
+from yuxi.channel.extensions.confluence.gateway import ConfluenceGateway
+from yuxi.channel.extensions.confluence.types import OutboundResult
+
+logger = logging.getLogger(__name__)
+
+
+class ConfluenceLabelManager:
+    def __init__(self, gateway: ConfluenceGateway):
+        self._gateway = gateway
+
+    async def get_page_labels(self, page_id: str, account_id: str = "default") -> list[dict]:
+        client = self._gateway.get_client(account_id)
+        if client is None:
+            return []
+        try:
+            result = await client._request("GET", f"/wiki/api/v2/pages/{page_id}/labels")
+            return result.get("results", [])
+        except Exception as e:
+            logger.error("Failed to get labels for page %s: %s", page_id, e)
+            return []
+
+    async def add_page_label(
+        self, page_id: str, label: str, prefix: str = "global", account_id: str = "default",
+    ) -> OutboundResult:
+        client = self._gateway.get_client(account_id)
+        if client is None:
+            return OutboundResult(success=False, error="Client not found")
+        try:
+            result = await client._request(
+                "POST",
+                f"/wiki/api/v2/pages/{page_id}/labels",
+                json={"prefix": prefix, "name": label},
+            )
+            return OutboundResult(success=True, comment_id=result.get("id", ""))
+        except Exception as e:
+            logger.error("Failed to add label '%s' to page %s: %s", label, page_id, e)
+            return OutboundResult(success=False, error=str(e))
+
+    async def delete_page_label(
+        self, page_id: str, label_id: str, account_id: str = "default",
+    ) -> OutboundResult:
+        client = self._gateway.get_client(account_id)
+        if client is None:
+            return OutboundResult(success=False, error="Client not found")
+        try:
+            await client._request(
+                "DELETE", f"/wiki/api/v2/pages/{page_id}/labels/{label_id}",
+            )
+            return OutboundResult(success=True, comment_id=label_id)
+        except Exception as e:
+            logger.error("Failed to delete label %s from page %s: %s", label_id, page_id, e)
+            return OutboundResult(success=False, error=str(e))
+
+    async def get_label_pages(
+        self, label_id: str, limit: int = 25, account_id: str = "default",
+    ) -> list[dict]:
+        client = self._gateway.get_client(account_id)
+        if client is None:
+            return []
+        try:
+            result = await client._request(
+                "GET", f"/wiki/api/v2/labels/{label_id}/pages", params={"limit": limit},
+            )
+            return result.get("results", [])
+        except Exception as e:
+            logger.error("Failed to get pages for label %s: %s", label_id, e)
+            return []

backend/package/yuxi/channel/extensions/confluence/mentions.py

@@ -0,0 +1,38 @@
+import json
+
+
+class ConfluenceMentions:
+    @staticmethod
+    def extract_mentions(raw_message: dict) -> list[dict]:
+        body = raw_message.get("body", {})
+        adf_raw = body.get("atlas_doc_format", {}).get("value", "")
+
+        try:
+            doc = json.loads(adf_raw) if isinstance(adf_raw, str) else adf_raw
+        except (json.JSONDecodeError, TypeError):
+            return []
+
+        mentions = []
+
+        def _walk(node):
+            if isinstance(node, dict):
+                if node.get("type") == "mention":
+                    attrs = node.get("attrs", {})
+                    mentions.append({
+                        "account_id": attrs.get("id", ""),
+                        "display_name": attrs.get("text", "").lstrip("@"),
+                    })
+                for child in node.get("content", []):
+                    _walk(child)
+            elif isinstance(node, list):
+                for item in node:
+                    _walk(item)
+
+        _walk(doc)
+        return mentions
+
+    @staticmethod
+    def build_mention(account_id: str, display_name: str) -> dict:
+        from yuxi.channel.extensions.confluence.format import ADFBuilder
+
+        return ADFBuilder.mention(account_id, display_name)

backend/package/yuxi/channel/extensions/confluence/outbound.py

@@ -0,0 +1,280 @@
+import asyncio
+import logging
+
+from yuxi.channel.extensions.confluence.format import ADFBuilder
+from yuxi.channel.extensions.confluence.gateway import ConfluenceGateway
+from yuxi.channel.extensions.confluence.search import CQLBuilder
+from yuxi.channel.extensions.confluence.types import OutboundResult
+
+logger = logging.getLogger(__name__)
+
+MAX_ADF_BLOCK_CHARS = 2000
+MAX_RETRY = 3
+RETRY_BASE_DELAY = 1.0
+
+
+class ConfluenceOutbound:
+    delivery_mode = "comment"
+
+    def __init__(self, gateway: ConfluenceGateway):
+        self._gateway = gateway
+
+    async def reply_to_comment(
+        self,
+        page_id: str,
+        parent_comment_id: str,
+        text: str,
+        account_id: str = "default",
+    ) -> OutboundResult:
+        client = self._gateway.get_client(account_id)
+        if client is None:
+            return OutboundResult(success=False, error="Client not found")
+
+        adf_body = ADFBuilder.from_markdown(text)
+
+        payload = {
+            "body": {
+                "representation": "atlas_doc_format",
+                "value": adf_body,
+            },
+            "containerId": page_id,
+            "parentCommentId": parent_comment_id,
+        }
+
+        return await self._send_with_retry(client, "POST", "/wiki/api/v2/footer-comments", payload)
+
+    async def create_comment(
+        self,
+        page_id: str,
+        text: str,
+        account_id: str = "default",
+    ) -> OutboundResult:
+        client = self._gateway.get_client(account_id)
+        if client is None:
+            return OutboundResult(success=False, error="Client not found")
+
+        adf_body = ADFBuilder.from_markdown(text)
+
+        payload = {
+            "body": {
+                "representation": "atlas_doc_format",
+                "value": adf_body,
+            },
+            "containerId": page_id,
+        }
+
+        return await self._send_with_retry(client, "POST", "/wiki/api/v2/footer-comments", payload)
+
+    async def edit_comment(
+        self,
+        comment_id: str,
+        text: str,
+        account_id: str = "default",
+    ) -> OutboundResult:
+        client = self._gateway.get_client(account_id)
+        if client is None:
+            return OutboundResult(success=False, error="Client not found")
+
+        adf_body = ADFBuilder.from_markdown(text)
+        return await self._send_with_retry(
+            client, "PUT", f"/wiki/api/v2/footer-comments/{comment_id}",
+            {"body": {"representation": "atlas_doc_format", "value": adf_body}},
+        )
+
+    async def delete_comment(
+        self,
+        comment_id: str,
+        account_id: str = "default",
+    ) -> OutboundResult:
+        client = self._gateway.get_client(account_id)
+        if client is None:
+            return OutboundResult(success=False, error="Client not found")
+        try:
+            await client.delete_footer_comment(comment_id)
+            return OutboundResult(success=True, comment_id=comment_id)
+        except Exception as e:
+            return OutboundResult(success=False, error=str(e))
+
+    async def get_comment_thread(
+        self,
+        comment_id: str,
+        account_id: str = "default",
+        max_depth: int = 3,
+    ) -> list[dict]:
+        client = self._gateway.get_client(account_id)
+        if client is None:
+            return []
+
+        results = []
+        children = await client.get_comment_children(comment_id, limit=25)
+        for child in children.get("results", []):
+            results.append(child)
+            if max_depth > 1:
+                deeper = await self.get_comment_thread(
+                    child["id"], account_id, max_depth - 1,
+                )
+                results.extend(deeper)
+        return results
+
+    async def send_text(
+        self,
+        target_id: str,
+        content: str,
+        *,
+        reply_to_id: str | None = None,
+        thread_id: str | None = None,
+        account_id: str = "default",
+    ) -> OutboundResult:
+        page_id = target_id
+        parent_id = reply_to_id or thread_id
+
+        if parent_id:
+            return await self.reply_to_comment(page_id, parent_id, content, account_id)
+        return await self.create_comment(page_id, content, account_id)
+
+    async def read_page(
+        self, page_id: str, account_id: str = "default"
+    ) -> dict | None:
+        client = self._gateway.get_client(account_id)
+        if client is None:
+            return None
+        return await client.get_page(page_id)
+
+    async def update_page(
+        self,
+        page_id: str,
+        title: str,
+        content: str,
+        space_id: str,
+        current_version: int,
+        version_comment: str = "AI 自动更新",
+        account_id: str = "default",
+    ) -> OutboundResult:
+        client = self._gateway.get_client(account_id)
+        if client is None:
+            return OutboundResult(success=False, error="Client not found")
+
+        adf_body = ADFBuilder.from_markdown(content)
+
+        payload = {
+            "id": page_id,
+            "status": "current",
+            "title": title,
+            "spaceId": space_id,
+            "body": {
+                "representation": "atlas_doc_format",
+                "value": adf_body,
+            },
+            "version": {
+                "number": current_version + 1,
+                "message": version_comment,
+            },
+        }
+
+        return await self._send_with_retry(client, "PUT", f"/wiki/api/v2/pages/{page_id}", payload)
+
+    async def update_page_with_conflict_retry(
+        self,
+        page_id: str,
+        title: str,
+        content: str,
+        space_id: str,
+        version_comment: str = "AI 自动更新",
+        account_id: str = "default",
+        max_retries: int = 3,
+    ) -> OutboundResult:
+        for attempt in range(max_retries):
+            page = await self.read_page(page_id, account_id)
+            if page is None:
+                return OutboundResult(success=False, error="Page not found")
+
+            current_version = page.get("version", {}).get("number", 1)
+            current_title = page.get("title", title)
+
+            result = await self.update_page(
+                page_id, current_title, content, space_id,
+                current_version, version_comment, account_id,
+            )
+
+            if result.success:
+                return result
+
+            if "409" in result.error or "conflict" in result.error.lower():
+                delay = RETRY_BASE_DELAY * (2 ** attempt)
+                logger.info(
+                    "Page update conflict, retry %d/%d after %.1fs",
+                    attempt + 1, max_retries, delay,
+                )
+                await asyncio.sleep(delay)
+                continue
+
+            return result
+
+        return OutboundResult(success=False, error="Max retries exceeded for update_page")
+
+    async def search_pages(
+        self,
+        query: str,
+        space_keys: list[str] | None = None,
+        limit: int = 10,
+        account_id: str = "default",
+    ) -> list[dict]:
+        client = self._gateway.get_client(account_id)
+        if client is None:
+            return []
+
+        cql = CQLBuilder.search_pages(query, space_keys)
+
+        try:
+            results = await client.cql_search(cql, limit=limit, expand="body.view")
+            return results.get("results", [])
+        except Exception as e:
+            logger.error("CQL search failed for query '%s': %s", query, e)
+            return []
+
+    async def _send_with_retry(
+        self,
+        client,
+        method: str,
+        path: str,
+        payload: dict,
+    ) -> OutboundResult:
+        for attempt in range(MAX_RETRY):
+            try:
+                result = await client._request(method, path, json=payload)
+                comment_id = result.get("id", "")
+                return OutboundResult(success=True, comment_id=comment_id)
+            except Exception as e:
+                status_code = getattr(getattr(e, "response", None), "status_code", 0)
+
+                if status_code == 409:
+                    return OutboundResult(success=False, error=f"409 Conflict: {e}")
+                if status_code == 403:
+                    return OutboundResult(success=False, error=f"403 Forbidden (permissions): {e}")
+                if attempt == MAX_RETRY - 1:
+                    return OutboundResult(success=False, error=str(e))
+
+                delay = RETRY_BASE_DELAY * (2 ** attempt)
+                logger.warning(
+                    "Outbound retry %d/%d after %.1fs: %s",
+                    attempt + 1, MAX_RETRY, delay, e,
+                )
+                await asyncio.sleep(delay)
+
+        return OutboundResult(success=False, error="Unknown retry failure")
+
+    def chunker(self, text: str, limit: int = 2000, ctx=None) -> list[str]:
+        chunks = []
+        paragraphs = text.split("\n\n")
+        current = ""
+
+        for para in paragraphs:
+            if len(current) + len(para) + 2 > limit and current:
+                chunks.append(current.rstrip())
+                current = para
+            else:
+                current = f"{current}\n\n{para}" if current else para
+
+        if current:
+            chunks.append(current.rstrip())
+        return chunks

backend/package/yuxi/channel/extensions/confluence/page_writer.py

@@ -0,0 +1,120 @@
+import asyncio
+import logging
+
+from yuxi.channel.extensions.confluence.format import ADFBuilder
+from yuxi.channel.extensions.confluence.gateway import ConfluenceGateway
+from yuxi.channel.extensions.confluence.types import OutboundResult
+
+logger = logging.getLogger(__name__)
+
+MAX_VERSION_RETRY = 3
+RETRY_BASE_DELAY = 1.0
+
+
+class ConfluencePageWriter:
+    def __init__(self, gateway: ConfluenceGateway):
+        self._gateway = gateway
+
+    async def create_page(
+        self,
+        space_id: str,
+        title: str,
+        content: str,
+        parent_page_id: str | None = None,
+        account_id: str = "default",
+    ) -> OutboundResult:
+        client = self._gateway.get_client(account_id)
+        if client is None:
+            return OutboundResult(success=False, error="Client not found")
+
+        adf_body = ADFBuilder.from_markdown(content)
+        payload = {
+            "spaceId": space_id,
+            "status": "current",
+            "title": title,
+            "body": {
+                "representation": "atlas_doc_format",
+                "value": adf_body,
+            },
+        }
+        if parent_page_id:
+            payload["parentId"] = parent_page_id
+
+        try:
+            result = await client._request("POST", "/wiki/api/v2/pages", json=payload)
+            return OutboundResult(success=True, comment_id=result.get("id", ""))
+        except Exception as e:
+            return OutboundResult(success=False, error=str(e))
+
+    async def update_page_with_conflict_retry(
+        self,
+        page_id: str,
+        title: str,
+        content: str,
+        space_id: str,
+        version_comment: str = "AI 自动更新",
+        account_id: str = "default",
+    ) -> OutboundResult:
+        client = self._gateway.get_client(account_id)
+        if client is None:
+            return OutboundResult(success=False, error="Client not found")
+
+        for attempt in range(MAX_VERSION_RETRY):
+            try:
+                page = await client.get_page(page_id)
+            except Exception as e:
+                return OutboundResult(success=False, error=f"Read page failed: {e}")
+
+            current_version = page.get("version", {}).get("number", 1)
+
+            adf_body = ADFBuilder.from_markdown(content)
+            payload = {
+                "id": page_id,
+                "status": "current",
+                "title": title,
+                "spaceId": space_id,
+                "body": {
+                    "representation": "atlas_doc_format",
+                    "value": adf_body,
+                },
+                "version": {
+                    "number": current_version + 1,
+                    "message": version_comment,
+                },
+            }
+
+            try:
+                result = await client._request(
+                    "PUT", f"/wiki/api/v2/pages/{page_id}", json=payload
+                )
+                return OutboundResult(success=True, comment_id=result.get("id", ""))
+            except Exception as e:
+                status = getattr(getattr(e, "response", None), "status_code", 0)
+                if status == 409 and attempt < MAX_VERSION_RETRY - 1:
+                    delay = RETRY_BASE_DELAY * (2 ** attempt)
+                    logger.info(
+                        "Page version conflict, retry %d/%d after %.1fs",
+                        attempt + 1, MAX_VERSION_RETRY, delay,
+                    )
+                    await asyncio.sleep(delay)
+                    continue
+                return OutboundResult(success=False, error=str(e))
+
+        return OutboundResult(success=False, error="Max retries exceeded")
+
+    async def generate_meeting_notes(
+        self,
+        space_id: str,
+        title: str,
+        transcript: str,
+        account_id: str = "default",
+    ) -> OutboundResult:
+        prompt = f"""请根据以下会议记录生成结构化的会议纪要：
+
+{transcript}
+
+格式要求：
+- 标题使用会议纪要标题
+- 分节：参会人员、会议摘要、讨论要点、决议事项、待办事项
+- 每个待办事项标注负责人和截止时间"""
+        return await self.create_page(space_id, title, prompt, account_id=account_id)

backend/package/yuxi/channel/extensions/confluence/plugin.json

@@ -0,0 +1,28 @@
+{
+    "id": "confluence",
+    "name": "Confluence",
+    "version": "0.1.0",
+    "description": "Confluence 知识管理渠道插件，支持页面评论交互、知识库检索与 AI 内容生成",
+    "author": "ForcePilot Team",
+    "order": 90,
+    "dependencies": ["httpx"],
+    "capabilities": {
+        "chat_types": ["comment"],
+        "message_types": ["text"],
+        "reactions": false,
+        "streaming": true,
+        "streaming_mode": "block",
+        "block_streaming": true,
+        "block_streaming_chunk_min_chars": 200,
+        "block_streaming_chunk_max_chars": 2000,
+        "block_streaming_chunk_break_preference": "paragraph",
+        "threads": true,
+        "reply": true,
+        "edit": true,
+        "unsend": true,
+        "media": false,
+        "polling_fallback": true
+    },
+    "enabled": false,
+    "python_requires": ">=3.12"
+}

backend/package/yuxi/channel/extensions/confluence/search.py

@@ -0,0 +1,48 @@
+from datetime import datetime, UTC
+
+
+class CQLBuilder:
+    @staticmethod
+    def comment_poll(
+        since_iso: str,
+        space_keys: list[str] | None = None,
+    ) -> str:
+        parts = ['type = "comment"']
+        if space_keys:
+            space_clause = " OR ".join(f'space = "{sk}"' for sk in space_keys)
+            parts.append(f"({space_clause})")
+        parts.append(f'lastModified > "{since_iso}"')
+        parts.append("ORDER BY lastModified ASC")
+        return " AND ".join(parts)
+
+    @staticmethod
+    def search_pages(
+        query: str,
+        space_keys: list[str] | None = None,
+    ) -> str:
+        parts = [
+            f'(title ~ "{query}" OR text ~ "{query}")',
+            'type = "page"',
+        ]
+        if space_keys:
+            space_clause = " OR ".join(f'space = "{sk}"' for sk in space_keys)
+            parts.append(f"({space_clause})")
+        return " AND ".join(parts)
+
+    @staticmethod
+    def recent_pages(
+        space_keys: list[str] | None = None,
+        hours: int = 24,
+    ) -> str:
+        parts = [
+            'type = "page"',
+            f'lastModified > now("-{hours}h")',
+        ]
+        if space_keys:
+            space_clause = " OR ".join(f'space = "{sk}"' for sk in space_keys)
+            parts.append(f"({space_clause})")
+        return " AND ".join(parts)
+
+    @staticmethod
+    def format_timestamp(ts: float) -> str:
+        return datetime.fromtimestamp(ts, tz=UTC).strftime("%Y-%m-%dT%H:%M:%S.000Z")

backend/package/yuxi/channel/extensions/confluence/security.py

@@ -0,0 +1,37 @@
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+class ConfluenceCommentPolicy:
+    VALID_POLICIES = frozenset({"open", "pairing", "allowlist", "disabled"})
+
+    def __init__(self, policy: str = "open", allow_from: list[str] | None = None):
+        self._policy = policy if policy in self.VALID_POLICIES else "open"
+        self._allowlist: set[str] = set(allow_from or [])
+
+    def is_allowed(self, author_id: str) -> bool:
+        if self._policy == "disabled":
+            return False
+        if self._policy == "open":
+            return True
+        if self._policy == "allowlist":
+            if not self._allowlist:
+                logger.warning("Comment policy is 'allowlist' but allowlist is empty")
+                return False
+            return author_id in self._allowlist
+        return True
+
+    def add_to_allowlist(self, account_id: str):
+        self._allowlist.add(account_id)
+
+    def remove_from_allowlist(self, account_id: str):
+        self._allowlist.discard(account_id)
+
+    @property
+    def policy(self) -> str:
+        return self._policy
+
+    @property
+    def allowlist(self) -> frozenset[str]:
+        return frozenset(self._allowlist)

backend/package/yuxi/channel/extensions/confluence/status.py

@@ -0,0 +1,97 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from yuxi.channel.extensions.confluence.gateway import ConfluenceClient
+    from yuxi.channel.extensions.confluence.types import ConfluenceAccount
+
+
+@dataclass
+class ChannelAccountSnapshot:
+    account_id: str
+    configured: bool
+    status_state: str
+    running: bool
+    connected: bool
+    health_state: str
+    dm_policy: str
+
+
+class ConfluenceStatus:
+    async def probe(
+        self,
+        account: ConfluenceAccount,
+        client: ConfluenceClient,
+    ) -> dict:
+        if not account.is_configured:
+            return {"success": False, "error": "Not configured"}
+
+        try:
+            spaces = await client.get_spaces(limit=1)
+            total = spaces.get("meta", {}).get("total_count", -1) if "meta" in spaces else -1
+            return {"success": True, "total_spaces": total}
+        except Exception as e:
+            return {"success": False, "error": str(e)}
+
+    async def check_space_permissions(
+        self,
+        account: ConfluenceAccount,
+        client: ConfluenceClient,
+    ) -> list[dict]:
+        results = []
+        for space_key in account.space_keys:
+            try:
+                cql = f'space = "{space_key}" AND type = "page"'
+                search = await client.cql_search(cql, limit=1)
+                results.append(
+                    {
+                        "space_key": space_key,
+                        "accessible": len(search.get("results", [])) > 0,
+                        "error": None,
+                    }
+                )
+            except Exception as e:
+                results.append(
+                    {
+                        "space_key": space_key,
+                        "accessible": False,
+                        "error": str(e),
+                    }
+                )
+        return results
+
+    def build_account_snapshot(
+        self,
+        account: ConfluenceAccount,
+        running: bool,
+        probe_result: dict,
+    ) -> ChannelAccountSnapshot:
+        return ChannelAccountSnapshot(
+            account_id=account.account_id,
+            configured=account.is_configured,
+            status_state="linked" if running else "stopped",
+            running=running,
+            connected=probe_result.get("success", False),
+            health_state="ok" if probe_result.get("success") else "unknown",
+            dm_policy=account.comment_policy,
+        )
+
+    def collect_status_issues(
+        self,
+        account: ConfluenceAccount,
+        probe_result: dict,
+        running: bool,
+    ) -> list[str]:
+        issues = []
+        if not account.is_configured:
+            issues.append("未配置：请设置 CONFLUENCE_URL, CONFLUENCE_EMAIL, CONFLUENCE_API_TOKEN")
+        if not probe_result.get("success"):
+            issues.append(f"API 连通性检查失败: {probe_result.get('error', 'Unknown error')}")
+        if not running:
+            issues.append("Gateway 未启动")
+        if account.space_keys and probe_result.get("success"):
+            space_list = ", ".join(account.space_keys)
+            issues.append(f"监听空间: {space_list}，请确认 Bot 用户对各空间有 View + Add Comments 权限")
+        return issues

backend/package/yuxi/channel/extensions/confluence/streaming.py

@@ -0,0 +1,30 @@
+class ConfluenceStreaming:
+    streaming_mode = "block"
+    block_streaming_enabled = True
+    MIN_CHARS = 200
+    MAX_CHARS = 2000
+    CHUNK_BREAK = "paragraph"
+    COALESCE_MIN = 80
+    COALESCE_MAX = 400
+    COALESCE_IDLE_MS = 500
+    MAX_BLOCKS = 5
+    FINISH_MARK = "..."
+
+    @classmethod
+    def should_stream(cls, estimated_length: int) -> bool:
+        return estimated_length > cls.MIN_CHARS
+
+    @classmethod
+    def build_chunk(cls, text: str, is_final: bool) -> str:
+        if is_final:
+            return text
+        return f"{text}\n\n{cls.FINISH_MARK}"
+
+    @classmethod
+    def create_chunker(cls):
+        return {
+            "mode": "paragraph",
+            "min_chars": cls.COALESCE_MIN,
+            "max_chars": cls.MAX_CHARS,
+            "break_on": cls.CHUNK_BREAK,
+        }

backend/package/yuxi/channel/extensions/confluence/types.py

@@ -0,0 +1,94 @@
+from dataclasses import dataclass, field
+from datetime import datetime
+
+
+@dataclass
+class ConfluenceAccount:
+    account_id: str = "default"
+    url: str = ""
+    email: str = ""
+    api_token: str = ""
+    oauth_client_id: str = ""
+    oauth_client_secret: str = ""
+    auth_mode: str = "api_token"
+    deployment_type: str = "cloud"
+    space_keys: list[str] = field(default_factory=list)
+    comment_poll_enabled: bool = True
+    poll_interval: float = 30.0
+    comment_policy: str = "open"
+    allow_from: list[str] = field(default_factory=list)
+    enable_page_writer: bool = False
+    enable_knowledge_retrieval: bool = False
+    enable_webhook: bool = False
+    name: str = ""
+
+    @property
+    def is_configured(self) -> bool:
+        if self.auth_mode == "api_token":
+            return bool(self.url and self.email and self.api_token)
+        return bool(self.url and self.oauth_client_id and self.oauth_client_secret)
+
+    @property
+    def base_url(self) -> str:
+        return self.url.rstrip("/")
+
+    @property
+    def api_v2_url(self) -> str:
+        return f"{self.base_url}/wiki/api/v2"
+
+    @property
+    def rest_url(self) -> str:
+        return f"{self.base_url}/rest/api"
+
+    @property
+    def auth_header(self) -> dict:
+        if self.auth_mode == "api_token":
+            import base64
+
+            credentials = base64.b64encode(f"{self.email}:{self.api_token}".encode()).decode()
+            return {"Authorization": f"Basic {credentials}"}
+        if self.auth_mode == "pat":
+            return {"Authorization": f"Bearer {self.api_token}"}
+        return {"Authorization": f"Bearer {self.oauth_client_secret}"}
+
+
+@dataclass
+class PageContext:
+    page_id: str
+    page_title: str = ""
+    space_key: str = ""
+    space_id: str = ""
+    page_url: str = ""
+
+    @property
+    def context_label(self) -> str:
+        return f"[{self.space_key}] {self.page_title}"
+
+
+@dataclass
+class InboundConfluenceComment:
+    comment_id: str
+    author_id: str
+    author_name: str
+    body_text: str
+    body_adf: str | None = None
+    body_storage: str | None = None
+    container_type: str = "page"
+    container_id: str = ""
+    parent_comment_id: str | None = None
+    page_context: PageContext | None = None
+    created_at: datetime | None = None
+    updated_at: datetime | None = None
+    version_number: int = 1
+
+    @property
+    def thread_id(self) -> str:
+        return self.parent_comment_id or self.comment_id
+
+
+@dataclass
+class OutboundResult:
+    success: bool
+    comment_id: str = ""
+    error: str = ""
+    detail: str = ""

backend/package/yuxi/channel/extensions/confluence/webhook.py

@@ -0,0 +1,146 @@
+import json
+import logging
+
+from yuxi.channel.extensions.confluence.types import InboundConfluenceComment, PageContext
+
+logger = logging.getLogger(__name__)
+
+SUPPORTED_EVENTS = {
+    "page_created", "page_updated", "page_removed",
+    "blog_created", "blog_updated",
+    "comment_created", "comment_updated", "comment_removed",
+    "attachment_created", "attachment_updated", "attachment_removed",
+    "label_added", "label_removed",
+    "page_moved", "page_restored", "page_trashed",
+    "content_created", "content_updated", "content_trashed", "content_restored",
+    "space_created", "space_updated", "space_removed",
+}
+
+
+class ConfluenceWebhookReceiver:
+    def __init__(self, on_page_changed=None, on_comment=None):
+        self._on_page_changed = on_page_changed
+        self._on_comment = on_comment
+
+    def parse_webhook(self, body: bytes, headers: dict) -> dict | None:
+        try:
+            payload = json.loads(body)
+        except json.JSONDecodeError:
+            logger.warning("Invalid webhook JSON payload")
+            return None
+
+        event_type = payload.get("eventType", "")
+        if event_type not in SUPPORTED_EVENTS:
+            logger.debug("Ignored webhook event type: %s", event_type)
+            return None
+
+        if event_type.startswith("comment_"):
+            return self._parse_comment_event(payload)
+
+        if event_type.startswith(("attachment_", "label_")):
+            return self._parse_resource_event(payload, event_type)
+
+        page_id = payload.get("page", {}).get("id")
+        page_title = payload.get("page", {}).get("title", "")
+        space_key = payload.get("space", {}).get("key", "")
+
+        if not page_id:
+            logger.warning("Webhook payload missing page.id")
+            return None
+
+        return {
+            "event_type": event_type,
+            "page_id": page_id,
+            "page_title": page_title,
+            "space_key": space_key,
+            "user_account_id": payload.get("userAccountId", ""),
+            "timestamp": payload.get("timestamp", ""),
+        }
+
+    def _parse_comment_event(self, payload: dict) -> dict | None:
+        comment = payload.get("comment", {})
+        comment_id = comment.get("id")
+        if not comment_id:
+            return None
+
+        container = payload.get("container") or payload.get("page") or {}
+        author = comment.get("createdBy", {}) or comment.get("lastUpdatedBy", {})
+
+        page_ctx = PageContext(
+            page_id=container.get("id", ""),
+            page_title=container.get("title", ""),
+            space_key=payload.get("space", {}).get("key", ""),
+        )
+
+        inbound = InboundConfluenceComment(
+            comment_id=comment_id,
+            author_id=author.get("accountId", ""),
+            author_name=author.get("displayName", ""),
+            body_text=self._extract_comment_body_text(comment),
+            container_id=container.get("id", ""),
+            parent_comment_id=comment.get("parentCommentId"),
+            page_context=page_ctx,
+        )
+
+        return {
+            "event_type": payload.get("eventType", ""),
+            "comment": inbound,
+            "timestamp": payload.get("timestamp", ""),
+        }
+
+    @staticmethod
+    def _extract_comment_body_text(comment: dict) -> str:
+        body = comment.get("body", {})
+        view_value = body.get("view", {}).get("value", "")
+        if view_value:
+            import re
+            clean = re.sub(r"<[^>]+>", "", view_value)
+            clean = re.sub(r"&[a-z]+;", " ", clean)
+            return re.sub(r"\s+", " ", clean).strip()
+        return ""
+
+    def _parse_resource_event(self, payload: dict, event_type: str) -> dict | None:
+        if event_type.startswith("attachment_"):
+            resource = payload.get("attachment", {})
+            return {
+                "event_type": event_type,
+                "attachment_id": resource.get("id"),
+                "filename": resource.get("title"),
+                "page_id": payload.get("page", {}).get("id"),
+                "timestamp": payload.get("timestamp", ""),
+            }
+
+        if event_type.startswith("label_"):
+            return {
+                "event_type": event_type,
+                "label_name": payload.get("label", {}).get("name"),
+                "page_id": payload.get("page", {}).get("id"),
+                "timestamp": payload.get("timestamp", ""),
+            }
+
+        return None
+
+    async def handle_page_event(self, event: dict, account_id: str = "default"):
+        if self._on_page_changed is None:
+            return
+
+        page_ctx = PageContext(
+            page_id=event["page_id"],
+            page_title=event["page_title"],
+            space_key=event["space_key"],
+        )
+
+        await self._on_page_changed(
+            event_type=event["event_type"],
+            page_context=page_ctx,
+            user_account_id=event["user_account_id"],
+            account_id=account_id,
+        )
+
+    async def handle_comment_event(self, event: dict, account_id: str = "default"):
+        if self._on_comment is None:
+            return
+        comment = event.get("comment")
+        if comment is None:
+            return
+        await self._on_comment(comment, account_id=account_id)