TelegramBotAdmin/tests/test_business_assistant.py

from __future__ import annotations

import json
from datetime import UTC, datetime, timedelta
from types import SimpleNamespace

import pytest
from aiohttp import CookieJar
from aiohttp.test_utils import TestClient, TestServer
from pyrogram.enums import ChatMemberStatus


def connection_payload(connection_id: str = "conn-a", *, can_reply: bool = True) -> dict:
    return {
        "id": connection_id,
        "user": {"id": 900, "username": "owner", "first_name": "店主"},
        "user_chat_id": 900,
        "date": int(datetime.now(UTC).timestamp()),
        "rights": {"can_reply": can_reply, "can_read_messages": True},
        "is_enabled": True,
    }


def customer_message(
    *,
    message_id: int,
    text: str | None = "营业时间是什么？",
    chat_id: int = 501,
    sender_id: int = 501,
    sent_at: datetime | None = None,
    **values,
) -> dict:
    message = {
        "business_connection_id": "conn-a",
        "message_id": message_id,
        "date": int((sent_at or datetime.now(UTC)).timestamp()),
        "chat": {"id": chat_id, "type": "private"},
        "from": {"id": sender_id, "username": f"user{sender_id}", "first_name": "客户"},
        **values,
    }
    if text is not None:
        message["text"] = text
    return message


class FakeProvider:
    configured = True

    def __init__(self, *, error: Exception | None = None) -> None:
        self.error = error
        self.calls: list[dict] = []
        self.extraction_calls: list[dict] = []

    async def decide(self, **values):
        self.calls.append(values)
        if self.error:
            raise self.error
        entry_id = str(values["knowledge"][0]["entry_id"])
        return {
            "action": "answer",
            "reply": "每天 09:00 至 18:00 营业。",
            "handoff_reason": "",
            "matched_entry_ids": [entry_id],
            "summary": "客户询问营业时间。",
        }

    async def test_connection(self):
        return {"ok": True, "model": "test-model", "response_id": "response-1"}

    async def extract_knowledge(self, **values):
        self.extraction_calls.append(values)
        if self.error:
            raise self.error
        return [
            {
                "question": values.get("question_context") or "如何办理？",
                "aliases": [],
                "keywords": ["办理"],
                "answer": values["content"],
                "tags": ["自动采集"],
                "confidence": 0.92,
            }
        ]


class FakeBusinessApi:
    def __init__(self) -> None:
        self.sent: list[dict] = []
        self.read: list[tuple[str, int, int]] = []

    async def send_message(
        self,
        chat_id: int,
        text: str,
        *,
        business_connection_id: str = "",
        reply_markup: dict | None = None,
    ) -> dict:
        self.sent.append(
            {
                "chat_id": chat_id,
                "text": text,
                "business_connection_id": business_connection_id,
                "reply_markup": reply_markup,
            }
        )
        return {
            "message_id": 1000 + len(self.sent),
            "sender_business_bot": {"id": 999},
        }

    async def read_business_message(
        self, connection_id: str, chat_id: int, message_id: int
    ) -> None:
        self.read.append((connection_id, chat_id, message_id))


class FakeFeishuWebhook:
    def __init__(self) -> None:
        self.sent: list[dict] = []

    async def send(self, webhook_url: str, text: str, *, signing_secret: str = "") -> None:
        self.sent.append(
            {
                "webhook_url": webhook_url,
                "text": text,
                "signing_secret": signing_secret,
            }
        )

    async def send_post(
        self,
        webhook_url: str,
        *,
        title: str,
        content: list[list[dict[str, str]]],
        signing_secret: str = "",
    ) -> None:
        self.sent.append(
            {
                "webhook_url": webhook_url,
                "title": title,
                "content": content,
                "text": "\n".join(
                    "".join(segment.get("text", "") for segment in line)
                    for line in content
                ),
                "signing_secret": signing_secret,
            }
        )


async def prepare_runtime(app_modules, *, provider: FakeProvider | None = None):
    dbassistant = app_modules.load("wbb.utils.dbassistant")
    service = app_modules.load("wbb.services.business_assistant")
    await dbassistant.upsert_business_connection(connection_payload())
    await dbassistant.update_account_settings("conn-a", {"assistant_enabled": True})
    knowledge = await dbassistant.create_knowledge_entry(
        "conn-a",
        {
            "question": "营业时间是什么？",
            "aliases": ["几点营业"],
            "keywords": ["营业时间", "营业"],
            "answer": "每天 09:00 至 18:00 营业。",
            "priority": 10,
        },
    )
    selected_provider = provider or FakeProvider()
    runtime = service.BusinessAssistantRuntime(
        token="123456:" + "A" * 30,
        session=SimpleNamespace(),
        provider=selected_provider,
    )
    runtime.api = FakeBusinessApi()
    return dbassistant, service, runtime, knowledge


