Agent/gateway/core/channels/feishu/manager.py

from __future__ import annotations

import os
from collections.abc import Mapping
from dataclasses import dataclass
from typing import Any

from gateway.core.channels.backends.memory_trace import MemoryTraceBackend
from gateway.core.channels.feishu.connector import FeishuConnector, WebhookParseError
from gateway.core.channels.feishu.http_run_executor import FeishuHttpRunApiExecutor
from gateway.core.channels.feishu.identity import DefaultUserIdentityResolver
from gateway.core.channels.feishu.router import FeishuMessageRouter
from gateway.core.channels.manager import ChannelRegistry
from gateway.core.channels.types import RouteResult


@dataclass
class FeishuChannelConfig:
    channel_id: str = "feishu"
    feishu_http_base_url: str = "http://127.0.0.1:4380"
    http_timeout: float = 120.0
    enabled: bool = True
    auto_create_trace: bool = True
    workspace_prefix: str = "feishu"
    default_agent_type: str = "personal_assistant"
    dispatch_reactions: bool = False
    dispatch_card_actions: bool = True  # 卡片授权等交互后需续跑 Agent；可用 CHANNELS_DISPATCH_CARD_ACTIONS=false 关闭
    agent_api_base_url: str = "http://127.0.0.1:8000"
    agent_run_model: str = "qwen3.5-flash"
    agent_run_max_iterations: int = 200
    agent_run_temperature: float = 0.3
    feishu_run_notify_on_submit: bool = True
    # 以下为「Trace 跟单」参数（WebSocket watch；不再 HTTP 轮询 messages）
    poll_assistant_messages: bool = True  # 是否把 assistant 推到飞书（False 时仍可连 WS 等终态清 Typing）
    poll_interval_seconds: float = 1.0  # WS recv 超时；无 WS 时 HTTP 查 trace 状态的间隔
    poll_request_timeout: float = 30.0  # 仅 HTTP 兜底 GET /api/traces/{id} 的超时
    poll_terminal_grace_rounds: int = 2  # 连续 N 轮终态后结束跟单
    poll_max_seconds: float = 0.0  # 跟单最长秒数，0=不限制
    assistant_max_text_chars: int = 8000
    typing_reaction_enabled: bool = True
    typing_reaction_emoji: str = "Typing"


class FeishuChannelManager(ChannelRegistry):
    """飞书渠道：组装连接器、Trace 后端、HTTP Run API 执行器与消息路由。"""

    def __init__(self, config: FeishuChannelConfig | None = None) -> None:
        super().__init__()
        self._config = config or FeishuChannelConfig()
        self.register_channel(self._config.channel_id, self._config)

        self._connector = FeishuConnector(
            feishu_http_base_url=self._config.feishu_http_base_url,
            timeout=self._config.http_timeout,
        )
        self._trace_backend = MemoryTraceBackend()
        self._identity = DefaultUserIdentityResolver()
        self._executor = FeishuHttpRunApiExecutor(
            base_url=self._config.agent_api_base_url,
            timeout=self._config.http_timeout,
            identity_resolver=self._identity,
            model=self._config.agent_run_model,
            max_iterations=self._config.agent_run_max_iterations,
            temperature=self._config.agent_run_temperature,
            notify_on_submit=self._config.feishu_run_notify_on_submit,
            poll_assistant_messages=self._config.poll_assistant_messages,
            poll_interval_seconds=self._config.poll_interval_seconds,
            poll_request_timeout=self._config.poll_request_timeout,
            poll_terminal_grace_rounds=self._config.poll_terminal_grace_rounds,
            poll_max_seconds=self._config.poll_max_seconds,
            assistant_max_text_chars=self._config.assistant_max_text_chars,
            typing_reaction_enabled=self._config.typing_reaction_enabled,
            typing_reaction_emoji=self._config.typing_reaction_emoji,
        )
        self._router = FeishuMessageRouter(
            connector=self._connector,
            trace_backend=self._trace_backend,
            executor_backend=self._executor,
            identity_resolver=self._identity,
            workspace_prefix=self._config.workspace_prefix,
            default_agent_type=self._config.default_agent_type,
            auto_create_trace=self._config.auto_create_trace,
            dispatch_reactions=self._config.dispatch_reactions,
            dispatch_card_actions=self._config.dispatch_card_actions,
        )

    @property
    def config(self) -> FeishuChannelConfig:
        return self._config

    @property
    def feishu_connector(self) -> FeishuConnector:
        return self._connector

    @property
    def message_router(self) -> FeishuMessageRouter:
        return self._router

    @classmethod
    def from_env(cls) -> FeishuChannelManager:
        """从环境变量构造实例（与 docker-compose / .env 配合）。"""
        return cls(
            FeishuChannelConfig(
                feishu_http_base_url=os.getenv("FEISHU_HTTP_BASE_URL", "http://127.0.0.1:4380").strip(),
                http_timeout=float(os.getenv("FEISHU_HTTP_TIMEOUT", "120")),
                dispatch_reactions=os.getenv("CHANNELS_DISPATCH_REACTIONS", "false").lower() in ("1", "true", "yes"),
                dispatch_card_actions=os.getenv("CHANNELS_DISPATCH_CARD_ACTIONS", "true").lower()
                in ("1", "true", "yes"),
                agent_api_base_url=os.getenv("GATEWAY_AGENT_API_BASE_URL", "http://127.0.0.1:8000").strip(),
                agent_run_model=os.getenv("FEISHU_AGENT_RUN_MODEL", "qwen3.5-flash").strip(),
                agent_run_max_iterations=int(os.getenv("FEISHU_AGENT_RUN_MAX_ITERATIONS", "200")),
                agent_run_temperature=float(os.getenv("FEISHU_AGENT_RUN_TEMPERATURE", "0.3")),
                feishu_run_notify_on_submit=os.getenv("CHANNELS_FEISHU_RUN_NOTIFY", "true").lower()
                in ("1", "true", "yes"),
                poll_assistant_messages=os.getenv("FEISHU_AGENT_POLL_ASSISTANTS", "true").lower()
                in ("1", "true", "yes"),
                poll_interval_seconds=float(os.getenv("FEISHU_AGENT_POLL_INTERVAL", "1.0")),
                poll_request_timeout=float(os.getenv("FEISHU_AGENT_POLL_REQUEST_TIMEOUT", "30")),
                poll_terminal_grace_rounds=int(os.getenv("FEISHU_AGENT_POLL_GRACE_ROUNDS", "2")),
                poll_max_seconds=float(os.getenv("FEISHU_AGENT_POLL_MAX_SECONDS", "0")),
                assistant_max_text_chars=int(os.getenv("FEISHU_AGENT_ASSISTANT_MAX_CHARS", "8000")),
                typing_reaction_enabled=os.getenv("FEISHU_TYPING_REACTION", "true").lower()
                in ("1", "true", "yes"),
                typing_reaction_emoji=os.getenv("FEISHU_TYPING_REACTION_EMOJI", "Typing").strip()
                or "Typing",
            )
        )

    async def handle_feishu_inbound_webhook(self, body: Mapping[str, Any]) -> RouteResult:
        """处理飞书适配层 POST 到 ``/api/channels/feishu/inbound/webhook`` 的规范化事件。"""
        cid = self._config.channel_id
        if not self._running.get(cid, False):
            return RouteResult(ok=False, error="channel_stopped")

        try:
            event = FeishuConnector.parse_feishu_inbound_event(body)
        except WebhookParseError as e:
            return RouteResult(ok=False, error=str(e))

        return await self._router.route_feishu_inbound_event(event)