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

from __future__ import annotations

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

from gateway.core.channels.feishu.bridge import FeishuHttpRunApiExecutor
from gateway.core.channels.feishu.connector import FeishuConnector, WebhookParseError
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
from gateway.core.lifecycle import LifecycleTraceBackend, TraceManager, WorkspaceManager
from utils.env_parse import env_bool, env_float, env_int, env_str


@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"
    # Trace 跟单结束后的生命周期（Workspace 沙箱 / 渠道绑定）
    # 默认不在单次 Trace 终态时停沙箱：同一用户共用一个 workspace/容器，多轮对话与多 trace 复用。
    stop_container_on_trace_terminal: bool = False
    stop_container_on_trace_not_found: bool = True
    release_ref_on_trace_terminal: bool = False
    # False：绑定新 trace 时仍保留旧 trace 在 workspace 的引用，便于 Executor 等按 trace_id 解析同一用户目录
    release_previous_trace_ref_on_bind: bool = False


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._workspace_manager = WorkspaceManager.from_env()
        self._trace_manager = TraceManager.from_env(self._workspace_manager)
        self._trace_backend = LifecycleTraceBackend(
            self._trace_manager,
            release_previous_trace_ref_on_bind=self._config.release_previous_trace_ref_on_bind,
        )
        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,
            workspace_manager=self._workspace_manager,
            workspace_prefix=self._config.workspace_prefix,
            channel_id=self._config.channel_id,
            lifecycle_trace_backend=self._trace_backend,
            stop_container_on_trace_terminal=self._config.stop_container_on_trace_terminal,
            stop_container_on_trace_not_found=self._config.stop_container_on_trace_not_found,
            release_ref_on_trace_terminal=self._config.release_ref_on_trace_terminal,
        )
        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=env_str("FEISHU_HTTP_BASE_URL", "http://127.0.0.1:4380"),
                http_timeout=env_float("FEISHU_HTTP_TIMEOUT", 120.0),
                dispatch_reactions=env_bool("CHANNELS_DISPATCH_REACTIONS", False),
                dispatch_card_actions=env_bool("CHANNELS_DISPATCH_CARD_ACTIONS", True),
                agent_api_base_url=env_str("GATEWAY_AGENT_API_BASE_URL", "http://127.0.0.1:8000"),
                agent_run_model=env_str("FEISHU_AGENT_RUN_MODEL", "qwen3.5-flash"),
                agent_run_max_iterations=env_int("FEISHU_AGENT_RUN_MAX_ITERATIONS", 200),
                agent_run_temperature=env_float("FEISHU_AGENT_RUN_TEMPERATURE", 0.3),
                feishu_run_notify_on_submit=env_bool("CHANNELS_FEISHU_RUN_NOTIFY", True),
                poll_assistant_messages=env_bool("FEISHU_AGENT_POLL_ASSISTANTS", True),
                poll_interval_seconds=env_float("FEISHU_AGENT_POLL_INTERVAL", 1.0),
                poll_request_timeout=env_float("FEISHU_AGENT_POLL_REQUEST_TIMEOUT", 30.0),
                poll_terminal_grace_rounds=env_int("FEISHU_AGENT_POLL_GRACE_ROUNDS", 2),
                poll_max_seconds=env_float("FEISHU_AGENT_POLL_MAX_SECONDS", 0.0),
                assistant_max_text_chars=env_int("FEISHU_AGENT_ASSISTANT_MAX_CHARS", 8000),
                typing_reaction_enabled=env_bool("FEISHU_TYPING_REACTION", True),
                typing_reaction_emoji=env_str("FEISHU_TYPING_REACTION_EMOJI", "Typing") or "Typing",
                stop_container_on_trace_terminal=env_bool("GATEWAY_WORKSPACE_STOP_ON_TRACE_TERMINAL", False),
                stop_container_on_trace_not_found=env_bool("GATEWAY_WORKSPACE_STOP_ON_TRACE_NOT_FOUND", True),
                release_ref_on_trace_terminal=env_bool("GATEWAY_LIFECYCLE_RELEASE_REF_ON_TRACE_TERMINAL", False),
                release_previous_trace_ref_on_bind=env_bool("GATEWAY_LIFECYCLE_RELEASE_PREV_TRACE_REF_ON_BIND", False),
            )
        )

    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)