"""Inbound half of the assistant channel — Feishu event intake and dispatch.

**Why this is not its own endpoint.** A Feishu app has exactly ONE event
subscription URL, and the platform's is already
``POST /api/v1/approval/feishu/callback`` (approval instance events). Adding the
``im.message.receive_v1`` subscription to the same app therefore delivers chat
messages to that same battle-tested handler — HMAC signature, verification-token
fallback, freshness window, nonce replay guard and webhook rate limit all
already enforced there. This module plugs in one fork after verification:
:func:`try_claim_event` takes the events the assistant owns and leaves every
other event untouched on the approval path.

**Why intake acks before it answers.** Feishu redelivers an event that is not
acknowledged quickly, and a model turn takes seconds. The handler claims the
event id, returns 200, and runs the turn in the background — so a slow answer
can never turn into a duplicate conversation.

**Private chats and rooms come through here together.** A group message is the
same event type with a different ``chat_type``, so it is parsed by the same
function and answered by the same turn — what differs is decided in two places
and nowhere else: :mod:`app.assistant.group_chat` answers "is this addressed to
me, and is this kind of room open", and :func:`reply_target` answers "where does
the reply go". Everything between them is the assistant it always was.
"""
from __future__ import annotations

import json
import logging
import re
from dataclasses import dataclass, replace
from typing import Any

from sqlalchemy.ext.asyncio import AsyncSession

from app.assistant import group_chat, help_card, messages, reply_language
from app.assistant.channel import ReplyTarget
from app.assistant.identity import SenderStatus, resolve_sender
from app.assistant.settings import (
    DEFAULT_TURNS_PER_HOUR,
    SETTING_TURNS_PER_HOUR,
    assistant_enabled,
    bounded_int_setting,
)
from app.audit.service import AuditService
from app.core import background
from app.core.locale import DEFAULT_LANG, Lang
from app.core.redis_client import get_redis
from app.identity import preferences

logger = logging.getLogger(__name__)

EVENT_MESSAGE_RECEIVE = "im.message.receive_v1"
EVENT_BOT_MENU = "application.bot.menu_v6"
#: 机器人被拉进 / 移出一个群。认领它们是为了两件事:被拉进去时**自我介绍**
#: (一个不说话又不解释自己的机器人在群里是纯噪音),被移出时把工单群的台账
#: 标成「群已经不在了」——否则后续每一次往那个群发消息都是一次注定失败的外部
#: 调用,而失败原因看起来像飞书故障。
EVENT_BOT_ADDED = "im.chat.member.bot.added_v1"
EVENT_BOT_REMOVED = "im.chat.member.bot.deleted_v1"
# These are deliberately stable contracts with the Feishu menu configuration.
# A menu event has an employee identity but no P2P chat id, so it must trigger
# only deterministic channel controls — never an unscoped model turn.
MENU_RESET = "assistant.menu.reset"
MENU_START = "assistant.menu.start"
MENU_HELP = "assistant.menu.help"
_OWNED_MENU_KEYS = frozenset({MENU_RESET, MENU_START, MENU_HELP})
#: Event types the assistant owns. Anything not listed stays with approval.
ASSISTANT_EVENT_TYPES = frozenset({
    EVENT_MESSAGE_RECEIVE,
    EVENT_BOT_MENU,
    EVENT_BOT_ADDED,
    EVENT_BOT_REMOVED,
})

_EVENT_DEDUP_PREFIX = "assistant:event:"
_EVENT_DEDUP_TTL_SECONDS = 6 * 60 * 60
_RATE_PREFIX = "assistant:turns:"
_RATE_NOTICE_PREFIX = "assistant:turns_notified:"
_RATE_WINDOW_SECONDS = 60 * 60
# INCR + conditional EXPIRE in one round trip: a crash between the two would
# otherwise strand a counter with no TTL and lock the sender out permanently.
_RATE_SCRIPT = """
local current = redis.call('INCR', KEYS[1])
local ttl = redis.call('TTL', KEYS[1])
if current == 1 or ttl < 0 then
  redis.call('EXPIRE', KEYS[1], ARGV[1])
end
return current
"""

_CHAT_TYPE_P2P = "p2p"
_CHAT_TYPE_GROUP = "group"
#: 每个群的回合计数,和每人计数各自独立(见 ``group_chat`` 那条注释)。
_GROUP_RATE_PREFIX = "assistant:group_turns:"
#: 「这个群里我不应答」这句话每个群最多说一次。沉默会被读成「机器人坏了」并把
#: 人送去问 IT;每次都回一句则是刷屏。
_GROUP_DECLINE_PREFIX = "assistant:group_declined:"
#: 开关关着这件事一天说一次就够 —— 它在管理员改设置之前不会变。
_GROUP_DECLINE_TTL_SECONDS = 24 * 60 * 60
#: 🔴 限速**按窗口**说,不按天。额度是每小时的,所以一个房间上午撞到上限、
#: 下午再撞到时必须再说一次 —— 沿用 24 小时的话,下午那次是纯沉默,而屋里的人
#: 完全无从知道「它为什么又不说话了」。每人那一侧的通知本来就是按窗口
#: (``_RATE_WINDOW_SECONDS``),两边保持一致。
_GROUP_RATE_NOTICE_TTL_SECONDS = _RATE_WINDOW_SECONDS
_MESSAGE_TYPE_TEXT = "text"
_SENDER_TYPE_USER = "user"
# A question longer than this is a paste, not a question; the kernel would
# spend tokens on it and answer worse. Trim at intake, one place.
_MAX_QUESTION_CHARS = 2000

# An asset tag is a human-facing identifier, not the UUID the approval service
# needs.  Employees naturally paste the tag that the assistant just showed
# them, so recognise that continuation before handing the short message to the
# model (which otherwise treats it as a brand-new question).
_ASSET_TAG = re.compile(r"\bP-[A-Z0-9-]{6,64}\b", re.IGNORECASE)
_BUYBACK_WORDS = ("回购", "买断", "buyback", "repurchase", "buy out")


@dataclass(frozen=True)
class IncomingMessage:
    event_id: str
    message_id: str
    chat_id: str
    chat_type: str
    message_type: str
    text: str
    sender_open_id: str
    sender_union_id: str | None
    #: When the employee REPLIED to an existing message rather than
    #: starting a new one, Feishu sets these. ``parent_id`` is the
    #: message replied to; ``root_id`` is the head of the thread.
    #:
    #: The platform pushes ~3,800 notification cards a month through the
    #: same app, into the same 1:1 chat this bot listens on — so a reply
    #: under a card has always reached the assistant. It just arrived
    #: with no idea what it was a reply TO, which is why 「这台我早就还了」
    #: could never resolve 「这台」. Parsing these two fields is the whole
    #: of the missing link on the inbound side.
    parent_message_id: str | None = None
    root_message_id: str | None = None
    #: 这条消息里的每一次 @。群里靠它回答「这句话是在跟我说吗」,并且
    #: 「转给 @张三」里的那个人也从这里来 —— 名字是正文里的字,``open_id``
    #: 是飞书验签过的身份,两者不能互相替代。
    mentions: tuple[group_chat.Mention, ...] = ()

    @property
    def is_supported(self) -> bool:
        """Whether the conversation kernel can act on this message at all."""
        return self.message_type == _MESSAGE_TYPE_TEXT and bool(self.text)

    @property
    def is_group(self) -> bool:
        return self.chat_type == _CHAT_TYPE_GROUP