async def test_connection_settings_knowledge_quota_and_bot_scope(app_modules):
    dbassistant = app_modules.load("wbb.utils.dbassistant")
    await dbassistant.upsert_business_connection(connection_payload("conn-a"))
    await dbassistant.upsert_business_connection(connection_payload("conn-b"))
    settings = await dbassistant.update_account_settings(
        "conn-a",
        {
            "assistant_enabled": True,
            "account_daily_limit": 2,
            "customer_daily_limit": 1,
            "timezone": "Asia/Shanghai",
        },
    )
    first = await dbassistant.create_knowledge_entry(
        "conn-a",
        {
            "question": "营业时间",
            "aliases": ["几点开门"],
            "keywords": ["开门"],
            "answer": "九点开门。",
            "priority": 20,
        },
    )
    await dbassistant.create_knowledge_entry(
        "conn-b",
        {"question": "营业时间", "keywords": ["开门"], "answer": "十点开门。"},
    )
    await app_modules.wbb.db.business_assistant_knowledge.insert_one(
        {
            "bot_id": "foreign-bot",
            "entry_id": "foreign-entry",
            "connection_id": "conn-a",
            "question": "营业时间",
            "aliases": [],
            "keywords": ["开门"],
            "answer": "不应跨 Bot 返回。",
            "priority": 1000,
            "enabled": True,
            "updated_at": datetime.now(UTC),
        }
    )

    matched = await dbassistant.match_knowledge("conn-a", "你们几点开门？")
    assert [item["entry_id"] for item in matched] == [first["entry_id"]]
    items, total = await dbassistant.list_knowledge_entries("conn-a")
    assert total == 1
    assert items[0]["answer"] == "九点开门。"

    await app_modules.wbb.db.business_assistant_connections.insert_one(
        {
            "bot_id": "foreign-bot",
            "connection_id": "foreign-connection",
            "updated_at": datetime.now(UTC),
        }
    )
    connections, total = await dbassistant.list_business_connections(page_size=100)
    assert total == 2
    assert {item["connection_id"] for item in connections} == {"conn-a", "conn-b"}

    assert await dbassistant.reserve_ai_usage("conn-a", 101, settings) == (True, "")
    assert await dbassistant.reserve_ai_usage("conn-a", 101, settings) == (
        False,
        "customer_daily_limit",
    )
    assert await dbassistant.reserve_ai_usage("conn-a", 102, settings) == (True, "")
    assert await dbassistant.reserve_ai_usage("conn-a", 103, settings) == (
        False,
        "account_daily_limit",
    )
    usage = await dbassistant.usage_metrics(connection_id="conn-a")
    assert usage["ai_calls"] == 2
    assert usage["customers"] == 2

    with pytest.raises(dbassistant.AssistantDataError) as error:
        dbassistant.normalize_account_settings({"timezone": "invalid/timezone"})
    assert error.value.code == "invalid_setting"


async def test_feishu_settings_are_validated_and_never_exposed(app_modules):
    dbassistant = app_modules.load("wbb.utils.dbassistant")
    await dbassistant.upsert_business_connection(connection_payload())
    webhook_url = "https://open.feishu.cn/open-apis/bot/v2/hook/12345678-abcd-4321-abcd-123456789abc"
    stored = await dbassistant.update_account_settings(
        "conn-a",
        {
            "feishu_webhook_enabled": True,
            "feishu_webhook_url": webhook_url,
            "feishu_webhook_signing_secret": "signing-secret",
            "feishu_message_preview_enabled": True,
        },
    )
    assert stored["feishu_webhook_url"] == webhook_url
    assert stored["feishu_webhook_signing_secret"] == "signing-secret"

    public = dbassistant.public_account_settings(stored)
    assert public["feishu_webhook_url"] == ""
    assert public["feishu_webhook_signing_secret"] == ""
    assert public["feishu_webhook_configured"] is True
    assert public["feishu_signing_secret_configured"] is True

    connections, _total = await dbassistant.list_business_connections()
    assert connections[0]["settings"]["feishu_webhook_url"] == ""
    assert connections[0]["settings"]["feishu_webhook_configured"] is True

    with pytest.raises(dbassistant.AssistantDataError):
        dbassistant.normalize_account_settings(
            {
                "feishu_webhook_enabled": True,
                "feishu_webhook_url": "https://example.com/internal-hook",
            }
        )


