TelegramBotAdmin/wbb/services/business_assistant.py

from __future__ import annotations

import asyncio
import json
import re
from contextlib import suppress
from datetime import UTC, datetime
from typing import Any

from aiohttp import ClientError, ClientSession

import wbb
from wbb.utils.dbassistant import (
    append_conversation_message,
    claim_update,
    classify_handoff,
    dead_letter_update,
    due_digest_connections,
    get_account_settings,
    get_business_connection,
    get_conversation,
    get_conversation_by_chat,
    get_or_create_conversation,
    list_business_connections,
    load_update_offset,
    mark_digest_sent,
    mark_update_done,
    mark_update_failed,
    match_knowledge,
    pause_conversation_for_human,
    recent_conversation_messages,
    reserve_ai_usage,
    resume_conversation,
    runtime_status,
    save_update_offset,
    set_conversation_handoff,
    touch_business_connection,
    update_conversation_summary,
    update_runtime_status,
    upsert_business_connection,
    usage_metrics,
    utc_now,
)

BUSINESS_ALLOWED_UPDATES = [
    "business_connection",
    "business_message",
    "edited_business_message",
    "deleted_business_messages",
]
BUSINESS_REPLY_WINDOW_SECONDS = 24 * 60 * 60


class TelegramBotApiError(RuntimeError):
    def __init__(
        self,
        message: str,
        *,
        error_code: int = 0,
        retry_after: int = 0,
    ) -> None:
        super().__init__(message)
        self.error_code = int(error_code or 0)
        self.retry_after = int(retry_after or 0)


class AssistantProviderError(RuntimeError):
    pass