def event_type_of(payload: dict[str, Any]) -> str | None:
    """Event type from either envelope shape (v2 header, or legacy top level)."""
    header = payload.get("header")
    if isinstance(header, dict):
        value = header.get("event_type")
        if isinstance(value, str) and value:
            return value
    value = payload.get("type")
    return value if isinstance(value, str) and value else None


def claims_event(payload: dict[str, Any]) -> bool:
    event_type = event_type_of(payload)
    if event_type == EVENT_BOT_MENU:
        event = payload.get("event")
        return isinstance(event, dict) and event.get("event_key") in _OWNED_MENU_KEYS
    return event_type in ASSISTANT_EVENT_TYPES


def _extract_text(message: dict[str, Any]) -> str:
    """Feishu ships message content as a JSON *string*, not an object."""
    raw = message.get("content")
    if not isinstance(raw, str):
        return ""
    try:
        parsed = json.loads(raw)
    except (ValueError, TypeError):
        return ""
    if not isinstance(parsed, dict):
        return ""
    text = parsed.get("text")
    if not isinstance(text, str):
        return ""
    return text.strip()[:_MAX_QUESTION_CHARS]


def parse_message_event(payload: dict[str, Any]) -> IncomingMessage | None:
    """Normalise a ``im.message.receive_v1`` push, or ``None`` if unusable.

    Returns ``None`` — meaning "claimed and deliberately ignored" — for messages
    sent by an application and for chat types the channel does not serve.
    Bot-authored messages are dropped before anything else can react to them,
    because a channel that answers its own output is one deploy away from an
    infinite loop.

    **Group messages are parsed, not answered.** Whether a group message is
    addressed to the assistant depends on an ``open_id`` comparison that needs
    an API round trip (:func:`group_chat.bot_open_id`), so it cannot happen in
    this pure function; the handler decides. Everything a group message needs
    for that decision — the chat type and the structured mentions — is carried
    out of here.
    """
    event = payload.get("event")
    if not isinstance(event, dict):
        return None
    message = event.get("message")
    sender = event.get("sender")
    if not isinstance(message, dict) or not isinstance(sender, dict):
        return None
    if sender.get("sender_type") != _SENDER_TYPE_USER:
        return None
    sender_id = sender.get("sender_id")
    if not isinstance(sender_id, dict):
        return None
    open_id = str(sender_id.get("open_id") or "")
    if not open_id:
        return None
    chat_type = str(message.get("chat_type") or "")
    if chat_type not in (_CHAT_TYPE_P2P, _CHAT_TYPE_GROUP):
        # Topic groups and whatever Feishu adds next: unknown room shapes get
        # the old behaviour (claimed, ignored) rather than a guess at how to
        # address them back.
        return None
    raw_header = payload.get("header")
    header: dict[str, Any] = raw_header if isinstance(raw_header, dict) else {}
    union_id = sender_id.get("union_id")
    parent_id = message.get("parent_id")
    root_id = message.get("root_id")
    return IncomingMessage(
        event_id=str(header.get("event_id") or message.get("message_id") or ""),
        message_id=str(message.get("message_id") or ""),
        chat_id=str(message.get("chat_id") or ""),
        chat_type=chat_type,
        message_type=str(message.get("message_type") or ""),
        text=_extract_text(message),
        sender_open_id=open_id,
        sender_union_id=str(union_id) if union_id else None,
        # Bounded at the column width the outbox stores, so a hostile or
        # malformed envelope cannot turn into an oversized bind value on
        # the lookup below.
        parent_message_id=str(parent_id)[:64] if parent_id else None,
        root_message_id=str(root_id)[:64] if root_id else None,
        mentions=group_chat.parse_mentions(message),
    )


def _selected_asset_tag(text: str) -> str | None:
    """Return one canonical asset tag from a terse follow-up, if present."""
    match = _ASSET_TAG.search(text.upper())
    return match.group(0) if match else None


def _menu_sender(payload: dict[str, Any]) -> tuple[str, str | None] | None:
    """Read the actor of ``application.bot.menu_v6`` without trusting text.

    Feishu's bot-menu envelope names the actor under ``event.operator`` (unlike
    IM messages, where it is ``event.sender.sender_id``).  Both identifiers are
    fed into the same server-side identity resolver used for chat turns.
    """
    event = payload.get("event")
    if not isinstance(event, dict):
        return None
    operator = event.get("operator")
    if not isinstance(operator, dict):
        return None
    raw_id = operator.get("operator_id")
    operator_id = raw_id if isinstance(raw_id, dict) else operator
    open_id = str(operator_id.get("open_id") or "")
    if not open_id:
        return None
    union_id = operator_id.get("union_id")
    return open_id, str(union_id) if union_id else None


async def _offer_selected_buyback(
    db: AsyncSession,
    *,
    person: Any,
    incoming: IncomingMessage,
    lang: Any,
    trace_id: str,
) -> bool:
    """Turn a selected, employee-owned laptop tag into a confirmation card.

    This is intentionally narrower than intent classification: it only runs
    when the current thread already discussed a buyback and the pasted tag is
    one of the employee's current assets.  The UUID and display name come from
    the asset record, never from model memory or from the employee's text.
    """
    from app.assistant import channel, confirm_card, session  # noqa: PLC0415
    from app.assistant.actions import (  # noqa: PLC0415
        ACTION_COMPUTER_BUYBACK,
        get_action,
    )
    from app.assistant.proposals import create_proposal  # noqa: PLC0415
    from app.domains.asset.service import AssetService  # noqa: PLC0415

    tag = _selected_asset_tag(incoming.text)
    if not tag:
        return False
    history = await session.load_history(
        db, person_id=person.id, chat_id=incoming.chat_id,
    )
    if not any(
        any(word in str(turn.get("content") or "").lower() for word in _BUYBACK_WORDS)
        for turn in history
    ):
        return False
    asset = next(
        (
            item for item in await AssetService(db).get_user_assets(person.id)
            if str(item.asset_tag or "").upper() == tag
        ),
        None,
    )
    if asset is None:
        return False

    action = get_action(ACTION_COMPUTER_BUYBACK)
    assert action is not None  # curated action, fixed at import time
    display = str(asset.asset_name or " ".join(
        part for part in (asset.brand, asset.model) if part
    ) or asset.asset_tag)
    params = action.normalise({
        "asset_id": str(asset.id),
        "asset_name": f"{display}（{asset.asset_tag}）",
        "reason": f"员工申请回购设备 {asset.asset_tag}",
    })
    issued = await create_proposal(
        db,
        person=person,
        action_key=action.key,
        params=params,
        conversation_id=None,
        trace_id=trace_id,
    )
    await db.commit()
    try:
        url = await confirm_card.confirm_url(
            db, token=issued.token, digest=issued.proposal.params_hash,
        )
        delivered = await channel.send_card(
            db,
            # 单聊限定的分支(调用方已经挡住了群),所以这里的目标就是本人。
            to=ReplyTarget.dm(incoming.sender_open_id),
            card=confirm_card.build_confirmation_card(
                action=action,
                params=params,
                digest=issued.proposal.params_hash,
                url=url,
                lang=lang,
                proposal_id=str(issued.proposal.id),
            ),
            idempotency_key=f"assistant-proposal-{issued.proposal.id}",
        )
    except Exception:
        logger.exception("assistant: buyback card delivery failed person=%s", person.id)
        delivered = False
    if not delivered:
        await channel.send_text(
            db, to=ReplyTarget.dm(incoming.sender_open_id),
            text=messages.offer_failed(lang),
            idempotency_key=incoming.event_id or None,
        )
        return True
    await session.save_history(
        db,
        person_id=person.id,
        chat_id=incoming.chat_id,
        history=[*history, {"role": "user", "content": incoming.text}, {
            "role": "assistant", "content": "已发送电脑回购确认卡片，等待员工确认。",
        }],
    )
    return True