def test_ai_contract_and_reply_window_validation(app_modules):
    service = app_modules.load("wbb.services.business_assistant")
    parsed = service.parse_ai_decision(
        json.dumps(
            {
                "action": "answer",
                "reply": "标准答案",
                "matched_entry_ids": ["entry-1", "foreign"],
                "summary": "摘要",
            }
        ),
        allowed_entry_ids={"entry-1"},
    )
    assert parsed["matched_entry_ids"] == ["entry-1"]

    with pytest.raises(service.AssistantProviderError):
        service.parse_ai_decision(
            '{"action":"answer","reply":"没有引用"}', allowed_entry_ids={"entry-1"}
        )
    now = datetime.now(UTC)
    assert service.business_reply_window_open(
        {"date": int((now - timedelta(hours=23)).timestamp())}, now=now
    )
    assert not service.business_reply_window_open(
        {"date": int((now - timedelta(hours=25)).timestamp())}, now=now
    )
    assert not service.business_reply_window_open({}, now=now)

    extracted = service.parse_knowledge_extraction(
        json.dumps(
            {
                "items": [
                    {
                        "question": "如何配送？",
                        "aliases": ["配送范围"],
                        "keywords": ["配送"],
                        "answer": "仅支持市区配送。",
                        "tags": ["配送"],
                        "confidence": 0.9,
                    }
                ]
            }
        )
    )
    assert extracted[0]["confidence"] == 0.9
    assert (
        service.redact_business_knowledge_text(
            "电话 138 0013 8000，邮箱 user@example.com，Telegram @private_user"
        )
        == "电话 [电话已脱敏]，邮箱 [邮箱已脱敏]，Telegram [用户名已脱敏]"
    )


async def test_source_event_candidates_publish_edit_delete_and_isolate(app_modules):
    dbassistant = app_modules.load("wbb.utils.dbassistant")
    await dbassistant.upsert_business_connection(connection_payload("conn-a"))
    await dbassistant.upsert_business_connection(connection_payload("conn-b"))
    source = await dbassistant.create_knowledge_source(
        "conn-a",
        {
            "source_type": "channel",
            "chat_id": -100100,
            "title": "官方频道",
            "publication_mode": "auto",
            "author_policy": "admins_or_allowlist",
        },
    )
    event, changed = await dbassistant.record_source_event(
        source,
        event_key="telegram:-100100:10",
        content="每天九点营业。",
        metadata={"chat_id": -100100, "message_id": 10, "trusted_author": True},
    )
    assert changed is True
    candidates = await dbassistant.replace_event_candidates(
        source,
        event,
        [
            {
                "question": "几点营业？",
                "aliases": ["营业时间"],
                "keywords": ["营业"],
                "answer": "每天九点营业。",
                "tags": ["营业"],
                "confidence": 0.95,
            }
        ],
        auto_publish=True,
    )
    candidate = candidates[0]
    first_entry_id = candidate["knowledge_entry_id"]
    first_entry = await app_modules.wbb.db.business_assistant_knowledge.find_one(
        {"bot_id": "primary", "entry_id": first_entry_id}
    )
    assert candidate["status"] == "published"
    assert first_entry["enabled"] is True

    edited_event, changed = await dbassistant.record_source_event(
        source,
        event_key="telegram:-100100:10",
        content="每天十点营业。",
        metadata={"chat_id": -100100, "message_id": 10, "trusted_author": True},
    )
    assert changed is True
    edited = await dbassistant.replace_event_candidates(
        source,
        edited_event,
        [
            {
                "question": "几点营业？",
                "aliases": [],
                "keywords": ["营业"],
                "answer": "每天十点营业。",
                "tags": ["营业"],
                "confidence": 0.96,
            }
        ],
        auto_publish=True,
    )
    assert edited[0]["knowledge_entry_id"] == first_entry_id
    updated_entry = await app_modules.wbb.db.business_assistant_knowledge.find_one(
        {"bot_id": "primary", "entry_id": first_entry_id}
    )
    assert updated_entry["answer"] == "每天十点营业。"
    assert await dbassistant.mark_source_event_deleted(
        source["source_id"], "telegram:-100100:10"
    )
    stale = await dbassistant.get_knowledge_candidate(candidate["candidate_id"])
    disabled_entry = await app_modules.wbb.db.business_assistant_knowledge.find_one(
        {"bot_id": "primary", "entry_id": first_entry_id}
    )
    assert stale["status"] == "stale"
    assert disabled_entry["enabled"] is False

    sources_a, total_a = await dbassistant.list_knowledge_sources("conn-a")
    sources_b, total_b = await dbassistant.list_knowledge_sources("conn-b")
    assert total_a == 1 and sources_a[0]["source_id"] == source["source_id"]
    assert total_b == 0 and sources_b == []
    event_indexes = await app_modules.wbb.db.business_assistant_source_events.index_information()
    assert event_indexes["expires_at_1"]["expireAfterSeconds"] == 0