class TelegramBusinessApi:
    def __init__(self, token: str, session: ClientSession) -> None:
        self._base_url = f"https://api.telegram.org/bot{token}"
        self._session = session

    async def call(
        self,
        method: str,
        payload: dict[str, Any] | None = None,
    ) -> Any:
        last_error: TelegramBotApiError | None = None
        for attempt in range(3):
            try:
                async with self._session.post(
                    f"{self._base_url}/{method}", json=payload or {}
                ) as response:
                    data = await response.json(content_type=None)
            except (ClientError, TimeoutError, ValueError) as exc:
                last_error = TelegramBotApiError("无法连接 Telegram Bot API。")
                if attempt < 2:
                    await asyncio.sleep(2**attempt)
                    continue
                raise last_error from exc
            if response.status == 200 and isinstance(data, dict) and data.get("ok"):
                return data.get("result")
            parameters = data.get("parameters") if isinstance(data, dict) else {}
            last_error = TelegramBotApiError(
                str(data.get("description") or "Telegram Bot API 请求失败。")
                if isinstance(data, dict)
                else "Telegram Bot API 请求失败。",
                error_code=int(data.get("error_code") or response.status)
                if isinstance(data, dict)
                else response.status,
                retry_after=int((parameters or {}).get("retry_after") or 0),
            )
            retryable = last_error.error_code == 429 or response.status >= 500
            if retryable and attempt < 2:
                await asyncio.sleep(last_error.retry_after or 2**attempt)
                continue
            raise last_error
        raise last_error or TelegramBotApiError("Telegram Bot API 请求失败。")

    async def get_me(self) -> dict[str, Any]:
        result = await self.call("getMe")
        return result if isinstance(result, dict) else {}

    async def get_webhook_info(self) -> dict[str, Any]:
        result = await self.call("getWebhookInfo")
        return result if isinstance(result, dict) else {}

    async def get_business_connection(self, connection_id: str) -> dict[str, Any]:
        result = await self.call(
            "getBusinessConnection", {"business_connection_id": str(connection_id)}
        )
        return result if isinstance(result, dict) else {}

    async def get_updates(
        self, *, offset: int, poll_timeout: int = 30
    ) -> list[dict[str, Any]]:
        result = await self.call(
            "getUpdates",
            {
                "offset": int(offset),
                "timeout": int(poll_timeout),
                "limit": 100,
                "allowed_updates": BUSINESS_ALLOWED_UPDATES,
            },
        )
        return [item for item in (result or []) if isinstance(item, dict)]

    async def send_message(
        self,
        chat_id: int,
        text: str,
        *,
        business_connection_id: str = "",
        reply_markup: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        payload: dict[str, Any] = {
            "chat_id": int(chat_id),
            "text": str(text)[:4096],
        }
        if business_connection_id:
            payload["business_connection_id"] = str(business_connection_id)
        if reply_markup:
            payload["reply_markup"] = reply_markup
        result = await self.call("sendMessage", payload)
        return result if isinstance(result, dict) else {}

    async def read_business_message(
        self,
        connection_id: str,
        chat_id: int,
        message_id: int,
    ) -> None:
        await self.call(
            "readBusinessMessage",
            {
                "business_connection_id": str(connection_id),
                "chat_id": int(chat_id),
                "message_id": int(message_id),
            },
        )


def _strip_json_fence(value: str) -> str:
    text = value.strip()
    if text.startswith("```"):
        text = re.sub(r"^```(?:json)?\s*", "", text, flags=re.IGNORECASE)
        text = re.sub(r"\s*```$", "", text)
    return text.strip()


def business_reply_window_open(
    message: dict[str, Any], *, now: datetime | None = None
) -> bool:
    try:
        sent_at = datetime.fromtimestamp(int(message.get("date") or 0), UTC)
    except (OSError, OverflowError, TypeError, ValueError):
        return False
    if sent_at.timestamp() <= 0:
        return False
    age = (now or utc_now()) - sent_at
    return age.total_seconds() <= BUSINESS_REPLY_WINDOW_SECONDS


def parse_ai_decision(value: str, *, allowed_entry_ids: set[str]) -> dict[str, Any]:
    try:
        payload = json.loads(_strip_json_fence(value))
    except (json.JSONDecodeError, TypeError) as exc:
        raise AssistantProviderError("模型没有返回有效的 JSON 结果。") from exc
    if not isinstance(payload, dict):
        raise AssistantProviderError("模型结果必须是 JSON 对象。")
    action = str(payload.get("action") or "")
    if action not in {"answer", "clarify", "handoff"}:
        raise AssistantProviderError("模型返回了不支持的接待动作。")
    reply = str(payload.get("reply") or "").strip()
    if action in {"answer", "clarify"} and not reply:
        raise AssistantProviderError("模型没有提供回复内容。")
    if len(reply) > 4000:
        raise AssistantProviderError("模型回复内容过长。")
    matched_ids = [
        str(item)
        for item in payload.get("matched_entry_ids") or []
        if str(item) in allowed_entry_ids
    ]
    if action == "answer" and not matched_ids:
        raise AssistantProviderError("业务回答没有引用知识条目。")
    return {
        "action": action,
        "reply": reply,
        "handoff_reason": str(payload.get("handoff_reason") or "ai_handoff")[:200],
        "matched_entry_ids": matched_ids,
        "summary": str(payload.get("summary") or "")[:4000],
    }


class OpenAICompatibleAssistant:
    def __init__(
        self,
        session: ClientSession,
        *,
        base_url: str,
        api_key: str,
        model: str,
        timeout_seconds: int = 30,
        max_output_tokens: int = 600,
    ) -> None:
        self._session = session
        self.base_url = str(base_url).rstrip("/")
        self.api_key = str(api_key)
        self.model = str(model)
        self.timeout_seconds = max(5, min(int(timeout_seconds), 120))
        self.max_output_tokens = max(100, min(int(max_output_tokens), 2000))

    @property
    def configured(self) -> bool:
        return bool(self.base_url and self.api_key and self.model)

    async def decide(
        self,
        *,
        settings: dict[str, Any],
        conversation: dict[str, Any],
        messages: list[dict[str, Any]],
        knowledge: list[dict[str, Any]],
        customer_text: str,
    ) -> dict[str, Any]:
        if not self.configured:
            raise AssistantProviderError("OpenAI 兼容模型尚未配置。")
        knowledge_payload = [
            {
                "entry_id": item["entry_id"],
                "question": item["question"],
                "answer": item["answer"],
                "tags": item.get("tags") or [],
            }
            for item in knowledge
        ]
        recent_payload = [
            {"direction": item.get("direction"), "text": item.get("text", "")}
            for item in messages[-12:]
        ]
        system_prompt = (
            f"{settings.get('system_prompt')}\n"
            "必须遵守：业务事实只能来自 knowledge；不得自行补全价格、承诺、退款或政策。"
            "仅输出一个 JSON 对象，不使用 Markdown。字段为 action、reply、handoff_reason、"
            "matched_entry_ids、summary。action 只能是 answer、clarify、handoff。"
            "answer 必须填写实际引用的 matched_entry_ids；无法可靠回答时使用 handoff。"
        )
        request_body = {
            "model": self.model,
            "temperature": 0.2,
            "max_tokens": self.max_output_tokens,
            "messages": [
                {"role": "system", "content": system_prompt},
                {
                    "role": "user",
                    "content": json.dumps(
                        {
                            "language": settings.get("language"),
                            "tone": settings.get("tone"),
                            "existing_summary": conversation.get("summary", ""),
                            "recent_messages": recent_payload,
                            "knowledge": knowledge_payload,
                            "customer_message": customer_text,
                        },
                        ensure_ascii=False,
                    ),
                },
            ],
        }
        try:
            async with asyncio.timeout(self.timeout_seconds):
                async with self._session.post(
                    f"{self.base_url}/chat/completions",
                    headers={
                        "Authorization": f"Bearer {self.api_key}",
                        "Content-Type": "application/json",
                    },
                    json=request_body,
                ) as response:
                    payload = await response.json(content_type=None)
        except (ClientError, TimeoutError, ValueError) as exc:
            raise AssistantProviderError("OpenAI 兼容服务暂时不可用。") from exc
        if response.status != 200:
            raise AssistantProviderError(f"OpenAI 兼容服务返回 HTTP {response.status}。")
        try:
            content = payload["choices"][0]["message"]["content"]
        except (KeyError, IndexError, TypeError) as exc:
            raise AssistantProviderError("OpenAI 兼容服务响应格式无效。") from exc
        return parse_ai_decision(
            str(content),
            allowed_entry_ids={str(item["entry_id"]) for item in knowledge},
        )

    async def test_connection(self) -> dict[str, Any]:
        if not self.configured:
            raise AssistantProviderError("OpenAI 兼容模型尚未配置。")
        request_body = {
            "model": self.model,
            "temperature": 0,
            "max_tokens": 12,
            "messages": [{"role": "user", "content": "只回复 OK"}],
        }
        try:
            async with asyncio.timeout(self.timeout_seconds):
                async with self._session.post(
                    f"{self.base_url}/chat/completions",
                    headers={
                        "Authorization": f"Bearer {self.api_key}",
                        "Content-Type": "application/json",
                    },
                    json=request_body,
                ) as response:
                    payload = await response.json(content_type=None)
        except (ClientError, TimeoutError, ValueError) as exc:
            raise AssistantProviderError("无法连接 OpenAI 兼容服务。") from exc
        if response.status != 200:
            raise AssistantProviderError(f"模型测试失败：HTTP {response.status}。")
        return {"ok": True, "model": self.model, "response_id": payload.get("id", "")}


def provider_from_wbb(session: ClientSession) -> OpenAICompatibleAssistant:
    return OpenAICompatibleAssistant(
        session,
        base_url=str(getattr(wbb, "BUSINESS_ASSISTANT_OPENAI_BASE_URL", "")),
        api_key=str(getattr(wbb, "BUSINESS_ASSISTANT_OPENAI_API_KEY", "")),
        model=str(getattr(wbb, "BUSINESS_ASSISTANT_OPENAI_MODEL", "")),
        timeout_seconds=int(
            getattr(wbb, "BUSINESS_ASSISTANT_OPENAI_TIMEOUT_SECONDS", 30)
        ),
        max_output_tokens=int(
            getattr(wbb, "BUSINESS_ASSISTANT_OPENAI_MAX_OUTPUT_TOKENS", 600)
        ),
    )


class BusinessAssistantRuntime:
    def __init__(
        self,
        *,
        token: str,
        session: ClientSession,
        provider: OpenAICompatibleAssistant | None = None,
    ) -> None:
        self.api = TelegramBusinessApi(token, session)
        self.provider = provider or provider_from_wbb(session)
        self._poll_task: asyncio.Task[None] | None = None
        self._digest_task: asyncio.Task[None] | None = None
        self._stopping = asyncio.Event()

    async def start(self) -> dict[str, Any]:
        if self._poll_task and not self._poll_task.done():
            return await runtime_status()
        self._stopping.clear()
        await update_runtime_status(
            {
                "polling_state": "starting",
                "model_configured": self.provider.configured,
                "last_error": "",
            }
        )
        try:
            me = await self.api.get_me()
            webhook = await self.api.get_webhook_info()
        except TelegramBotApiError as exc:
            return await update_runtime_status(
                {"polling_state": "error", "last_error": str(exc)}
            )
        webhook_url = str(webhook.get("url") or "")
        supported = bool(me.get("can_connect_to_business"))
        await update_runtime_status(
            {
                "business_mode_supported": supported,
                "webhook_conflict": bool(webhook_url),
                "webhook_url": webhook_url,
                "bot_username": str(me.get("username") or ""),
            }
        )
        if webhook_url:
            return await update_runtime_status(
                {
                    "polling_state": "blocked",
                    "last_error": "检测到 Telegram webhook，Business 长轮询未启动。",
                }
            )
        if not supported:
            return await update_runtime_status(
                {
                    "polling_state": "blocked",
                    "last_error": "请先在 BotFather 为机器人开启 Business Mode。",
                }
            )
        await self._refresh_connections()
        self._poll_task = asyncio.create_task(
            self._poll_loop(), name="business-assistant-poller"
        )
        self._digest_task = asyncio.create_task(
            self._digest_loop(), name="business-assistant-digest"
        )
        return await update_runtime_status(
            {"polling_state": "running", "last_error": "", "started_at": utc_now()}
        )

    async def _refresh_connections(self) -> None:
        page = 1
        while True:
            connections, total = await list_business_connections(
                page=page, page_size=100
            )
            for connection in connections:
                connection_id = str(connection["connection_id"])
                try:
                    payload = await self.api.get_business_connection(connection_id)
                    await upsert_business_connection(payload)
                except TelegramBotApiError as exc:
                    await touch_business_connection(
                        connection_id,
                        error=str(exc),
                        is_enabled=False if exc.error_code in {400, 403} else None,
                    )
            if not connections or page * 100 >= total:
                return
            page += 1

    async def stop(self) -> None:
        self._stopping.set()
        tasks = [task for task in (self._poll_task, self._digest_task) if task]
        for task in tasks:
            task.cancel()
        if tasks:
            await asyncio.gather(*tasks, return_exceptions=True)
        self._poll_task = None
        self._digest_task = None
        await update_runtime_status(
            {"polling_state": "stopped", "stopped_at": utc_now()}
        )

    async def _poll_loop(self) -> None:
        offset = await load_update_offset()
        backoff = 1
        while not self._stopping.is_set():
            try:
                updates = await self.api.get_updates(offset=offset, poll_timeout=30)
                backoff = 1
                await update_runtime_status(
                    {"polling_state": "running", "last_poll_at": utc_now(), "last_error": ""}
                )
                for update in updates:
                    update_id = int(update.get("update_id") or 0)
                    if update_id <= 0:
                        continue
                    await self.process_update(update)
                    offset = max(offset, update_id + 1)
                    await save_update_offset(offset)
            except asyncio.CancelledError:
                raise
            except TelegramBotApiError as exc:
                wait_for = exc.retry_after or backoff
                await update_runtime_status(
                    {"polling_state": "retrying", "last_error": str(exc)}
                )
                await asyncio.sleep(min(max(wait_for, 1), 60))
                backoff = min(backoff * 2, 60)
            except Exception as exc:
                await update_runtime_status(
                    {"polling_state": "retrying", "last_error": str(exc)[:1000]}
                )
                await asyncio.sleep(backoff)
                backoff = min(backoff * 2, 60)

    async def process_update(self, update: dict[str, Any]) -> None:
        update_id = int(update.get("update_id") or 0)
        if not await claim_update(update_id):
            return
        last_error = ""
        for attempt in range(1, 4):
            try:
                if isinstance(update.get("business_connection"), dict):
                    await upsert_business_connection(update["business_connection"])
                elif isinstance(update.get("business_message"), dict):
                    await self._handle_business_message(update["business_message"])
                elif isinstance(update.get("edited_business_message"), dict):
                    await self._handle_edited_message(update["edited_business_message"])
                elif isinstance(update.get("deleted_business_messages"), dict):
                    await self._handle_deleted_messages(update["deleted_business_messages"])
                await mark_update_done(update_id)
                return
            except TelegramBotApiError as exc:
                last_error = str(exc)
                if exc.retry_after:
                    await asyncio.sleep(min(exc.retry_after, 60))
            except Exception as exc:
                last_error = str(exc)
            if attempt < 3:
                await asyncio.sleep(attempt)
        await mark_update_failed(update_id, last_error)
        await dead_letter_update(update, last_error or "unknown update error")

    async def _resolve_connection(self, connection_id: str) -> dict[str, Any]:
        connection = await get_business_connection(connection_id)
        if connection:
            return connection
        payload = await self.api.get_business_connection(connection_id)
        return await upsert_business_connection(payload)

    async def _handle_business_message(self, message: dict[str, Any]) -> None:
        connection_id = str(message.get("business_connection_id") or "")
        if not connection_id or message.get("is_from_offline"):
            return
        if message.get("sender_business_bot"):
            return
        connection = await self._resolve_connection(connection_id)
        await touch_business_connection(connection_id)
        chat = message.get("chat") if isinstance(message.get("chat"), dict) else {}
        sender = message.get("from") if isinstance(message.get("from"), dict) else {}
        chat_id = int(chat.get("id") or 0)
        message_id = int(message.get("message_id") or 0)
        if not chat_id or not message_id:
            return
        owner_id = int((connection.get("user") or {}).get("id") or 0)
        owner_reply = int(sender.get("id") or 0) == owner_id
        conversation = await get_or_create_conversation(
            connection_id,
            chat_id,
            customer=None if owner_reply else sender,
        )
        if owner_reply:
            settings = await get_account_settings(connection_id)
            await append_conversation_message(
                conversation["conversation_id"],
                direction="human",
                telegram_message_id=message_id,
                text=str(message.get("text") or message.get("caption") or ""),
                sender_id=owner_id,
            )
            await pause_conversation_for_human(
                conversation["conversation_id"],
                hours=int(settings.get("human_pause_hours") or 24),
            )
            return
        await self._handle_customer_message(
            connection=connection,
            conversation=conversation,
            message=message,
            sender=sender,
        )

    async def _handle_customer_message(
        self,
        *,
        connection: dict[str, Any],
        conversation: dict[str, Any],
        message: dict[str, Any],
        sender: dict[str, Any],
    ) -> None:
        connection_id = str(connection["connection_id"])
        chat_id = int((message.get("chat") or {}).get("id") or 0)
        message_id = int(message.get("message_id") or 0)
        text = str(message.get("text") or "").strip()
        settings = await get_account_settings(connection_id)
        await append_conversation_message(
            conversation["conversation_id"],
            direction="incoming",
            telegram_message_id=message_id,
            text=text or "[非文本消息]",
            sender_id=int(sender.get("id") or 0),
            metadata={"content_type": "text" if text else "unsupported"},
        )
        if not connection.get("is_enabled") or not settings.get("assistant_enabled"):
            return
        current = await get_conversation(conversation["conversation_id"]) or conversation
        if current.get("status") == "human_paused":
            paused_until = current.get("paused_until")
            if isinstance(paused_until, datetime) and (
                paused_until.replace(tzinfo=UTC) if paused_until.tzinfo is None else paused_until
            ) <= utc_now():
                current = await resume_conversation(current["conversation_id"])
            else:
                return
        if current.get("status") in {"handoff", "closed"}:
            return
        if not business_reply_window_open(message):
            await self._handoff(
                connection,
                current,
                settings,
                "reply_window_expired",
                send_customer_notice=False,
            )
            return
        rights = connection.get("rights") or {}
        if not rights.get("can_reply"):
            await self._handoff(
                connection, current, settings, "can_reply_missing", send_customer_notice=False
            )
            return
        if rights.get("can_read_messages"):
            with suppress(TelegramBotApiError):
                await self.api.read_business_message(connection_id, chat_id, message_id)
        if not text:
            await self._handoff(
                connection,
                current,
                settings,
                "unsupported_message",
                customer_notice=str(settings.get("unsupported_message")),
            )
            return
        classification = classify_handoff(text)
        if classification:
            await self._handoff(connection, current, settings, classification)
            return
        knowledge = await match_knowledge(connection_id, text)
        if not knowledge:
            await self._handoff(connection, current, settings, "knowledge_not_found")
            return
        allowed, quota_reason = await reserve_ai_usage(
            connection_id, chat_id, settings
        )
        if not allowed:
            await self._handoff(connection, current, settings, quota_reason)
            return
        messages = await recent_conversation_messages(current["conversation_id"], limit=12)
        try:
            decision = await self.provider.decide(
                settings=settings,
                conversation=current,
                messages=messages,
                knowledge=knowledge,
                customer_text=text,
            )
        except AssistantProviderError as exc:
            await update_runtime_status(
                {"provider_last_error": str(exc)[:1000], "provider_last_error_at": utc_now()}
            )
            await self._handoff(
                connection,
                current,
                settings,
                "provider_error",
            )
            return
        if decision["summary"]:
            await update_conversation_summary(
                current["conversation_id"], decision["summary"]
            )
        if decision["action"] == "handoff":
            await self._handoff(
                connection,
                current,
                settings,
                decision["handoff_reason"],
                summary=decision["summary"],
            )
            return
        response = await self.api.send_message(
            chat_id,
            decision["reply"],
            business_connection_id=connection_id,
        )
        await append_conversation_message(
            current["conversation_id"],
            direction="assistant",
            telegram_message_id=int(response.get("message_id") or 0),
            text=decision["reply"],
            sender_id=int((response.get("sender_business_bot") or {}).get("id") or 0),
            metadata={
                "action": decision["action"],
                "matched_entry_ids": decision["matched_entry_ids"],
            },
        )

    async def _handoff(
        self,
        connection: dict[str, Any],
        conversation: dict[str, Any],
        settings: dict[str, Any],
        reason: str,
        *,
        summary: str = "",
        customer_notice: str = "",
        send_customer_notice: bool = True,
    ) -> None:
        updated = await set_conversation_handoff(
            conversation["conversation_id"], reason, summary=summary
        )
        notice = customer_notice or str(settings.get("handoff_message") or "")
        if send_customer_notice and notice and connection.get("rights", {}).get("can_reply"):
            try:
                response = await self.api.send_message(
                    int(updated["chat_id"]),
                    notice,
                    business_connection_id=str(connection["connection_id"]),
                )
                await append_conversation_message(
                    updated["conversation_id"],
                    direction="assistant",
                    telegram_message_id=int(response.get("message_id") or 0),
                    text=notice,
                    metadata={"action": "handoff", "reason": reason},
                )
            except TelegramBotApiError as exc:
                await update_runtime_status(
                    {
                        "customer_notice_last_error": str(exc)[:1000],
                        "customer_notice_last_error_at": utc_now(),
                    }
                )
        await self.notify_handoff(connection, updated, settings, reason)

    async def notify_handoff(
        self,
        connection: dict[str, Any],
        conversation: dict[str, Any],
        settings: dict[str, Any],
        reason: str,
    ) -> None:
        customer = conversation.get("customer") or {}
        display = " ".join(
            item for item in (customer.get("first_name"), customer.get("last_name")) if item
        ) or (f"@{customer.get('username')}" if customer.get("username") else str(conversation["chat_id"]))
        latest = await recent_conversation_messages(conversation["conversation_id"], limit=1)
        latest_text = str(latest[-1].get("text") or "")[:1000] if latest else ""
        text = (
            "需要人工接待\n"
            f"账号：{(connection.get('user') or {}).get('first_name') or connection['connection_id']}\n"
            f"客户：{display}\n"
            f"原因：{reason}\n"
            f"原消息：chat_id={conversation['chat_id']} / "
            f"message_id={latest[-1].get('telegram_message_id') if latest else '未知'}\n"
            f"摘要：{conversation.get('summary') or latest_text or '暂无'}"
        )
        markup = {
            "inline_keyboard": [
                [
                    {
                        "text": "恢复自动回复",
                        "callback_data": f"ba:resume:{conversation['conversation_id']}",
                    }
                ]
            ]
        }
        destinations: list[int] = []
        destination = str(settings.get("notification_destination") or "owner")
        if destination in {"owner", "both"} and int(connection.get("user_chat_id") or 0):
            destinations.append(int(connection["user_chat_id"]))
        if destination in {"ops", "both"} and int(settings.get("ops_group_id") or 0):
            destinations.append(int(settings["ops_group_id"]))
        for chat_id in dict.fromkeys(destinations):
            try:
                await self.api.send_message(chat_id, text, reply_markup=markup)
            except TelegramBotApiError as exc:
                await update_runtime_status(
                    {
                        "notification_last_error": str(exc)[:1000],
                        "notification_last_error_at": utc_now(),
                    }
                )

    async def _handle_edited_message(self, message: dict[str, Any]) -> None:
        connection_id = str(message.get("business_connection_id") or "")
        chat_id = int((message.get("chat") or {}).get("id") or 0)
        if not connection_id or not chat_id:
            return
        await touch_business_connection(connection_id)
        conversation = await get_conversation_by_chat(connection_id, chat_id)
        if not conversation:
            return
        await append_conversation_message(
            conversation["conversation_id"],
            direction="incoming",
            telegram_message_id=-int(message.get("message_id") or 0),
            text=str(message.get("text") or "[消息已编辑]"),
            sender_id=int((message.get("from") or {}).get("id") or 0),
            metadata={"edited_message_id": int(message.get("message_id") or 0)},
        )

    async def _handle_deleted_messages(self, payload: dict[str, Any]) -> None:
        connection_id = str(payload.get("business_connection_id") or "")
        chat_id = int((payload.get("chat") or {}).get("id") or 0)
        if connection_id:
            await touch_business_connection(connection_id)
        conversation = await get_conversation_by_chat(connection_id, chat_id)
        if not conversation:
            return
        await append_conversation_message(
            conversation["conversation_id"],
            direction="incoming",
            telegram_message_id=-abs(int((payload.get("message_ids") or [0])[0] or 0)) - 1,
            text="[消息已删除]",
            metadata={"deleted_message_ids": payload.get("message_ids") or []},
        )

    async def _digest_loop(self) -> None:
        while not self._stopping.is_set():
            try:
                for item in await due_digest_connections():
                    connection = item["connection"]
                    settings = item["settings"]
                    metrics = await usage_metrics(
                        connection_id=str(connection["connection_id"]),
                        day=str(item["day"]),
                    )
                    text = (
                        f"智能接待运营简报 · {item['day']}\n"
                        f"客户数：{metrics['customers']}\n"
                        f"AI 调用：{metrics['ai_calls']}\n"
                        f"待人工：{metrics['handoffs']}\n"
                        f"人工暂停：{metrics['human_paused']}"
                    )
                    destinations: list[int] = []
                    destination = str(settings.get("notification_destination") or "owner")
                    if destination in {"owner", "both"} and connection.get("user_chat_id"):
                        destinations.append(int(connection["user_chat_id"]))
                    if destination in {"ops", "both"} and settings.get("ops_group_id"):
                        destinations.append(int(settings["ops_group_id"]))
                    sent = False
                    for chat_id in dict.fromkeys(destinations):
                        try:
                            await self.api.send_message(chat_id, text)
                            sent = True
                        except TelegramBotApiError as exc:
                            await update_runtime_status(
                                {
                                    "digest_last_error": str(exc)[:1000],
                                    "digest_last_error_at": utc_now(),
                                }
                            )
                    if sent:
                        await mark_digest_sent(connection["connection_id"], item["day"])
            except asyncio.CancelledError:
                raise
            except Exception as exc:
                await update_runtime_status({"digest_last_error": str(exc)[:1000]})
            await asyncio.sleep(60)

    async def preview_answer(self, connection_id: str, text: str) -> dict[str, Any]:
        settings = await get_account_settings(connection_id)
        knowledge = await match_knowledge(connection_id, text)
        classification = classify_handoff(text)
        if classification or not knowledge:
            return {
                "action": "handoff",
                "reason": classification or "knowledge_not_found",
                "matched_entries": knowledge,
            }
        conversation = {"summary": ""}
        decision = await self.provider.decide(
            settings=settings,
            conversation=conversation,
            messages=[],
            knowledge=knowledge,
            customer_text=text,
        )
        return {**decision, "matched_entries": knowledge}


async def runtime_overview() -> dict[str, Any]:
    status = await runtime_status()
    status["usage"] = await usage_metrics()
    return status