async def claim_event(event_id: str) -> bool:
    """Claim an event id exactly once. Fail-open when Redis is unavailable.

    A Redis outage degrades to "possibly answer twice", never to "silently drop
    the employee's message" — the same trade the approval webhook already makes.
    """
    if not event_id:
        return True
    try:
        redis = get_redis()
        claimed = await redis.set(
            f"{_EVENT_DEDUP_PREFIX}{event_id}", "1",
            nx=True, ex=_EVENT_DEDUP_TTL_SECONDS,
        )
        return bool(claimed)
    except Exception as exc:  # noqa: BLE001 — fail-open on Redis fault
        logger.warning("assistant: event dedup unavailable (%s) — accepting", exc)
        return True


async def _refuse_to(
    db: AsyncSession,
    *,
    target: ReplyTarget,
    group: group_chat.GroupContext | None,
    chat_id: str,
    reason: str,
    text: str,
    event_id: str | None,
) -> None:
    """说一次「为什么不能应答」——在群里**每个房间每天最多一次**。

    单聊里这句话是给当事人的指引,每次都该说。群里它是给一屋子人的通报,而且
    内容是**某一个人**的身份问题:一个没绑定的人(外部访客、邮箱对不上的人)
    每 @ 一次机器人就让房间里多一条关于他的通报,而这句话对屋里其他人没有任何
    意义 —— 更要紧的是,群频次闸排在身份分支**之后**,所以这条路是没有上限的:
    一个人可以让机器人在群里无限说话。

    复用 ``_claim_group_decline_notice`` 的键(``reason`` 在键里),所以「群关
    着」和「你没绑定」不会互相把对方按住一整天。
    """
    from app.assistant import channel  # noqa: PLC0415 — cycle

    if group is not None and not await _claim_group_decline_notice(
        chat_id=chat_id, reason=reason,
    ):
        return
    await channel.send_text(
        db, to=target, text=text, idempotency_key=event_id or None,
    )


async def _within_rate_limit(db: AsyncSession, *, person_id: Any) -> bool:
    """Per-person hourly turn budget. Fail-open on Redis fault."""
    limit = await bounded_int_setting(
        db,
        SETTING_TURNS_PER_HOUR,
        default=DEFAULT_TURNS_PER_HOUR,
        minimum=5,
        maximum=600,
    )
    try:
        redis = get_redis()
        count = int(await redis.eval(
            _RATE_SCRIPT, 1, f"{_RATE_PREFIX}{person_id}", _RATE_WINDOW_SECONDS,
        ))
    except Exception as exc:  # noqa: BLE001 — fail-open on Redis fault
        logger.warning("assistant: rate limiter unavailable: %s", exc)
        return True
    return count <= limit


async def _claim_rate_limit_notice(*, person_id: Any) -> bool:
    """True the first time this person hits the ceiling in the current window."""
    try:
        claimed = await get_redis().set(
            f"{_RATE_NOTICE_PREFIX}{person_id}", "1",
            nx=True, ex=_RATE_WINDOW_SECONDS,
        )
        return bool(claimed)
    except Exception as exc:  # noqa: BLE001 — say it rather than stay silent
        logger.warning("assistant: rate-limit notice claim unavailable: %s", exc)
        return True


async def _within_group_rate_limit(db: AsyncSession, *, chat_id: str) -> bool:
    """Per-room hourly turn budget. Fail-open on Redis fault.

    Independent of the per-person budget on purpose: ten people in one room
    asking six questions each is under every personal ceiling and still an
    unbounded bill. The room is the thing that has to be bounded here.
    """
    from app.assistant.group_chat import (  # noqa: PLC0415 — cycle
        DEFAULT_GROUP_TURNS_PER_HOUR,
        SETTING_GROUP_TURNS_PER_HOUR,
    )

    limit = await bounded_int_setting(
        db,
        SETTING_GROUP_TURNS_PER_HOUR,
        default=DEFAULT_GROUP_TURNS_PER_HOUR,
        minimum=5,
        maximum=2000,
    )
    try:
        count = int(await get_redis().eval(
            _RATE_SCRIPT, 1, f"{_GROUP_RATE_PREFIX}{chat_id}", _RATE_WINDOW_SECONDS,
        ))
    except Exception as exc:  # noqa: BLE001 — fail-open on Redis fault
        logger.warning("assistant: group rate limiter unavailable: %s", exc)
        return True
    return count <= limit


async def _claim_group_decline_notice(
    *, chat_id: str, reason: str, ttl_seconds: int = _GROUP_DECLINE_TTL_SECONDS,
) -> bool:
    """True at most once a day per room **per reason**, so a refusal is said
    but not spammed.

    ``reason`` is part of the key on purpose. 「群内应答关掉了」 and 「这个群这
    小时问得太多了」 are different facts with different fixes, and a single
    shared key would let the first one silence the second for 24 hours — the
    room would then get no explanation at all for the thing that is actually
    happening.

    Fail-open (say it) rather than fail-closed (stay silent): silence in a room
    where somebody just @-mentioned the bot reads as a broken integration, and
    the next step after that reading is a ticket about the assistant itself.
    """
    try:
        claimed = await get_redis().set(
            f"{_GROUP_DECLINE_PREFIX}{reason}:{chat_id}", "1",
            nx=True, ex=ttl_seconds,
        )
        return bool(claimed)
    except Exception as exc:  # noqa: BLE001
        logger.warning("assistant: group decline claim unavailable: %s", exc)
        return True


def _group_kind(group: group_chat.GroupContext | None) -> str:
    """这一轮发生在哪种房间。

    返回的是 ``models.CHAT_KIND_*`` 本身,不是长得一样的字面量:同一个值既写进
    审计明细,也写进 ``assistant_conversations.chat_kind``,两处必须永远一致 ——
    而「碰巧拼得一样」不是一致。
    """
    from app.assistant.models import (  # noqa: PLC0415 — cycle
        CHAT_KIND_DM,
        CHAT_KIND_GROUP,
        CHAT_KIND_HANDOFF,
    )

    if group is None:
        return CHAT_KIND_DM
    return CHAT_KIND_HANDOFF if group.is_handoff else CHAT_KIND_GROUP


def reply_target(
    incoming: IncomingMessage, group: group_chat.GroupContext | None,
) -> ReplyTarget:
    """Where this message's answers go. Resolved once, used for every send.

    A handoff group holds two or three people and one subject, so a plain post
    reads best. Any other room may be busy and about anything, so the answer is
    a **threaded reply** under the question — otherwise a bystander sees an
    answer with no visible question and the room fills with orphans.
    """
    if group is None:
        return ReplyTarget.dm(incoming.sender_open_id)
    if group.is_handoff:
        return ReplyTarget.chat(incoming.chat_id)
    return ReplyTarget.chat(
        incoming.chat_id,
        reply_to_message_id=incoming.message_id or None,
        thread=True,
    )