async def test_business_human_reply_learns_only_when_source_enabled(app_modules):
    dbassistant, _service, runtime, _knowledge = await prepare_runtime(app_modules)
    await runtime.process_update(
        {
            "update_id": 9,
            "business_message": customer_message(
                message_id=9, text="配送到哪里？", chat_id=509, sender_id=509
            ),
        }
    )
    await runtime.process_update(
        {
            "update_id": 10,
            "business_message": customer_message(
                message_id=10, text="仅支持市区配送。", chat_id=509, sender_id=900
            ),
        }
    )
    assert runtime.provider.extraction_calls == []

    await dbassistant.create_knowledge_source(
        "conn-a",
        {
            "source_type": "business",
            "title": "人工接待对话",
            "publication_mode": "auto",
        },
    )
    await runtime.process_update(
        {
            "update_id": 15,
            "business_message": customer_message(
                message_id=15, text="支持哪些区域？", chat_id=515, sender_id=515
            ),
        }
    )
    await runtime.process_update(
        {
            "update_id": 16,
            "business_message": customer_message(
                message_id=16, text="仅支持市区配送。", chat_id=515, sender_id=900
            ),
        }
    )
    assert len(runtime.provider.extraction_calls) == 1
    candidates, total = await dbassistant.list_knowledge_candidates(
        "conn-a", status="published"
    )
    assert total == 1
    assert candidates[0]["answer"] == "仅支持市区配送。"
    entry_id = candidates[0]["knowledge_entry_id"]

    await runtime.process_update(
        {
            "update_id": 17,
            "edited_business_message": customer_message(
                message_id=16, text="仅二环内支持配送。", chat_id=515, sender_id=900
            ),
        }
    )
    edited = await dbassistant.get_knowledge_candidate(candidates[0]["candidate_id"])
    assert edited["answer"] == "仅二环内支持配送。"
    assert edited["knowledge_entry_id"] == entry_id

    await runtime.process_update(
        {
            "update_id": 18,
            "deleted_business_messages": {
                "business_connection_id": "conn-a",
                "chat": {"id": 515},
                "message_ids": [16],
            },
        }
    )
    stale = await dbassistant.get_knowledge_candidate(candidates[0]["candidate_id"])
    entry = await app_modules.wbb.db.business_assistant_knowledge.find_one(
        {"bot_id": "primary", "entry_id": entry_id}
    )
    assert stale["status"] == "stale"
    assert entry["enabled"] is False


async def test_group_collection_requires_trusted_author_for_auto_publish(app_modules):
    dbassistant, _service, runtime, _knowledge = await prepare_runtime(app_modules)
    module = app_modules.load("wbb.modules.business_assistant")
    module._runtime = runtime
    await dbassistant.create_knowledge_source(
        "conn-a",
        {
            "source_type": "group",
            "chat_id": -100300,
            "title": "运营讨论群",
            "publication_mode": "auto",
            "author_policy": "admins_or_allowlist",
        },
    )

    def group_message(message_id: int, author_id: int, text: str):
        return SimpleNamespace(
            id=message_id,
            text=text,
            caption=None,
            outgoing=False,
            is_automatic_forward=False,
            date=datetime.now(UTC),
            chat=SimpleNamespace(id=-100300),
            from_user=SimpleNamespace(id=author_id, is_bot=False),
        )

    ordinary = group_message(1, 301, "市区支持当日配送。")
    await module.process_knowledge_source_message(ordinary)
    pending, pending_total = await dbassistant.list_knowledge_candidates(
        "conn-a", status="pending"
    )
    assert pending_total == 1
    assert pending[0]["status"] == "pending"

    app_modules.app.members[(-100300, 302)] = SimpleNamespace(
        status=ChatMemberStatus.ADMINISTRATOR
    )
    admin_message = group_message(2, 302, "每周一至周五提供配送。")
    await module.process_knowledge_source_message(admin_message)
    published, published_total = await dbassistant.list_knowledge_candidates(
        "conn-a", status="published"
    )
    assert published_total == 1
    assert published[0]["source_snapshot"]["author_id"] == 302
    extraction_count = len(runtime.provider.extraction_calls)
    await module.process_knowledge_source_message(admin_message)
    assert len(runtime.provider.extraction_calls) == extraction_count


async def test_business_updates_are_idempotent_and_manual_reply_pauses(app_modules):
    dbassistant, _service, runtime, knowledge = await prepare_runtime(app_modules)
    update = {
        "update_id": 11,
        "business_message": customer_message(message_id=1),
    }
    await runtime.process_update(update)
    await runtime.process_update(update)

    assert len(runtime.api.sent) == 1
    assert runtime.api.sent[0]["business_connection_id"] == "conn-a"
    assert runtime.api.read == [("conn-a", 501, 1)]
    assert len(runtime.provider.calls) == 1
    assert runtime.provider.calls[0]["knowledge"][0]["entry_id"] == knowledge["entry_id"]
    conversation = await dbassistant.get_conversation_by_chat("conn-a", 501)
    assert conversation["summary"] == "客户询问营业时间。"
    messages = await dbassistant.recent_conversation_messages(
        conversation["conversation_id"], limit=10
    )
    assert [item["direction"] for item in messages] == ["incoming", "assistant"]
    assert messages[0]["expires_at"] - messages[0]["created_at"] == timedelta(days=30)
    message_indexes = await app_modules.wbb.db.business_assistant_messages.index_information()
    assert message_indexes["expires_at_1"]["expireAfterSeconds"] == 0

    await runtime.process_update(
        {
            "update_id": 12,
            "business_message": customer_message(
                message_id=2, text="账号本人回复", sender_id=900
            ),
        }
    )
    paused = await dbassistant.get_conversation(conversation["conversation_id"])
    assert paused["status"] == "human_paused"
    assert paused["handoff_reason"] == "human_reply_detected"
    assert paused["customer"]["id"] == 501

    await runtime.process_update(
        {
            "update_id": 13,
            "business_message": customer_message(
                message_id=3,
                sender_business_bot={"id": 999},
            ),
        }
    )
    await runtime.process_update(
        {
            "update_id": 14,
            "business_message": customer_message(
                message_id=4,
                is_from_offline=True,
            ),
        }
    )
    assert len(runtime.api.sent) == 1