def try_claim_event(payload: dict[str, Any], *, trace_id: str) -> bool:
    """Fork point for the shared Feishu callback.

    Returns ``True`` when the assistant has taken ownership of this event, in
    which case the caller must acknowledge it and do nothing else. Returns
    ``False`` for every event the assistant does not own, leaving the approval
    path exactly as it was.

    Synchronous by design: the caller is on the ack path, so the only work done
    here is handing a snapshot to the background runner.
    """
    if not claims_event(payload):
        return False
    event_type = event_type_of(payload)
    if event_type == EVENT_BOT_MENU:
        background.spawn(
            _run_menu_event(payload, trace_id=trace_id),
            name="assistant.menu_event",
        )
    elif event_type in (EVENT_BOT_ADDED, EVENT_BOT_REMOVED):
        background.spawn(
            _run_membership_event(payload, trace_id=trace_id),
            name="assistant.membership_event",
        )
    else:
        background.spawn(
            _run_message_turn(payload, trace_id=trace_id),
            name="assistant.message_turn",
        )
    return True


async def _run_message_turn(payload: dict[str, Any], *, trace_id: str) -> None:
    """Background entry point — owns its own session and never raises upward."""
    from app.core.database import async_session_factory  # noqa: PLC0415

    async with async_session_factory() as db:
        try:
            await handle_message_event(db, payload, trace_id=trace_id)
        except Exception:
            await db.rollback()
            raise


async def _run_membership_event(payload: dict[str, Any], *, trace_id: str) -> None:
    """Background entry point for bot-added / bot-removed."""
    from app.core.database import async_session_factory  # noqa: PLC0415

    async with async_session_factory() as db:
        try:
            await handle_membership_event(db, payload, trace_id=trace_id)
        except Exception:
            await db.rollback()
            raise


async def _run_menu_event(payload: dict[str, Any], *, trace_id: str) -> None:
    """Background entry point for a deterministic bot-menu action."""
    from app.core.database import async_session_factory  # noqa: PLC0415

    async with async_session_factory() as db:
        try:
            await handle_menu_event(db, payload, trace_id=trace_id)
        except Exception:
            await db.rollback()
            raise


async def handle_menu_event(
    db: AsyncSession, payload: dict[str, Any], *, trace_id: str,
) -> str:
    """Execute one agent-oriented bot menu action.

    There is no user message or chat id in a Feishu menu event.  Keeping this
    path deterministic avoids asking the model to invent a request context and
    guarantees that a menu tap cannot create a ticket or approval by itself.
    """
    from app.assistant import channel, session  # noqa: PLC0415

    if not await assistant_enabled(db):
        return "disabled"
    event = payload.get("event")
    if not isinstance(event, dict):
        return "ignored"
    event_key = str(event.get("event_key") or "")
    if event_key not in _OWNED_MENU_KEYS:
        return "ignored"
    sender_ids = _menu_sender(payload)
    if sender_ids is None:
        return "unbound"
    open_id, union_id = sender_ids
    header = payload.get("header")
    header = header if isinstance(header, dict) else {}
    event_id = str(header.get("event_id") or "")
    if not await claim_event(event_id):
        return "duplicate"

    sender = await resolve_sender(db, union_id=union_id, open_id=open_id)
    # 菜单事件没有文字可判,所以和卡片按钮同一条规矩:这个人最近那段对话在用
    # 什么语言,就用什么语言答;不认识就退回他的偏好。一段英文对话之后点开
    # 菜单里的「使用帮助」,不该拿到一张中文卡。
    lang = await reply_language.for_card(
        person=sender.person,
        chat_id=(
            await session.active_chat(person_id=sender.person.id)
            if sender.person is not None else ""
        ) or "",
    )
    if sender.status is not SenderStatus.OK or sender.person is None:
        text = (
            messages.unbound(lang)
            if sender.status is SenderStatus.UNBOUND
            else messages.departed(lang)
            if sender.status is SenderStatus.DEPARTED
            else messages.sender_mismatch(lang)
        )
        await channel.send_text(
            db, to=ReplyTarget.dm(open_id), text=text,
            idempotency_key=event_id or None,
        )
        return sender.status.value

    person_id = sender.person.id
    if event_key == MENU_RESET:
        chat_id = await session.active_chat(person_id=person_id)
        cleared = await session.clear_history(person_id=person_id, chat_id=chat_id) if chat_id else True
        text = messages.reset_done(lang) if cleared else messages.reset_failed(lang)
        outcome = "reset" if cleared else "reset_failed"
    else:
        # 「使用帮助」和(已下线的)「开始求助」现在是同一张卡:两个菜单项当初
        # 分别回一段长文本,而它们要说的其实是同一件事 —— 你能问我什么,以及
        # 现在该说什么。卡片把第二半从「读完自己想」变成「点一下就问」。
        text = messages.help_guide(lang)
        outcome = "help" if event_key == MENU_HELP else "start"

    # 菜单事件没有房间 —— 它来自机器人自己的菜单栏,回复只可能是私聊。
    target = ReplyTarget.dm(open_id)
    delivered = False
    if outcome in ("help", "start"):
        delivered = await channel.send_card(
            db, to=target, card=help_card.build_home_card(
                lang, chosen=preferences.current_lang(sender.person),
            ),
            idempotency_key=event_id or None,
        )
    if not delivered:
        # 卡片是排版,内容才是目的。飞书拒了这张卡(校验器、限流、权限),员工
        # 仍然要拿到那份说明 —— 纯文本版一直留着就是为了这一刻。
        await channel.send_text(
            db, to=target, text=text,
            idempotency_key=event_id or None,
        )
    await _audit_menu(
        db, trace_id=trace_id, person_id=person_id, event_id=event_id,
        event_key=event_key, outcome=outcome,
    )
    return outcome


async def handle_message_event(
    db: AsyncSession, payload: dict[str, Any], *, trace_id: str,
) -> str:
    """Process one inbound message end to end. Returns the outcome for tests.

    Outcomes: ``disabled`` (gate closed — deliberately silent), ``duplicate``,
    ``ignored`` (group chat / bot / unparseable), ``unbound``, ``departed``,
    ``rate_limited``, ``unsupported``, ``reset``, ``busy``, ``offered``
    (a confirmation card was sent), and the turn's own outcome
    (``answered`` / ``unavailable`` / ``failed``).
    """
    from app.assistant import (  # noqa: PLC0415
        channel,
        conversation,
        escalation,
        session,
    )
    from app.assistant import ticket_chat as ticket_chat_mod  # noqa: PLC0415

    # The gate is checked before anything is parsed or replied to. While the
    # product is unreleased the bot must behave exactly as it did before this
    # code existed: a notification sender that does not talk back.
    if not await assistant_enabled(db):
        return "disabled"

    incoming = parse_message_event(payload)
    if incoming is None:
        return "ignored"
    if not await claim_event(incoming.event_id):
        logger.info("assistant: duplicate event %s dropped", incoming.event_id)
        return "duplicate"

    # ── 群:第一个问题永远是「这句话是在跟我说吗」 ────────────────────
    #
    # 不是「飞书把它投给我了吗」。投递范围取决于应用当前的权限,而权限会变;
    # 「被 @ 了」是消息自己带着的事实。没有 @ 的群消息一律不处理,并且**不留
    # 回复** —— 在一个没叫你的房间里说话,比不说话糟得多。
    group: group_chat.GroupContext | None = None
    bot_id: str | None = None
    if incoming.is_group:
        bot_id = await group_chat.bot_open_id(db)
        if not group_chat.addressed_to_bot(incoming.mentions, bot_id=bot_id):
            return "not_addressed"
        # 正文里的 ``@_user_1`` 对模型没有意义,而且它会照抄。
        incoming = replace(
            incoming,
            text=group_chat.strip_bot_mention(
                incoming.text, incoming.mentions, bot_id=bot_id,
            ),
        )
        group = await group_chat.resolve_group_context(
            db, chat_id=incoming.chat_id,
        )
    target = reply_target(incoming, group)

    sender = await resolve_sender(
        db, union_id=incoming.sender_union_id, open_id=incoming.sender_open_id,
    )
    # 回答一句话的语言,应该是**那句话的语言** —— 判不出来时才退回这个会话上
    # 一次用的语言,再退回这个人的偏好。见 :mod:`app.assistant.reply_language`。
    # 这一步在身份分支**之前**:没绑定、身份对不上的人同样要收到一句话,而那
    # 一句该用他写的那种语言说,否则一个英文同事收到的「请先登录一次 Helix」
    # 是中文的 —— 和没收到差不多。
    lang = await reply_language.for_turn(
        text=incoming.text, person=sender.person, chat_id=incoming.chat_id,
    )

    # ── 群:这一类房间此刻通不通 ──────────────────────────────────────
    #
    # 在身份分支之前:关掉的群里不该念出「你还没绑定 Helix」——那是给一屋子人
    # 的通报,而不是给当事人的指引。
    if group is not None and not await group_chat.group_enabled(
        db, is_handoff=group.is_handoff,
    ):
        if await _claim_group_decline_notice(
            chat_id=incoming.chat_id, reason="disabled",
        ):
            await channel.send_text(
                db, to=target,
                text=(
                    messages.group_disabled(lang) if group.is_handoff
                    else messages.group_any_disabled(lang)
                ),
                idempotency_key=incoming.event_id or None,
            )
        await _audit(
            db, trace_id=trace_id,
            person_id=sender.person.id if sender.person else None,
            outcome="group_disabled", incoming=incoming,
            detail={"handoff_group": group.is_handoff},
        )
        return "group_disabled"

    if sender.status is SenderStatus.UNBOUND:
        await _refuse_to(
            db, target=target, group=group, chat_id=incoming.chat_id,
            reason="unbound", text=messages.unbound(lang),
            event_id=incoming.event_id,
        )
        await _audit(
            db, trace_id=trace_id, person_id=None, outcome="unbound",
            incoming=incoming,
            detail={"open_id": incoming.sender_open_id},
        )
        return "unbound"

    if sender.status is SenderStatus.MISMATCH:
        await _refuse_to(
            db, target=target, group=group, chat_id=incoming.chat_id,
            reason="mismatch", text=messages.sender_mismatch(lang),
            event_id=incoming.event_id,
        )
        await _audit(
            db, trace_id=trace_id, person_id=None, outcome="mismatch",
            incoming=incoming, detail={"open_id": incoming.sender_open_id},
        )
        return "mismatch"

    if sender.status is SenderStatus.DEPARTED:
        await _refuse_to(
            db, target=target, group=group, chat_id=incoming.chat_id,
            reason="departed", text=messages.departed(lang),
            event_id=incoming.event_id,
        )
        await _audit(
            db, trace_id=trace_id,
            person_id=sender.person.id if sender.person else None,
            outcome="departed", incoming=incoming, detail={},
        )
        return "departed"

    person = sender.person
    assert person is not None  # noqa: S101 — guaranteed by SenderStatus.OK
    # Captured once. Everything below runs AFTER paths that may roll this session
    # back — the turn's own failure handler, and the transcript write's seq-clash
    # retry — and a rollback expires every instance, so a later ``person.id`` is
    # lazy IO that raises ``MissingGreenlet`` and takes the audit row with it.
    # The id is the only attribute any of this needs.
    person_id = person.id
    # 只记单聊。这个键是「机器人菜单里的『新话题』该清哪一段记忆」的答案,而群
    # 里的对话不是这个人的私人线程 —— 让一次群发言把它指到群上,员工再点「新
    # 话题」清掉的就是别人也在的那段,而他自己的线程原封不动地留着。
    if not incoming.is_group:
        await session.remember_active_chat(
            person_id=person_id, chat_id=incoming.chat_id,
        )

    # A bare asset tag is often the answer to the assistant's immediately
    # preceding buyback question. Resolve it deterministically against the
    # employee's inventory rather than asking the model to rediscover the
    # conversation and translate a human tag into an internal UUID.
    #
    # 单聊限定:这条捷径的前提是「刚才我私下把这个资产号给过你」。群里一个
    # 裸资产号既可能是别人念出来的,也可能根本不是在跟助手说话,而它的产物是
    # 一张回购确认卡 —— 猜错的代价是一张没人要的卡。
    if not incoming.is_group and await _offer_selected_buyback(
        db,
        person=person,
        incoming=incoming,
        lang=lang,
        trace_id=trace_id,
    ):
        await _audit(
            db, trace_id=trace_id, person_id=person_id,
            outcome="buyback_offer", incoming=incoming,
            detail={"selection": "asset_tag"},
        )
        return "offered"

    if group is not None and not await _within_group_rate_limit(
        db, chat_id=incoming.chat_id,
    ):
        logger.info("assistant: group rate limited chat=%s", incoming.chat_id)
        if await _claim_group_decline_notice(
            chat_id=incoming.chat_id, reason="rate",
            ttl_seconds=_GROUP_RATE_NOTICE_TTL_SECONDS,
        ):
            await channel.send_text(
                db, to=target, text=messages.rate_limited(lang),
                idempotency_key=incoming.event_id or None,
            )
        await _audit(
            db, trace_id=trace_id, person_id=person_id,
            outcome="group_rate_limited", incoming=incoming, detail={},
        )
        return "group_rate_limited"

    if not await _within_rate_limit(db, person_id=person_id):
        logger.info("assistant: rate limited person=%s", person_id)
        # Told once per window, not once per message: silence reads as "the bot
        # is broken" (and sends support down the wrong diagnosis), but repeating
        # it on every message would be its own kind of spam.
        if await _claim_rate_limit_notice(person_id=person_id):
            await channel.send_text(
                db, to=target,
                text=messages.rate_limited(lang),
                idempotency_key=incoming.event_id or None,
            )
        await _audit(
            db, trace_id=trace_id, person_id=person_id, outcome="rate_limited",
            incoming=incoming, detail={},
        )
        return "rate_limited"

    if not incoming.is_supported:
        await channel.send_text(
            db, to=target,
            text=messages.unsupported_message(lang),
            idempotency_key=incoming.event_id or None,
        )
        await _audit(
            db, trace_id=trace_id, person_id=person_id, outcome="unsupported",
            incoming=incoming, detail={"message_type": incoming.message_type},
        )
        return "unsupported"

    # ── 工单群里的命令 ────────────────────────────────────────────────
    #
    # 在通用的 /help、新话题之前:这个房间里「帮助」该说的是**这张工单能怎么推
    # 进**,而不是助手的通用用法。命令走字面匹配,认不出来的句子照常交给模型。
    if group is not None and group.is_handoff:
        from app.assistant import ticket_commands  # noqa: PLC0415 — cycle

        handled = await ticket_commands.try_handle(
            db,
            text=incoming.text,
            actor=person,
            context=group,
            mentions=incoming.mentions,
            bot_id=bot_id,
            lang=lang,
            target=target,
            trace_id=trace_id,
        )
        if handled is not None:
            ticket_chat_mod.touch(group.ticket_chat)
            await db.commit()
            await _audit(
                db, trace_id=trace_id, person_id=person_id,
                outcome=handled.outcome, incoming=incoming,
                detail={"ticket_id": str(group.ticket_id)},
            )
            return handled.outcome

    if session.is_help_command(incoming.text):
        # 单聊里回卡片(可点、可分类、按一下就问);群里回文本 —— 那张卡上的
        # 每一颗按钮都只对按的人有意义,而群里任何人都能按,答案却会私下发给
        # 按的人,在屋里看就是「机器人没反应」。
        delivered = False
        if not incoming.is_group:
            delivered = await channel.send_card(
                db, to=target,
                card=help_card.build_home_card(
                    lang, chosen=preferences.current_lang(person),
                ),
                idempotency_key=incoming.event_id or None,
            )
        if not delivered:
            await channel.send_text(
                db, to=target, text=messages.help_guide(lang),
                idempotency_key=incoming.event_id or None,
            )
        await _audit(
            db, trace_id=trace_id, person_id=person_id, outcome="help",
            incoming=incoming, detail={},
        )
        return "help"

    if session.is_reset_command(incoming.text):
        # The reply follows the RESULT. Announcing 「好的，我们从头开始」 while the
        # clear had silently failed left the employee believing the model had
        # forgotten a thread it still had in full — the one control they have
        # over its memory, taken away without a trace.
        cleared = await session.clear_history(
            person_id=person_id, chat_id=incoming.chat_id,
        )
        await channel.send_text(
            db, to=target,
            text=messages.reset_done(lang) if cleared else messages.reset_failed(lang),
            idempotency_key=incoming.event_id or None,
        )
        await _audit(
            db, trace_id=trace_id, person_id=person_id,
            outcome="reset" if cleared else "reset_failed",
            incoming=incoming, detail={"cleared": cleared},
        )
        return "reset" if cleared else "reset_failed"

    if escalation.is_escalation_request(incoming.text):
        # 在工单群里说「转人工」—— 他们已经到了。再拟一张工单会凭空多出第二条
        # 台账,而这个房间本身就是第一条的产物。
        if group is not None and group.is_handoff:
            await channel.send_text(
                db, to=target, text=messages.group_already_human(lang),
                idempotency_key=incoming.event_id or None,
            )
            await _audit(
                db, trace_id=trace_id, person_id=person_id,
                outcome="already_handed_off", incoming=incoming, detail={},
            )
            return "already_handed_off"
        outcome = await escalation.offer_ticket(
            db,
            person=person,
            chat_id=incoming.chat_id,
            open_id=incoming.sender_open_id,
            lang=lang,
            trace_id=trace_id,
        )
        # 卡片是私聊发的(它绑本人、只有本人能按),所以群里必须留一句话说明
        # 事情去哪了 —— 否则在别人看来 @ 了机器人而它什么都没做。
        if group is not None and outcome == "offered":
            await channel.send_text(
                db, to=target, text=messages.group_detail_sent_privately(lang),
            )
        await _audit(
            db, trace_id=trace_id, person_id=person_id, outcome=outcome,
            incoming=incoming, detail={},
        )
        return outcome

    # ── 「我要找人工」之后补上的那句说明 ──────────────────────────────
    #
    # 上一步(空对话里的转人工)记了一笔,并且答应过「你的下一条消息我直接整理
    # 成工单」。所以这一条**不进模型**:员工已经说了他要人,再让模型接一次话
    # 是把他推回原点,而那句承诺也就成了假话。
    #
    # 位置在转人工判断之后:armed 期间又说一次「转人工」的人是在重复请求,
    # 不是在描述问题 —— 那一支会把他再问一遍,而不是开一张标题叫「转人工」的
    # 工单。群里不参与:那面没有这颗按钮,标记也永远不会被点亮。
    if group is None and await session.take_handoff_armed(
        person_id=person_id, chat_id=incoming.chat_id,
    ):
        outcome = await escalation.offer_ticket(
            db,
            person=person,
            chat_id=incoming.chat_id,
            open_id=incoming.sender_open_id,
            lang=lang,
            trace_id=trace_id,
            extra_user_message=incoming.text,
        )
        await _audit(
            db, trace_id=trace_id, person_id=person_id, outcome=outcome,
            incoming=incoming, detail={"source": "handoff_detail"},
        )
        return outcome

    # One turn at a time per chat. Two questions a second apart would otherwise
    # read the same history and answer out of order.
    lock_owner = await session.acquire_turn_lock(
        person_id=person_id, chat_id=incoming.chat_id,
    )
    if lock_owner is None:
        await channel.send_text(
            db, to=target, text=messages.busy(lang),
            idempotency_key=incoming.event_id or None,
        )
        await _audit(
            db, trace_id=trace_id, person_id=person_id, outcome="busy",
            incoming=incoming, detail={},
        )
        return "busy"

    async def _holding_note() -> None:
        # Its own session on purpose: this runs concurrently with the turn, and
        # sharing the turn's AsyncSession would corrupt the connection both are
        # driving — and could strand the turn's own transcript commit.
        from app.core.database import async_session_factory  # noqa: PLC0415

        async with async_session_factory() as note_db:
            await channel.send_text(
                note_db, to=target, text=messages.working(lang),
            )

    # Which card is this a reply to? The platform pushes ~3,800 cards a
    # month through this same app into this same chat, so a reply under
    # one has always reached us — it just arrived with no idea what it
    # was about. Resolved BEFORE the turn because it belongs in the
    # system prompt, not in the answer.
    #
    # Best-effort by construction: a miss is the ordinary case (a fresh
    # question, a reply to a plain message, a card sent before migration
    # 569 recorded ids), and a failure here must never cost the employee
    # their answer.
    card_context = None
    if incoming.parent_message_id or incoming.root_message_id:
        try:
            from app.assistant.card_context import resolve_card_context

            card_context = await resolve_card_context(
                db,
                person_id=person_id,
                parent_message_id=incoming.parent_message_id,
                root_message_id=incoming.root_message_id,
            )
        except Exception:  # noqa: BLE001 — context is a bonus, not a gate
            logger.warning(
                "assistant: card-context lookup failed person=%s", person_id,
                exc_info=True,
            )

    try:
        turn = await conversation.run_turn(
            db,
            person=person,
            chat_id=incoming.chat_id,
            question=incoming.text,
            lang=lang,
            trace_id=trace_id,
            on_slow=_holding_note,
            open_id=incoming.sender_open_id,
            card_context=card_context,
            # 群里的行为守则。追加一段系统上下文,而不是换一套提示词 ——
            # 助手在群里和在单聊里必须是同一个助手,只是多知道一件事:
            # 这句话有旁人在听。
            extra_system=(
                group_chat.group_prompt_block(group) if group is not None else ""
            ),
            chat_kind=_group_kind(group),
        )
    finally:
        await session.release_turn_lock(
            person_id=person_id, chat_id=incoming.chat_id, owner=lock_owner,
        )

    delivered = await deliver_turn(
        db, target=target, turn=turn, lang=lang,
        idempotency_key=incoming.event_id or None,
    )
    if group is not None and group.is_handoff:
        # 这个群还活着。保留策略(工单关了 N 天后解散)和「这个群是不是已经
        # 没人管了」都读这一列,所以每一次有人跟助手说话都要落一次。
        ticket_chat_mod.touch(group.ticket_chat)
        await db.commit()
    await _audit(
        db, trace_id=trace_id, person_id=person_id, outcome=turn.outcome,
        incoming=incoming,
        detail={
            "delivered": delivered,
            "tool_calls": turn.tool_calls,
            "provider": turn.provider,
            "latency_ms": turn.latency_ms,
            # The two card-reply failure modes look identical from
            # outside and have opposite fixes: nobody replies to cards
            # (the affordance is not discoverable) vs replies arrive but
            # resolve to nothing (the id was never recorded, or the card
            # belongs to somebody else). Recording BOTH the attempt and
            # the outcome is what separates them.
            "card_context_resolved": card_context is not None,
            "card_context_resource": (
                card_context.resource_type if card_context else None
            ),
            # 群里的一轮和单聊的一轮成本、风险都不同,而 ``chat_type`` 只说
            # 「是不是群」,不说「是哪种群」。运营台按它分开看两条曲线。
            "group_kind": _group_kind(group),
        },
    )
    return turn.outcome