async def test_every_user_message_notifies_feishu_once(app_modules):
    dbassistant, _service, runtime, _knowledge = await prepare_runtime(app_modules)
    webhook_url = "https://open.feishu.cn/open-apis/bot/v2/hook/12345678-abcd-4321-abcd-123456789abc"
    await dbassistant.update_account_settings(
        "conn-a",
        {
            "assistant_enabled": False,
            "feishu_webhook_enabled": True,
            "feishu_webhook_url": webhook_url,
            "feishu_webhook_signing_secret": "secret",
            "feishu_message_preview_enabled": False,
        },
    )
    feishu = FakeFeishuWebhook()
    runtime.feishu = feishu

    await runtime.process_update(
        {"update_id": 101, "business_message": customer_message(message_id=1)}
    )
    await runtime.process_update(
        {"update_id": 102, "business_message": customer_message(message_id=1)}
    )
    assert len(feishu.sent) == 1
    assert feishu.sent[0]["webhook_url"] == webhook_url
    assert feishu.sent[0]["signing_secret"] == "secret"
    assert "营业时间是什么" not in feishu.sent[0]["text"]
    assert feishu.sent[0]["title"] == "消息提醒"
    assert "用户：" in feishu.sent[0]["text"]
    assert "客户：" not in feishu.sent[0]["text"]
    assert "消息标识：" not in feishu.sent[0]["text"]
    assert feishu.sent[0]["content"][1] == [
        {"tag": "text", "text": "用户："},
        {"tag": "a", "text": "客户", "href": "https://t.me/user501"},
    ]

    await dbassistant.update_account_settings(
        "conn-a", {"feishu_message_preview_enabled": True}
    )
    await runtime.process_update(
        {
            "update_id": 103,
            "business_message": customer_message(
                message_id=2, text="第二条用户消息"
            ),
        }
    )
    assert len(feishu.sent) == 2
    assert "第二条用户消息" in feishu.sent[1]["text"]

    await runtime.process_update(
        {
            "update_id": 1030,
            "business_message": customer_message(
                message_id=20,
                text=None,
                caption="图片说明",
                photo=[{"file_id": "photo"}],
            ),
        }
    )
    assert len(feishu.sent) == 3
    assert "类型：非文本" in feishu.sent[2]["text"]
    assert "内容：图片说明" in feishu.sent[2]["text"]

    await runtime.process_update(
        {
            "update_id": 1031,
            "business_message": customer_message(
                message_id=21,
                chat_id=502,
                sender_id=502,
                **{"from": {"id": 502, "first_name": "无用户名用户"}},
            ),
        }
    )
    assert len(feishu.sent) == 4
    assert feishu.sent[3]["content"][1] == [
        {"tag": "text", "text": "用户："},
        {"tag": "text", "text": "无用户名用户"},
    ]

    await runtime.process_update(
        {
            "update_id": 104,
            "business_message": customer_message(
                message_id=3, text="账号本人回复", sender_id=900
            ),
        }
    )
    await runtime.process_update(
        {
            "update_id": 105,
            "business_message": customer_message(message_id=4, is_from_offline=True),
        }
    )
    await runtime.process_update(
        {
            "update_id": 106,
            "business_message": customer_message(
                message_id=5, sender_business_bot={"id": 999}
            ),
        }
    )
    assert len(feishu.sent) == 4
    conversation = await dbassistant.get_conversation_by_chat("conn-a", 501)
    messages = await dbassistant.recent_conversation_messages(
        conversation["conversation_id"], limit=10
    )
    incoming = [item for item in messages if item["telegram_message_id"] == 2][0]
    assert incoming["metadata"]["feishu_notification_status"] == "sent"