async def deliver_turn(
    db: AsyncSession,
    *,
    target: ReplyTarget,
    turn: Any,
    lang: Lang,
    idempotency_key: str | None,
) -> bool:
    """把一轮的产物送到聊天里,按内容自己挑形状。返回是否送达。

    三条路,顺序是有讲究的:

    * **卡片已经是回复本身**(``card_only``)—— 确认卡已经发出去了,``reply``
      只是留给记忆和台账的替身。再发一遍等于在员工正在读的卡下面贴一段旁白。
    * **有依据的实质回答** → 答案卡,它能带上参考文档和三个回应方式。
    * **其余** → 富文本(保住加粗/链接/列表),不值得富文本的短句走纯文本。

    每一层都往下退:卡片是排版,答案才是目的,任何一次投递失败都不该让员工
    什么也收不到。

    共用的原因不是省行数,而是**两条入口必须给出同一种回答**:员工打字问的
    和在帮助卡上点出来的,是同一个问题,不该一个有参考文档、另一个是一坨纯文本。
    """
    from app.assistant import answer_card, channel  # noqa: PLC0415 — cycle

    if turn.card_only:
        return True
    if turn.reply and turn.message_id and answer_card.should_render_as_card(
        turn.reply,
        grounded=bool(turn.tool_calls),
        has_sources=bool(turn.docs),
    ) and "[[helix_mention:" not in turn.reply:
        delivered = await channel.send_card(
            db,
            to=target,
            card=answer_card.build_answer_card(
                # The card renders Markdown, so it gets the unflattened text.
                reply=turn.reply_rich or turn.reply,
                message_id=turn.message_id,
                lang=lang,
                docs=turn.docs,
            ),
            idempotency_key=idempotency_key,
        )
        if delivered:
            return True
        # The card is the nicety; the answer is the point. Never let a
        # card-delivery failure swallow a reply the employee is waiting on.
        return await channel.send_text(
            db, to=target, text=turn.reply, idempotency_key=idempotency_key,
        )
    # 值得保留格式的回复走**富文本**。``turn.reply`` 是被 ``to_feishu_text``
    # 剥光的那份(纯文本消息渲染不了标记,留着更难看),而 ``reply_rich`` 是
    # 模型原文 —— 加粗、链接、列表都还在。助手回答的常常是「你有 2 台电脑,
    # 分别是…」这种天生有结构的内容,把它压成一坨无层次的字是这条链路上最便宜
    # 也最明显的损失。
    #
    # 短句、无标记的回复仍然走纯文本:给「好的,已记下」套一层富文本没有任何
    # 收益。富文本发失败也退回纯文本 —— 格式是加分项,答案才是目的。
    rich = turn.reply_rich or turn.reply
    if channel.wants_rich_text(rich) and await channel.send_post(
        db, to=target, markdown=rich, idempotency_key=idempotency_key,
    ):
        return True
    return await channel.send_text(
        db, to=target, text=turn.reply, idempotency_key=idempotency_key,
    )