async def test_handoff_sensitive_non_text_provider_error_and_expired_window(app_modules):
    dbassistant, service, runtime, _knowledge = await prepare_runtime(app_modules)
    await dbassistant.update_account_settings(
        "conn-a",
        {
            "notification_destination": "both",
            "ops_group_id": -100123,
            "account_daily_limit": 1,
            "customer_daily_limit": 1,
        },
    )

    await runtime.process_update(
        {
            "update_id": 21,
            "business_message": customer_message(
                message_id=1, text="我要退款", chat_id=601, sender_id=601
            ),
        }
    )
    sensitive = await dbassistant.get_conversation_by_chat("conn-a", 601)
    assert sensitive["status"] == "handoff"
    assert sensitive["handoff_reason"] == "sensitive_request"
    assert runtime.api.sent[-1]["business_connection_id"] == ""
    assert "message_id=1" in runtime.api.sent[-1]["text"]
    assert {item["chat_id"] for item in runtime.api.sent[-2:]} == {900, -100123}

    await runtime.process_update(
        {
            "update_id": 22,
            "business_message": customer_message(
                message_id=2, text=None, chat_id=602, sender_id=602, photo=[{"file_id": "x"}]
            ),
        }
    )
    unsupported = await dbassistant.get_conversation_by_chat("conn-a", 602)
    assert unsupported["handoff_reason"] == "unsupported_message"

    runtime.provider = FakeProvider(error=service.AssistantProviderError("模型超时"))
    await runtime.process_update(
        {
            "update_id": 23,
            "business_message": customer_message(
                message_id=3, chat_id=603, sender_id=603
            ),
        }
    )
    provider_failed = await dbassistant.get_conversation_by_chat("conn-a", 603)
    assert provider_failed["handoff_reason"] == "provider_error"

    await runtime.process_update(
        {
            "update_id": 230,
            "business_message": customer_message(
                message_id=30, chat_id=605, sender_id=605
            ),
        }
    )
    quota_handoff = await dbassistant.get_conversation_by_chat("conn-a", 605)
    assert quota_handoff["handoff_reason"] == "account_daily_limit"

    sent_before = len(runtime.api.sent)
    await runtime.process_update(
        {
            "update_id": 24,
            "business_message": customer_message(
                message_id=4,
                chat_id=604,
                sender_id=604,
                sent_at=datetime.now(UTC) - timedelta(hours=25),
            ),
        }
    )
    expired = await dbassistant.get_conversation_by_chat("conn-a", 604)
    assert expired["handoff_reason"] == "reply_window_expired"
    new_sends = runtime.api.sent[sent_before:]
    assert len(new_sends) == 2
    assert all(item["business_connection_id"] == "" for item in new_sends)


async def test_failed_update_goes_to_dead_letter_without_blocking(app_modules, monkeypatch):
    dbassistant, _service, runtime, _knowledge = await prepare_runtime(app_modules)

    async def fail(_message):
        raise RuntimeError("persistent failure")

    async def no_sleep(_seconds):
        return None

    runtime._handle_business_message = fail
    monkeypatch.setattr("wbb.services.business_assistant.asyncio.sleep", no_sleep)
    await runtime.process_update(
        {"update_id": 31, "business_message": customer_message(message_id=1)}
    )

    dead_letter = await app_modules.wbb.db.business_assistant_dead_letters.find_one(
        {"bot_id": "primary", "update_id": 31}
    )
    update = await app_modules.wbb.db.business_assistant_updates.find_one(
        {"bot_id": "primary", "update_id": 31}
    )
    assert dead_letter["error"] == "persistent failure"
    assert update["status"] == "done"
    assert await dbassistant.claim_update(31) is False


async def test_connection_updates_revoke_permissions_and_persist_offset(app_modules):
    service = app_modules.load("wbb.services.business_assistant")
    dbassistant = app_modules.load("wbb.utils.dbassistant")
    runtime = service.BusinessAssistantRuntime(
        token="123456:" + "A" * 30,
        session=SimpleNamespace(),
        provider=FakeProvider(),
    )
    runtime.api = FakeBusinessApi()
    await runtime.process_update(
        {"update_id": 41, "business_connection": connection_payload(can_reply=True)}
    )
    connected = await dbassistant.get_business_connection("conn-a")
    assert connected["is_enabled"] is True
    assert connected["rights"]["can_reply"] is True
    await dbassistant.update_account_settings("conn-a", {"assistant_enabled": True})

    revoked = connection_payload(can_reply=False)
    revoked["is_enabled"] = False
    await runtime.process_update({"update_id": 42, "business_connection": revoked})
    disconnected = await dbassistant.get_business_connection("conn-a")
    assert disconnected["is_enabled"] is False
    assert "can_reply" not in disconnected["rights"]
    await runtime.process_update(
        {"update_id": 420, "business_message": customer_message(message_id=20)}
    )
    assert runtime.api.sent == []

    await dbassistant.save_update_offset(43)
    assert await dbassistant.load_update_offset() == 43
    await dbassistant.save_update_offset(44)
    assert await dbassistant.load_update_offset() == 44


class StartupApi:
    def __init__(self, *, supported: bool, webhook_url: str = "") -> None:
        self.supported = supported
        self.webhook_url = webhook_url

    async def get_me(self):
        return {"username": "business_bot", "can_connect_to_business": self.supported}

    async def get_webhook_info(self):
        return {"url": self.webhook_url}


async def test_runtime_blocks_webhook_conflict_and_disabled_business_mode(app_modules):
    service = app_modules.load("wbb.services.business_assistant")
    runtime = service.BusinessAssistantRuntime(
        token="123456:" + "A" * 30,
        session=SimpleNamespace(),
        provider=FakeProvider(),
    )
    runtime.api = StartupApi(supported=True, webhook_url="https://hooks.example.com/tg")
    webhook_status = await runtime.start()
    assert webhook_status["polling_state"] == "blocked"
    assert webhook_status["webhook_conflict"] is True
    assert runtime._poll_task is None

    runtime.api = StartupApi(supported=False)
    business_status = await runtime.start()
    assert business_status["polling_state"] == "blocked"
    assert business_status["business_mode_supported"] is False
    assert "BotFather" in business_status["last_error"]


class FakeResponse:
    def __init__(self, status: int, payload: dict) -> None:
        self.status = status
        self.payload = payload
        self.headers: dict[str, str] = {}

    async def __aenter__(self):
        return self

    async def __aexit__(self, *_args):
        return None

    async def json(self, **_kwargs):
        return self.payload


class RetrySession:
    def __init__(self, responses: list[FakeResponse]) -> None:
        self.responses = responses
        self.calls = 0

    def post(self, *_args, **_kwargs):
        response = self.responses[self.calls]
        self.calls += 1
        return response


async def test_telegram_api_retries_429_and_5xx(app_modules, monkeypatch):
    service = app_modules.load("wbb.services.business_assistant")
    session = RetrySession(
        [
            FakeResponse(
                429,
                {
                    "ok": False,
                    "error_code": 429,
                    "description": "Too Many Requests",
                    "parameters": {"retry_after": 3},
                },
            ),
            FakeResponse(
                502,
                {"ok": False, "error_code": 502, "description": "Bad Gateway"},
            ),
            FakeResponse(200, {"ok": True, "result": {"id": 999}}),
        ]
    )
    sleeps: list[int] = []

    async def record_sleep(seconds):
        sleeps.append(seconds)

    monkeypatch.setattr("wbb.services.business_assistant.asyncio.sleep", record_sleep)
    api = service.TelegramBusinessApi("123456:" + "A" * 30, session)
    assert await api.get_me() == {"id": 999}
    assert session.calls == 3
    assert sleeps == [3, 2]


async def test_feishu_webhook_signature_and_success_contract(app_modules):
    service = app_modules.load("wbb.services.business_assistant")
    response = FakeResponse(200, {"code": 0, "msg": "success"})

    class CaptureSession:
        def __init__(self) -> None:
            self.payload = None

        def post(self, _url, *, json, timeout):
            self.payload = json
            assert timeout.total == 8
            return response

    session = CaptureSession()
    client = service.FeishuWebhookClient(session)
    await client.send(
        "https://open.feishu.cn/open-apis/bot/v2/hook/12345678-abcd-4321-abcd-123456789abc",
        "消息提醒",
        signing_secret="secret",
    )
    assert session.payload["msg_type"] == "text"
    assert session.payload["content"]["text"] == "消息提醒"
    assert session.payload["timestamp"]
    assert session.payload["sign"]

    post_content = [
        [
            {"tag": "text", "text": "用户："},
            {"tag": "a", "text": "测试用户", "href": "https://t.me/test_user"},
        ]
    ]
    await client.send_post(
        "https://open.feishu.cn/open-apis/bot/v2/hook/12345678-abcd-4321-abcd-123456789abc",
        title="消息提醒",
        content=post_content,
    )
    assert session.payload == {
        "msg_type": "post",
        "content": {
            "post": {
                "zh_cn": {
                    "title": "消息提醒",
                    "content": post_content,
                }
            }
        },
    }


async def _login_and_change_password(client: TestClient) -> str:
    login = await client.post(
        "/api/admin/v1/auth/login",
        json={"username": "admin", "password": "qwe0.123456"},
    )
    login_data = (await login.json())["data"]
    changed = await client.put(
        "/api/admin/v1/auth/password",
        headers={"X-CSRF-Token": login_data["csrf_token"]},
        json={
            "current_password": "qwe0.123456",
            "new_password": "changed-pass-123",
        },
    )
    return (await changed.json())["data"]["csrf_token"]