async def ask_starter(
    *,
    person_id: Any,
    open_id: str,
    chat_id: str,
    question: str,
    lang: Lang,
    trace_id: str,
    starter_key: str,
    event_id: str = "",
) -> str:
    """员工在帮助卡上点了一句开场白 —— 当成他自己打的那句话跑一整轮。

    **和打字进来的一轮走同一条路**:同一个限速窗口、同一把每聊天的回合锁、
    同一个 :func:`deliver_turn`、同一张审计行。点一颗按钮和敲一行字对平台
    的成本完全一样,所以它们必须受同样的约束 —— 否则这张卡就成了绕过限速的
    后门,而「同一个问题两种答案」也会立刻出现。

    **自带 session**:调用方是飞书卡片回调,它必须在 3 秒内把 HTTP 响应还给
    飞书,而一轮要几秒到几十秒。所以这个协程是 ``background.spawn`` 出来的,
    活得比那次请求久 —— 用请求的 session 会在 FastAPI 拆掉它之后炸在半路。

    人也在这里重新读一遍,而不是从调用方传对象过来:跨 session 复用 ORM 实例
    会拿到另一个 identity map 里的旧副本,而这一轮要把它交给模型当身份用。

    ``event_id`` 挡住飞书自己的重投。卡片回调**故意**不走 webhook 那道重放闸
    (那批分支都是幂等的),而「跑一整轮」不是 —— 没有这一claim,一次被判慢的
    重投会让同一个问题问两遍;回合锁会把第二遍变成一句「我还在处理你上一条」,
    于是员工为一次自己没做过的点击收到一句莫名其妙的话。
    """
    from app.assistant import channel, conversation, session  # noqa: PLC0415
    from app.core.database import async_session_factory  # noqa: PLC0415
    from app.identity.person_models import IdentityPerson  # noqa: PLC0415

    if event_id and not await claim_event(event_id):
        logger.info("assistant: duplicate starter tap %s", event_id)
        return "duplicate"

    target = ReplyTarget.dm(open_id)
    async with async_session_factory() as db:
        if not await assistant_enabled(db):
            return "disabled"
        person = await db.get(IdentityPerson, person_id)
        if person is None:
            logger.warning("assistant: starter tap for unknown person %s", person_id)
            return "unbound"

        # 只用卡片、从不打字的人也该有一个「新话题」能清掉的线程。这个指针是
        # 菜单里那颗按钮唯一的线索(菜单事件不带会话 id)。
        #
        # **只在还没有指针时写**:调用方已经优先用了这个指针作为线程键,所以
        # 正常情况下这里写的就是它自己;而一张被转发进群、在群里被点的卡不该
        # 把一个已经指对了的私聊指针改指到群上 —— 消息那一侧同样只记单聊
        # (见 ``handle_message_event`` 的「只记单聊」)。
        if not await session.active_chat(person_id=person_id):
            await session.remember_active_chat(
                person_id=person_id, chat_id=chat_id,
            )

        # 点出去的这一句**就是**这个人这一轮说的话,所以它和打字进来的一句走
        # 同一个语言判定 —— 顺带把会话的语言记忆更新到这句话的语言上,不然他
        # 下一句「ok」会退回上一段对话的语言。
        lang = await reply_language.for_turn(
            text=question, person=person, chat_id=chat_id,
        )

        if not await _within_rate_limit(db, person_id=person_id):
            logger.info("assistant: starter rate limited person=%s", person_id)
            if await _claim_rate_limit_notice(person_id=person_id):
                await channel.send_text(
                    db, to=target, text=messages.rate_limited(lang),
                )
            await _audit_starter(
                db, trace_id=trace_id, person_id=person_id, chat_id=chat_id,
                starter_key=starter_key, outcome="rate_limited", detail={},
            )
            return "rate_limited"

        lock_owner = await session.acquire_turn_lock(
            person_id=person_id, chat_id=chat_id,
        )
        if lock_owner is None:
            await channel.send_text(db, to=target, text=messages.busy(lang))
            await _audit_starter(
                db, trace_id=trace_id, person_id=person_id, chat_id=chat_id,
                starter_key=starter_key, outcome="busy", detail={},
            )
            return "busy"

        async def _holding_note() -> None:
            # 自己的 session:它和这一轮并发跑,共用连接会把两边都毁掉。
            async with async_session_factory() as note_db:
                await channel.send_text(
                    note_db, to=target, text=messages.working(lang),
                )

        try:
            turn = await conversation.run_turn(
                db,
                person=person,
                chat_id=chat_id,
                question=question,
                lang=lang,
                trace_id=trace_id,
                on_slow=_holding_note,
                open_id=open_id,
            )
        finally:
            await session.release_turn_lock(
                person_id=person_id, chat_id=chat_id, owner=lock_owner,
            )

        delivered = await deliver_turn(
            db, target=target, turn=turn, lang=lang, idempotency_key=None,
        )
        await _audit_starter(
            db, trace_id=trace_id, person_id=person_id, chat_id=chat_id,
            starter_key=starter_key, outcome=turn.outcome,
            detail={
                "delivered": delivered,
                "tool_calls": turn.tool_calls,
                "provider": turn.provider,
                "latency_ms": turn.latency_ms,
            },
        )
        return turn.outcome