async def test_business_assistant_api_permission_csrf_audit_and_clear(app_modules):
    dbassistant = app_modules.load("wbb.utils.dbassistant")
    await dbassistant.upsert_business_connection(connection_payload())
    conversation = await dbassistant.get_or_create_conversation(
        "conn-a", 701, customer={"id": 701, "first_name": "待清理客户"}
    )
    await dbassistant.append_conversation_message(
        conversation["conversation_id"],
        direction="incoming",
        telegram_message_id=1,
        text="请清理",
    )
    await dbassistant.update_conversation_summary(conversation["conversation_id"], "待清理摘要")

    admin_api = app_modules.load("wbb.admin.api")
    application = admin_api.build_admin_application()
    await application["admin_api"].initialize()
    client = TestClient(TestServer(application), cookie_jar=CookieJar(unsafe=True))
    await client.start_server()
    try:
        csrf = await _login_and_change_password(client)
        app_modules.wbb.BOT_PERMISSIONS = set()
        denied = await client.get("/api/admin/v1/business-assistant/status")
        assert denied.status == 403

        app_modules.wbb.BOT_PERMISSIONS = {"business_assistant.manage"}
        no_csrf = await client.put(
            "/api/admin/v1/business-assistant/settings",
            json={"connection_id": "conn-a", "assistant_enabled": True, "confirm": True},
        )
        assert no_csrf.status == 403
        unconfirmed = await client.put(
            "/api/admin/v1/business-assistant/settings",
            headers={"X-CSRF-Token": csrf},
            json={"connection_id": "conn-a", "assistant_enabled": True},
        )
        assert unconfirmed.status == 409

        saved = await client.put(
            "/api/admin/v1/business-assistant/settings",
            headers={"X-CSRF-Token": csrf},
            json={
                "connection_id": "conn-a",
                "assistant_enabled": True,
                "feishu_webhook_enabled": True,
                "feishu_webhook_url": "https://open.feishu.cn/open-apis/bot/v2/hook/12345678-abcd-4321-abcd-123456789abc",
                "feishu_webhook_signing_secret": "api-secret",
                "confirm": True,
            },
        )
        assert saved.status == 200
        saved_data = (await saved.json())["data"]
        assert saved_data["assistant_enabled"] is True
        assert saved_data["feishu_webhook_url"] == ""
        assert saved_data["feishu_webhook_signing_secret"] == ""
        assert saved_data["feishu_webhook_configured"] is True
        stored_settings = await dbassistant.get_account_settings("conn-a")
        assert stored_settings["feishu_webhook_signing_secret"] == "api-secret"
        listed_connections = await client.get(
            "/api/admin/v1/business-assistant/connections?page=1&page_size=100"
        )
        listed_settings = (await listed_connections.json())["data"]["items"][0][
            "settings"
        ]
        assert listed_settings["feishu_webhook_url"] == ""
        assert listed_settings["feishu_webhook_signing_secret"] == ""

        class FakeRuntime:
            def __init__(self) -> None:
                self.calls = []

            async def test_feishu_webhook(self, connection_id, overrides):
                self.calls.append((connection_id, overrides))
                return {"ok": True, "account": "店主"}

        module = app_modules.load("wbb.modules.business_assistant")
        fake_runtime = FakeRuntime()
        module._runtime = fake_runtime
        tested = await client.post(
            "/api/admin/v1/business-assistant/feishu-webhook/test",
            headers={"X-CSRF-Token": csrf},
            json={"connection_id": "conn-a", "confirm": True},
        )
        assert tested.status == 200
        assert (await tested.json())["data"]["ok"] is True
        assert fake_runtime.calls == [("conn-a", {})]

        created = await client.post(
            "/api/admin/v1/business-assistant/knowledge",
            headers={"X-CSRF-Token": csrf},
            json={
                "connection_id": "conn-a",
                "question": "配送范围",
                "keywords": ["配送"],
                "answer": "仅限市区。",
                "confirm": True,
            },
        )
        assert created.status == 201
        listed = await client.get(
            "/api/admin/v1/business-assistant/knowledge?connection_id=conn-a"
        )
        assert (await listed.json())["data"]["total"] == 1

        created_source = await client.post(
            "/api/admin/v1/business-assistant/knowledge-sources",
            headers={"X-CSRF-Token": csrf},
            json={
                "connection_id": "conn-a",
                "source_type": "business",
                "title": "人工对话",
                "publication_mode": "review",
                "confirm": True,
            },
        )
        assert created_source.status == 201
        source = (await created_source.json())["data"]
        event, _ = await dbassistant.record_source_event(
            source,
            event_key="business:701:8",
            content="市区可配送。",
            metadata={"chat_id": 701, "message_id": 8, "trusted_author": True},
        )
        candidates = await dbassistant.replace_event_candidates(
            source,
            event,
            [
                {
                    "question": "配送范围？",
                    "keywords": ["配送"],
                    "answer": "市区可配送。",
                    "confidence": 0.9,
                }
            ],
            auto_publish=False,
        )
        candidate_page = await client.get(
            "/api/admin/v1/business-assistant/knowledge-candidates"
            "?connection_id=conn-a&status=pending"
        )
        assert candidate_page.status == 200
        assert (await candidate_page.json())["data"]["total"] == 1
        approved = await client.post(
            "/api/admin/v1/business-assistant/knowledge-candidates/"
            f"{candidates[0]['candidate_id']}/actions",
            headers={"X-CSRF-Token": csrf},
            json={"action": "approve", "confirm": True},
        )
        assert approved.status == 200
        assert (await approved.json())["data"]["status"] == "published"

        cleared = await client.post(
            f"/api/admin/v1/business-assistant/conversations/{conversation['conversation_id']}/actions",
            headers={"X-CSRF-Token": csrf},
            json={"action": "clear", "confirm": True},
        )
        assert cleared.status == 200
        detail = await dbassistant.conversation_detail(conversation["conversation_id"])
        assert detail["summary"] == ""
        assert detail["messages"] == []

        audit = await app_modules.wbb.db.admin_audit_logs.find_one(
            {"action": "business_assistant.conversation.clear"}
        )
        assert audit["success"] is True
    finally:
        await client.close()