async def _audit_starter(
    db: AsyncSession,
    *,
    trace_id: str,
    person_id: Any,
    chat_id: str,
    starter_key: str,
    outcome: str,
    detail: dict[str, Any],
) -> None:
    """一次开场白点击的审计行。

    动作名和打字进来的一轮**故意相同**:它就是一条被回答了的消息,管理台那
    几条曲线不该因为员工用了按钮而漏计。``starter`` 才是区分二者的字段 ——
    顺带也回答了「这八类里哪几类真的有人点」,而那正是下一轮该改哪张卡的依据。
    """
    await AuditService(db).log(
        trace_id=trace_id,
        actor_type="user",
        actor_user_id=person_id,
        action="assistant.message.received",
        resource_type="assistant_conversation",
        resource_id=chat_id or None,
        status="success",
        detail={
            "outcome": outcome,
            "channel": "feishu",
            "chat_type": _CHAT_TYPE_P2P,
            "source": "help_card",
            "starter": starter_key,
            **detail,
        },
    )
    await db.commit()


async def handle_membership_event(
    db: AsyncSession, payload: dict[str, Any], *, trace_id: str,
) -> str:
    """机器人被拉进 / 被移出一个群。

    **被拉进去** → 说一句自我介绍。一个进了群、不说话、也不解释自己能干什么的
    机器人,对屋里的人是纯粹的噪音;而「@我 就能问」这件事没人会去猜。介绍只在
    这一类群该应答时才发 —— 开关关着还自我介绍,等于承诺一件做不到的事。

    **被移出** → 只记一行。台账里那条 ``feishu_chat_id`` 从此指向一个不存在的
    房间,后面每一次往它发消息都会失败,而失败原因读起来像飞书故障。真正的
    状态列在群生命周期那一版落地;这一版先把事实记进审计,让「群什么时候没的」
    可查。
    """
    from app.assistant import channel  # noqa: PLC0415 — cycle

    if not await assistant_enabled(db):
        return "disabled"
    event = payload.get("event")
    if not isinstance(event, dict):
        return "ignored"
    chat_id = str(event.get("chat_id") or "")
    if not chat_id:
        return "ignored"
    header = payload.get("header")
    header = header if isinstance(header, dict) else {}
    event_id = str(header.get("event_id") or "")
    if not await claim_event(event_id):
        return "duplicate"

    event_type = event_type_of(payload)
    group = await group_chat.resolve_group_context(db, chat_id=chat_id)
    if event_type == EVENT_BOT_REMOVED:
        # 机器人被移出去了 ⇒ 这个房间对 Helix 来说已经不存在:往里发的每一条
        # 消息都会失败,而失败原因读起来像飞书故障。所以**落状态**,而不是
        # 留给下一次调用去发现。
        if group.ticket_chat is not None:
            from app.assistant import ticket_chat as ticket_chat_mod  # noqa: PLC0415

            ticket_chat_mod.mark_closed(group.ticket_chat, reason="bot_removed")
            await db.commit()
        await _audit_chat(
            db, trace_id=trace_id, chat_id=chat_id, event_id=event_id,
            outcome="bot_removed", detail={"handoff_group": group.is_handoff},
        )
        return "bot_removed"

    if not await group_chat.group_enabled(db, is_handoff=group.is_handoff):
        await _audit_chat(
            db, trace_id=trace_id, chat_id=chat_id, event_id=event_id,
            outcome="bot_added_muted",
            detail={"handoff_group": group.is_handoff},
        )
        return "bot_added_muted"

    # 群的语言无从得知(房间没有语言,人才有),而这句话是对整屋人说的 ——
    # 用平台默认语言,和其它面向群的确定性文案同一条规矩。
    delivered = await channel.send_text(
        db,
        to=ReplyTarget.chat(chat_id),
        text=messages.group_welcome(DEFAULT_LANG),
        idempotency_key=f"assistant-welcome-{event_id}" if event_id else None,
    )
    await _audit_chat(
        db, trace_id=trace_id, chat_id=chat_id, event_id=event_id,
        outcome="bot_added",
        detail={"handoff_group": group.is_handoff, "delivered": delivered},
    )
    return "bot_added"


async def _audit_chat(
    db: AsyncSession,
    *,
    trace_id: str,
    chat_id: str,
    event_id: str,
    outcome: str,
    detail: dict[str, Any],
) -> None:
    """群成员事件的审计行。没有消息正文,也没有当事人 —— 主体是那个房间。"""
    await AuditService(db).log(
        trace_id=trace_id,
        actor_type="system",
        actor_user_id=None,
        action="assistant.chat.membership",
        resource_type="assistant_group_chat",
        resource_id=chat_id or None,
        status="success",
        detail={"outcome": outcome, "event_id": event_id, **detail},
    )
    await db.commit()


async def _audit(
    db: AsyncSession,
    *,
    trace_id: str,
    person_id: Any,
    outcome: str,
    incoming: IncomingMessage,
    detail: dict[str, Any],
) -> None:
    """One audit row per handled message, committed by this background session.

    The message text is deliberately NOT stored here — the conversation record
    with its own retention and admin surface is the next slice's job, and an
    audit row is the wrong place to accumulate employee free text.
    """
    audit = AuditService(db)
    await audit.log(
        trace_id=trace_id,
        actor_type="user" if person_id else "system",
        actor_user_id=person_id,
        action="assistant.message.received",
        resource_type="assistant_conversation",
        resource_id=incoming.chat_id or incoming.message_id or None,
        status="success",
        detail={
            "outcome": outcome,
            "event_id": incoming.event_id,
            "message_id": incoming.message_id,
            "chat_type": incoming.chat_type,
            "channel": "feishu",
            # Whether this turn was a reply to a card, and whether the
            # card was resolvable. Recorded because "did card context
            # fire?" must be answerable from data: the two failure modes
            # (nobody replies to cards / replies arrive but resolve to
            # nothing) look identical from the outside and have opposite
            # fixes. ``detail`` may override — the turn path sets
            # ``card_context`` once it knows.
            "is_card_reply": bool(incoming.parent_message_id),
            **detail,
        },
    )
    await db.commit()


async def _audit_menu(
    db: AsyncSession,
    *,
    trace_id: str,
    person_id: Any,
    event_id: str,
    event_key: str,
    outcome: str,
) -> None:
    """Audit a menu action without collecting employee message text."""
    audit = AuditService(db)
    await audit.log(
        trace_id=trace_id,
        actor_type="user",
        actor_user_id=person_id,
        action="assistant.menu.selected",
        resource_type="assistant_conversation",
        resource_id=None,
        status="success",
        detail={
            "outcome": outcome,
            "event_id": event_id,
            "event_key": event_key,
            "channel": "feishu",
        },
    )
    await db.commit()