"""Confirming inside Feishu, with one tap and no page.

The first shape of the write path put a URL button on the card: tapping opened
a Helix page that rendered the parameter snapshot and its digest. That was the
right call when the app had no card callback registered — but it costs the
employee a browser trip, reads badly on a phone, and gives no way to show the
result where they are looking.

With ``card.action.trigger`` subscribed, Feishu posts the tap to the same
verified callback the chat events already use and *renders whatever card this
returns*. So the confirmation completes in place: press, execute, and the card
itself becomes the receipt.

**Authorisation.** The URL path carries a signed one-time token because a link
can be forwarded to anyone. A card tap instead names the pressing account in the
callback envelope, and the proposal executes only when that account resolves to
the person the proposal belongs to — the button's payload carries an id, never a
credential, so a captured payload grants nothing on its own. What authenticates
the envelope is the shared verification token (Feishu sends no
``X-Lark-Signature`` on a card callback — see ``approval/router.py``), which is
the same trust level every approval callback already runs on; an earlier version
of this docstring claimed a per-request signature and was wrong.

**Three seconds.** Feishu requires a response within 3s and shows the employee
an error if it does not arrive, while confirming a proposal runs a real domain
flow that may call an external API. So the tap is raced against a response
budget: whatever finishes in time is answered in place, and anything slower
returns a "submitting" card and delivers its outcome as a follow-up card. The
employee always gets a definitive answer either way.
"""
from __future__ import annotations

import asyncio
import logging
import uuid
from collections.abc import Awaitable, Callable
from typing import Any, cast

from sqlalchemy.ext.asyncio import AsyncSession

from app.assistant import (
    answer_card,
    help_card,
    reply_language,
    session,
    starters,
    ticket_card,
)
from app.assistant import messages as canned
from app.assistant.identity import SenderStatus, resolve_sender
from app.assistant.models import STATUS_EXECUTED
from app.assistant.proposals import (
    CODE_ALREADY_HANDLED,
    ProposalError,
    assert_actionable,
    confirm_and_execute_for_person,
)
from app.assistant.reply_hygiene import to_feishu_text
from app.assistant.settings import assistant_enabled
from app.core import background
from app.core.exceptions import AppException
from app.core.locale import DEFAULT_LANG, Lang
from app.identity import preferences
from app.identity.person_models import IdentityPerson
from app.notifications import card_design

logger = logging.getLogger(__name__)

#: Feishu's own deadline is 3s; the work is cut off early enough to leave room
#: for serialising the response and the network hop back.
_RESPONSE_BUDGET_SECONDS = 2.2

#: What the button posts back. Carries an identifier, never a credential: the
#: caller's identity comes from the signed envelope, so a leaked value is inert.
ACTION_CONFIRM = "assistant.confirm_proposal"

CARD_ACTION_EVENT = "card.action.trigger"

#: Internal key on a branch's response: the compact card to send as its OWN
#: message when the tap outran the budget. Stripped before the response reaches
#: Feishu — it is routing, not payload.
FOLLOWUP_KEY = "_followup_card"

#: ``SupportService.create`` raises this when the per-(requester, title) dedup
#: lock is held — i.e. the same escalation is being filed right now. Matched on
#: the code rather than the class because it arrives as a plain
#: :class:`AppException` like every other refusal from that service.
_TICKET_IN_FLIGHT_CODE = "SUPPORT_TICKET_IN_FLIGHT"

_TOAST_DONE: dict[Lang, str] = {
    "zh": "已提交", "en": "Submitted", "ja": "送信しました",
}
_TOAST_FAILED: dict[Lang, str] = {
    "zh": "没能提交", "en": "Could not submit", "ja": "送信できませんでした",
}
_DONE_TITLE: dict[Lang, str] = {
    "zh": "✅ 已提交", "en": "✅ Submitted", "ja": "✅ 送信しました",
}
_DONE_BODY: dict[Lang, str] = {
    "zh": "已经以你的名义提交，走的是和网页申请完全一样的流程。",
    "en": "Submitted in your name, through the same flow the web form uses.",
    "ja": "あなたの名義で送信しました（Web と同じフロー）。",
}
_TOAST_RATED: dict[Lang, str] = {
    "zh": "已记下", "en": "Noted", "ja": "記録しました",
}
_TOAST_ESCALATED: dict[Lang, str] = {
    "zh": "已转人工", "en": "Sent to IT", "ja": "IT に転送しました",
}
_ESCALATED_TITLE: dict[Lang, str] = {
    "zh": "已转人工处理", "en": "Handed to IT", "ja": "IT に引き継ぎました",
}
_ESCALATED_BODY: dict[Lang, str] = {
    "zh": "已把你的问题和刚才的回答一起发给 IT，不用再复述一遍。工单号：",
    "en": "Your question and the answer that missed both went to IT — no need to retype. Ticket:",
    "ja": "ご質問と先ほどの回答を IT に送りました。再入力は不要です。チケット：",
}
_ESCALATED_Q: dict[Lang, str] = {
    "zh": "【员工的问题】", "en": "[Question]", "ja": "【質問】",
}
_ESCALATED_A: dict[Lang, str] = {
    "zh": "【助手的回答（未解决）】", "en": "[Assistant answer (did not help)]",
    "ja": "【アシスタントの回答（未解決）】",
}

_TOAST_WORKING: dict[Lang, str] = {
    "zh": "正在处理…", "en": "Working…", "ja": "処理中…",
}
_WORKING_TITLE: dict[Lang, str] = {
    "zh": "⏳ 正在处理", "en": "⏳ Working", "ja": "⏳ 処理中",
}
#: Deliberately says "your tap" rather than "your confirmation": this card is
#: reached by EVERY slow button on this channel — 转人工 and the rating buttons
#: as much as 确认提交 — and 「已收到你的确认」 over a 转人工 tap describes
#: something the employee never did.
_WORKING_BODY: dict[Lang, str] = {
    "zh": "已收到你的点击，正在处理。结果会紧接着发给你，不用重复点击。",
    "en": "Your tap is in. Working on it — the result follows in a moment, no need to tap again.",
    "ja": "タップを受け付けました。処理中です。結果はこの後お知らせしますので、再度タップは不要です。",
}
#: What the answer card's own note says while a slow tap is still running. It
#: has to stay true FOREVER: on the slow path the outcome arrives as its own
#: message and nothing ever comes back to rewrite this line.
_NOTE_ESCALATING: dict[Lang, str] = {
    "zh": "已转人工，正在建单，工单号稍后单独发给你。",
    "en": "Handed to IT. Filing the ticket — the number follows in its own message.",
    "ja": "IT に引き継ぎました。チケットを作成中で、番号は別メッセージでお知らせします。",
}
_NOTE_ESCALATED: dict[Lang, str] = {
    "zh": "已转人工 · 工单号 {number}",
    "en": "Handed to IT · ticket {number}",
    "ja": "IT に引き継ぎました · チケット {number}",
}
#: Feishu re-posts a callback it judged slow, and the ticket dedup lock refuses
#: the second one while the first is still filing. That is a submission in
#: progress, not a failure, and saying 「没能提交」 over it is a lie the employee
#: acts on — the same defect ``_already_done_card`` fixes on the confirm path.
_NOTE_ESCALATE_IN_FLIGHT: dict[Lang, str] = {
    "zh": "已转人工，正在建单，请勿重复点击。",
    "en": "Already going to IT — the ticket is being filed, no need to tap again.",
    "ja": "すでに IT へ転送済みです。チケットを作成中のため、再度のタップは不要です。",
}
#: 转人工成功且群建起来了 —— 说清「谁」被拉进来了,员工才知道去哪等回音。
_NOTE_HANDOFF_CHAT: dict[Lang, str] = {
    "zh": "已拉 {owner} 和你建了一个群,聊天已转过去 · 工单号 {number}",
    "en": "Opened a group with {owner}; the exchange is forwarded · ticket {number}",
    "ja": "{owner} さんとのグループを作成し、やり取りを転送しました · チケット {number}",
}
#: 转人工但没拉到人(这一类还没指派负责人 / 建群失败 / 功能关掉)。
#: 必须说实话:拉不到人时说拉到了,是这条链路最坏的失败。
_NOTE_HANDOFF_NO_CHAT: dict[Lang, str] = {
    "zh": "已转给 IT,会有人接手 · 工单号 {number}",
    "en": "Handed to IT — someone will pick it up · ticket {number}",
    "ja": "IT に引き継ぎました。担当者が対応します · チケット {number}",
}
_HANDOFF_BODY_CHAT: dict[Lang, str] = {
    "zh": "已按「{label}」把你和 {owner} 拉进一个群,刚才的问答已经转过去了,不用再复述一遍。工单号:",
    "en": "Routed as “{label}”: you and {owner} are in a group now and the exchange was forwarded — no need to retype it. Ticket:",
    "ja": "「{label}」として {owner} さんとのグループにお繋ぎし、やり取りを転送しました。再入力は不要です。チケット:",
}
_NOTE_ESCALATE_FAILED: dict[Lang, str] = {
    "zh": "没能转人工，请稍后再试，或直接联系 IT。",
    "en": "Could not hand this to IT. Try again shortly, or contact IT directly.",
    "ja": "IT へ転送できませんでした。しばらくしてから再試行するか、IT にご連絡ください。",
}
#: The confirm receipt repeats what was submitted. Replacing the confirmation
#: card with a bare 「✅ 已提交」 deleted 将要提交的内容 — the parameter rows the
#: employee had just approved — from the only place they were ever shown.
_SUBMITTED_SECTION: dict[Lang, str] = {
    "zh": "已提交的内容", "en": "What was submitted", "ja": "送信した内容",
}
_WHERE: dict[Lang, str] = {
    "zh": "进度可在 https://helix.pingcap.net/my/requests 查看。",
    "en": "Track it at https://helix.pingcap.net/my/requests",
    "ja": "進捗は https://helix.pingcap.net/my/requests で確認できます。",
}


def _with_ticket(body: str, number: str) -> str:
    """Append the ticket number to a follow-up body that ends with 「工单号:」.

    Every ``_..._BODY`` above ends on that dangling label and none of them
    carries a ``{number}`` slot — the number is appended here, on its own line
    and in a code span, because it is the one string the employee will scroll
    back to and select. Routed through one helper on purpose: the 建群 branch
    used to build its body with ``.format(number=number)``, and ``str.format``
    discards a keyword the template has no slot for **silently**. The card went
    out reading 「…不用再复述一遍。工单号:」 with nothing after the colon, on the
    exact path where the in-place card had already promised 「工单号稍后单独发给
    你」 — so on a slow tap (the normal one: ticket creation fans out to every IT
    admin) the number was never delivered anywhere at all.
    """
    return f"{body}\n`{number}`"


#: Every button the assistant puts on a card, routed below.
#: 工单卡上的浮层提示。飞书的 toast 是**一闪而过**的,所以它只说「刚才那下
#: 成功了没有」;真正要留下的东西(状态、处理人)在重画的卡片上。
_TOAST_REFRESHED: dict[Lang, str] = {
    "zh": "已经是最新的了", "en": "Up to date", "ja": "最新の状態です",
}
_TOAST_CLAIMED: dict[Lang, str] = {
    "zh": "你接手了这张工单", "en": "It's yours now", "ja": "担当になりました",
}
_TOAST_ALREADY_YOURS: dict[Lang, str] = {
    "zh": "这张工单本来就在你名下",
    "en": "Already yours",
    "ja": "すでにご担当です",
}
_TOAST_PRIORITY: dict[Lang, str] = {
    "zh": "已标为高优先级", "en": "Priority raised", "ja": "優先度を上げました",
}
_TOAST_NOT_AGENT: dict[Lang, str] = {
    "zh": "只有 IT 管理员或处理人({who})能改这张工单",
    "en": "Only an IT admin or the assignee ({who}) can change this ticket",
    "ja": "IT 管理者または担当者（{who}）のみ変更できます",
}
_TOAST_NOBODY: dict[Lang, str] = {
    "zh": "还没有人", "en": "nobody yet", "ja": "未割り当て",
}
_TOAST_NOT_REQUESTER: dict[Lang, str] = {
    "zh": "这两个只有求助人本人能按",
    "en": "Only the requester can use these",
    "ja": "これは依頼者ご本人のみ操作できます",
}
_TOAST_CLOSED: dict[Lang, str] = {
    "zh": "已关单,谢谢确认", "en": "Closed — thanks", "ja": "クローズしました",
}
_TOAST_REOPENED: dict[Lang, str] = {
    "zh": "已重新打开,IT 会继续跟",
    "en": "Reopened — IT will pick it back up",
    "ja": "再オープンしました",
}
_TOAST_NOTED: dict[Lang, str] = {
    "zh": "已经记下了", "en": "Noted", "ja": "記録しました",
}
_TOAST_CANNOT: dict[Lang, str] = {
    "zh": "这张工单现在的状态做不了这个",
    "en": "The ticket's current state does not allow that",
    "ja": "現在のステータスでは実行できません",
}
_TOAST_CANNOT_RATE: dict[Lang, str] = {
    "zh": "现在评不了(可能已经评过,或者工单还没解决)",
    "en": "Cannot rate right now (already rated, or not resolved yet)",
    "ja": "現在は評価できません（既に評価済み、または未解決）",
}
#: 工单满意度的回执。与答案卡那颗 👍 的 ``_TOAST_RATED`` 是两件事:一个评的是
#: 「这次答得怎么样」,一个评的是「这次 IT 处理得怎么样」。
_TOAST_CSAT_RECORDED: dict[Lang, str] = {
    "zh": "谢谢,已经记下了", "en": "Thanks — recorded", "ja": "ありがとうございます",
}
_TOAST_RESOLUTION_DRAFTED: dict[Lang, str] = {
    "zh": "结案说明发在群里了,确认一下",
    "en": "The draft resolution is in the room — confirm it",
    "ja": "対応内容の案をグループに送りました",
}
_TOAST_TYPE_RESOLUTION: dict[Lang, str] = {
    "zh": "群里还没有可整理的讨论 —— 直接说「解决了 <说明>」就行",
    "en": "Nothing to summarise yet — just say “resolved <your text>”",
    "ja": "まとめる議論がありません。「解決 <説明>」とお送りください",
}
_TOAST_STALE_RESOLUTION: dict[Lang, str] = {
    "zh": "这段结案说明已经不是最新的了,请重新点「我弄完了」",
    "en": "That draft is no longer current — tap 「It's fixed」 again",
    "ja": "この案は最新ではありません。もう一度お試しください",
}
#: 群里那句「你自己写一句」。发在房间里而不是只做一个 toast:toast 一闪而过,
#: 而这是一件需要对方接着做的事。
_ROOM_TYPE_RESOLUTION: dict[Lang, str] = {
    "zh": "群里还没有可以整理成结案说明的讨论。直接说一句「解决了 <怎么解决的>」"
          "我就写进工单。",
    "en": "There is no discussion here I can turn into a resolution yet. Say "
          "“resolved <what fixed it>” and I'll put it on the ticket.",
    "ja": "対応内容にまとめられる議論がまだありません。「解決 <対応内容>」と"
          "お送りいただければチケットに記録します。",
}
_NOTE_STILL_BROKEN: dict[Lang, str] = {
    "zh": "求助人在群里说问题还没解决。",
    "en": "The requester says the problem is not solved yet.",
    "ja": "依頼者から未解決との連絡がありました。",
}
_RESOLVED_TITLE: dict[Lang, str] = {
    "zh": "已标记为解决", "en": "Marked resolved", "ja": "解決済みにしました",
}
_RESOLVED_BODY: dict[Lang, str] = {
    "zh": "工单 {number} 已标记为解决,结案说明已经写进工单。等求助人确认后关单。",
    "en": "Ticket {number} is resolved and the write-up is on the record. "
          "It closes once the requester confirms.",
    "ja": "チケット {number} を解決済みにし、対応内容を記録しました。",
}

_OWNED_ACTIONS = frozenset({
    ACTION_CONFIRM,
    # 帮助卡上的四颗:两颗只是换一张卡(纯渲染),两颗会真的做事(问一句 /
    # 找人工)。四个名字必须和 ``help_card`` 同源 —— 一边画按钮、一边认领事件,
    # 对不上的表现是「按钮在,点了没反应」。
    help_card.ACTION_HOME,
    help_card.ACTION_CATEGORY,
    help_card.ACTION_ASK,
    help_card.ACTION_HUMAN,
    help_card.ACTION_LANG,
    answer_card.ACTION_HELPFUL,
    answer_card.ACTION_UNHELPFUL,
    answer_card.ACTION_FEEDBACK_REASON,
    answer_card.ACTION_ESCALATE,
    answer_card.ACTION_HANDOFF,
    # 工单群里那张卡上的按钮。**必须和 ``ticket_card`` 那三个常量同源** ——
    # 一边画按钮、一边认领事件,两处对不上的表现是「按钮存在但点了没反应」。
    ticket_card.ACTION_CLAIM,
    ticket_card.ACTION_PRIORITY_UP,
    ticket_card.ACTION_REFRESH,
    ticket_card.ACTION_RESOLVE,
    ticket_card.ACTION_RESOLVE_CONFIRM,
    ticket_card.ACTION_CLOSE,
    ticket_card.ACTION_REOPEN,
    ticket_card.ACTION_RATE,
})


def claims_card_action(payload: dict[str, Any]) -> bool:
    """True when this callback is one of our buttons."""
    header = payload.get("header") or {}
    if (header.get("event_type") or payload.get("type")) != CARD_ACTION_EVENT:
        return False
    return _action_value(payload).get("action") in _OWNED_ACTIONS


def _action_value(payload: dict[str, Any]) -> dict[str, Any]:
    """What the tapped button carried, from either card schema's envelope.

    A v1 button declares its payload as a top-level ``value``; a card 2.0
    button declares it inside ``behaviors[].value``. Feishu reports the tap at
    ``event.action.value`` for both — but the cards Helix had already sent
    when the renderer moved to 2.0 keep living in people's chat history for
    months, and a tap on one of those must not fall through to "not our
    button". So both shapes are read, most-specific first.
    """
    event = payload.get("event") or {}
    action = event.get("action") or {}
    value = action.get("value")
    if isinstance(value, dict) and value:
        return value
    for behavior in action.get("behaviors") or []:
        if not isinstance(behavior, dict):
            continue
        nested = behavior.get("value")
        if isinstance(nested, dict) and nested:
            return nested
    return {}


def _pick(table: dict[Lang, str], lang: Lang) -> str:
    return table.get(lang) or table[DEFAULT_LANG]


def _open_message_id(payload: dict[str, Any]) -> str | None:
    """按钮所在那条飞书消息的 id,用来把它原样转发进群。

    读得**防御性**:这个字段没有像 ``action.value`` 那样被实测过,而转人工不该
    因为一个可选字段的形状而失败。拿不到就退回「把回答贴成文本」——群里仍然有
    内容,只是格式差一点。
    """
    context = (payload.get("event") or {}).get("context") or {}
    raw = context.get("open_message_id")
    return raw if isinstance(raw, str) and raw else None


def _closing_card(
    title: str, body: str, *, colour: str, status: str | None = None,
) -> dict[str, Any]:
    """The card the employee is left looking at once the tap is handled.

    Rendered by the platform builder, in card 2.0, for one non-cosmetic
    reason: this card REPLACES the 2.0 confirmation card in place, and the two
    schemas do not mix inside one message. It is also the receipt for an
    action the employee just took, so it should look like every other receipt
    Helix produces rather than like a bare paragraph.
    """
    from app.notifications.service import (  # noqa: PLC0415 — cycle
        _build_interactive_card,
    )

    return _build_interactive_card(
        title=title,
        body_text=body,
        url=None,
        button_label="",
        is_en=False,
        status=status,
        header_color=colour,
        footer="",
    )


class _Handoff:
    """Which side of the 3-second race owns delivering the outcome.

    Single-threaded by construction: each transition happens with no ``await``
    between the read and the write, so whichever coroutine the loop resumes
    first claims delivery and the other stands down. That is what keeps a fast
    tap from being answered twice — once in place and once as a follow-up.
    """

    __slots__ = ("in_flight", "lang", "timed_out")

    def __init__(self) -> None:
        self.timed_out = False
        #: Filled in by the worker as soon as it knows who tapped, so a
        #: "submitting" card written after the budget lapses is still in the
        #: employee's own language.
        self.lang: Lang = DEFAULT_LANG
        #: What to leave in the card's place if the budget lapses, published by
        #: the branch once it knows what the employee is looking at. Without it
        #: the generic working card is the answer — and a slow 转人工 then wiped
        #: the answer and its 参考文档 off the card the employee was reading,
        #: which is the whole defect this carries. Set BEFORE the slow work
        #: starts, or the race it exists for is already lost.
        self.in_flight: dict[str, Any] | None = None

    def claim_followup(self) -> None:
        """Called by the request path when the budget lapsed."""
        self.timed_out = True

    @property
    def request_gave_up(self) -> bool:
        return self.timed_out


async def handle_card_action(
    payload: dict[str, Any], *, trace_id: str,
) -> dict[str, Any]:
    """Answer a tapped button inside Feishu's 3-second window.

    Always returns a card: Feishu renders the response, so an error path that
    returned nothing would leave the employee staring at a button that appears
    to have done nothing.

    Takes no session on purpose. The work runs on its OWN, and all three
    reasons are load-bearing: it may outlive the request (see the budget
    above); FastAPI closes the request session at teardown, so a task still
    holding it would raise into a request nobody is reading any more; and
    touching the caller's session would open a transaction there whose snapshot
    predates the work's commit, which is how the caller ends up reading a row
    the tap has already moved on from.
    """
    handoff = _Handoff()
    task = background.spawn(
        _run_tap(payload, trace_id=trace_id, handoff=handoff),
        name="assistant.card_action",
    )
    try:
        done, _pending = await asyncio.wait(
            {task}, timeout=_RESPONSE_BUDGET_SECONDS,
        )
    except asyncio.CancelledError:
        # The request went away (shutdown, a proxy giving up). The work is still
        # running and is about to have an outcome nobody is listening for, so
        # hand delivery to the follow-up path before unwinding — otherwise the
        # employee's tap silently executes and never reports back.
        handoff.claim_followup()
        raise
    if task in done:
        try:
            response = cast("dict[str, Any]", task.result())
        except Exception:
            # ``_run_tap`` catches its own failures, so this is a bug in the
            # guard itself rather than in a branch. Still answered with a card:
            # Feishu cannot render the API's 500 envelope.
            logger.exception("assistant: card action task raised through guard")
            return _refusal(canned.failed(handoff.lang), handoff.lang)
        # Internal routing key — Feishu is handed the card, never this.
        response.pop(FOLLOWUP_KEY, None)
        return response
    handoff.claim_followup()
    logger.info(
        "assistant: card tap exceeded the %.1fs budget — answering by follow-up",
        _RESPONSE_BUDGET_SECONDS,
    )
    # The branch's own in-place card when it published one: it keeps whatever
    # the employee was reading. The generic card is the fallback for a tap that
    # was still resolving the sender when the budget ran out.
    return handoff.in_flight or _working_card(handoff.lang)


async def _run_tap(
    payload: dict[str, Any], *, trace_id: str, handoff: _Handoff,
) -> dict[str, Any]:
    """Run one tap to completion, then deliver wherever the race left it."""
    from app.core.database import async_session_factory  # noqa: PLC0415

    operator = (payload.get("event") or {}).get("operator") or {}
    async with async_session_factory() as db:
        lang: Lang = DEFAULT_LANG
        try:
            # Resolved once, here, and passed down: the gates below need the
            # sender and the cards need the language, and asking twice would
            # mean two answers to the same question.
            sender = await resolve_sender(
                db,
                union_id=operator.get("union_id"),
                open_id=operator.get("open_id"),
            )
            person = getattr(sender, "person", None)
            # 这一段对话在用哪种语言。**不是**直接 ``resolve_lang`` —— 员工可能
            # 正用英文和助手说话(见 :mod:`app.assistant.reply_language`),而
            # 一颗按钮上没有文字可判,所以按会话走:一段英文对话点一下按钮不该
            # 突然回一句中文。会话不认识就退回这个人的偏好。
            # 这里问的是「**这个房间**在用哪种语言」,所以先看按钮所在的会话
            # (工单群里的那张卡就该跟着群走),拿不到才退回这个人最近的私聊。
            # 与下面 ``_help_action`` 的顺序**故意相反** —— 那边问的是「他的
            # 私聊线程」,不是「这个房间」。
            chat_id = _open_chat_id(payload)
            if not chat_id and person is not None:
                chat_id = await session.active_chat(person_id=person.id)
            lang = await reply_language.for_card(
                person=person, chat_id=chat_id or "",
            )
            handoff.lang = lang
            response = await _dispatch_card_action(
                db, payload, trace_id=trace_id, sender=sender, lang=lang,
                handoff=handoff,
            )
        except Exception:
            # Anything escaping a branch — a database error resolving the
            # sender, an integrity error, a bug — would otherwise surface as
            # the API's 500 error envelope, which Feishu cannot render: the
            # employee sees a dead button and Feishu retries into the same
            # failure. A red card is a worse outcome than success and a better
            # one than silence.
            logger.exception("assistant: card action handler failed")
            await db.rollback()
            response = _refusal(canned.failed(lang), lang)

        if handoff.request_gave_up:
            await _deliver_followup(db, payload, response, lang=lang)
        return response


async def _deliver_followup(
    db: AsyncSession,
    payload: dict[str, Any],
    response: dict[str, Any],
    *,
    lang: Lang,
) -> None:
    """Send the outcome as its own card, for a tap that outran the budget.

    A follow-up message rather than Feishu's delayed card-update API: the
    send-card path is the one this channel already proves on every reply, and a
    result the employee can scroll back to is worth more than replacing a card
    they have already stopped looking at.
    """
    open_id = str(((payload.get("event") or {}).get("operator") or {}).get("open_id") or "")
    # A branch whose in-place card is the employee's whole answer publishes a
    # SEPARATE compact receipt here. Sending the in-place card as the follow-up
    # instead would post the entire answer a second time, under itself.
    if response.get(FOLLOWUP_KEY) is False:
        # 这一支已经把结果送到该去的地方了(群里那张卡原地重画 / 房间里另发了
        # 一张卡)。再往操作者私聊里补一份,等于把一整张工单卡塞进他的对话。
        return
    card = response.get(FOLLOWUP_KEY) or (response.get("card") or {}).get("data")
    if not open_id or not isinstance(card, dict) or not card:
        logger.warning("assistant: no follow-up target for a slow card tap")
        return
    from app.assistant import channel  # noqa: PLC0415 — cycle

    # Keyed on the EVENT, so Feishu's own retry of a slow tap cannot produce two
    # result cards. That retry is deliberately allowed through the webhook's
    # replay guard (the branches here are idempotent), and it lands as a second
    # ``_run_tap``: without this key the employee could get 「已提交」 twice, or —
    # if the retry arrived while the first claim was still in flight — one card
    # saying 「没能提交」 and another saying 「已提交」.
    event_id = str((payload.get("header") or {}).get("event_id") or "")
    delivered = await channel.send_card(
        db, to=channel.ReplyTarget.dm(open_id), card=card,
        idempotency_key=f"assistant-card-followup-{event_id}" if event_id else None,
    )
    if not delivered:
        # The card is the nicety; knowing what happened is the point.
        await channel.send_text(
            db, to=channel.ReplyTarget.dm(open_id),
            text=canned.offer_failed(lang),
        )


async def _dispatch_card_action(
    db: AsyncSession,
    payload: dict[str, Any],
    *,
    trace_id: str,
    sender: Any,
    lang: Lang,
    handoff: _Handoff,
) -> dict[str, Any]:
    """Route one tap to its branch. Wrapped by :func:`handle_card_action`."""
    value = _action_value(payload)
    what = value.get("action")
    person = getattr(sender, "person", None)
    # Captured BEFORE any branch can roll back. A rollback expires every
    # instance in the session, so ``person.id`` read afterwards triggers a lazy
    # refresh and raises ``MissingGreenlet`` from inside the audit helper — the
    # savepoint guard then swallowed it and the refusal went UNRECORDED, which
    # is precisely the case the audit exists to make visible.
    person_id = person.id if person is not None else None

    # The release gate is checked here too. Cards live in chats for an hour and
    # in scrollback forever; switching the assistant off has to stop them, not
    # just stop new conversations.
    if not await assistant_enabled(db):
        await _audit_refusal(
            db, person_id=person_id, reason="assistant_disabled", what=what,
            target=None, trace_id=trace_id,
        )
        return _refusal(canned.kernel_unavailable(lang), lang)

    if person is None:
        await _audit_refusal(
            db, person_id=None, reason="unbound_sender", what=what,
            target=None, trace_id=trace_id,
        )
        return _refusal(canned.unbound(lang), lang)

    # ``resolve_sender`` returns DEPARTED *with* the person attached, so a
    # ``person is None`` test alone lets a leaver through. The message path and
    # the token path both refuse them; this one has to as well, or a pending
    # card stays actionable for an hour after the account was closed.
    if sender.status is not SenderStatus.OK:
        logger.info(
            "assistant: card tap refused, sender status=%s", sender.status,
        )
        await _audit_refusal(
            db, person_id=person_id, reason=f"sender_{sender.status}", what=what,
            target=None, trace_id=trace_id,
        )
        return _refusal(
            canned.departed(lang)
            if sender.status is SenderStatus.DEPARTED
            else canned.sender_mismatch(lang),
            lang,
        )

    if what in (answer_card.ACTION_HELPFUL, answer_card.ACTION_UNHELPFUL):
        return await _record_verdict(
            db,
            person=person,
            message_id=str(value.get("message") or ""),
            helpful=what == answer_card.ACTION_HELPFUL,
            lang=lang,
            trace_id=trace_id,
        )
    if what == answer_card.ACTION_FEEDBACK_REASON:
        return await _record_feedback_reason(
            db,
            person=person,
            message_id=str(value.get("message") or ""),
            reason=str(value.get("reason") or ""),
            lang=lang,
            trace_id=trace_id,
        )
    if what == answer_card.ACTION_HANDOFF:
        return await _escalate(
            db, person=person, message_id=str(value.get("message") or ""),
            lang=lang, trace_id=trace_id, handoff=handoff,
            with_chat=True,
            open_message_id=_open_message_id(payload),
        )
    if what == answer_card.ACTION_ESCALATE:
        return await _escalate(
            db,
            person=person,
            message_id=str(value.get("message") or ""),
            lang=lang,
            trace_id=trace_id,
            handoff=handoff,
        )

    if what in (
        help_card.ACTION_HOME,
        help_card.ACTION_CATEGORY,
        help_card.ACTION_ASK,
        help_card.ACTION_HUMAN,
        help_card.ACTION_LANG,
    ):
        return await _help_action(
            db,
            payload,
            person=person,
            what=str(what),
            value=value,
            lang=lang,
            trace_id=trace_id,
        )

    if what == ticket_card.ACTION_RATE:
        return await _rate_from_room(
            db,
            person=person,
            ticket_id=str(value.get("ticket") or ""),
            score=value.get("score"),
            lang=lang,
            trace_id=trace_id,
        )

    if what in (
        ticket_card.ACTION_CLAIM,
        ticket_card.ACTION_PRIORITY_UP,
        ticket_card.ACTION_REFRESH,
        ticket_card.ACTION_RESOLVE,
        ticket_card.ACTION_RESOLVE_CONFIRM,
        ticket_card.ACTION_CLOSE,
        ticket_card.ACTION_REOPEN,
    ):
        return await _ticket_action(
            db,
            person=person,
            handoff=handoff,
            what=str(what),
            ticket_id=str(value.get("ticket") or ""),
            # 结案确认那一颗带着它显示的那段文本的哈希 —— 见
            # ``_confirm_resolution``:「你确认的」必须和「写进去的」是同一段。
            digest=str(value.get("digest") or ""),
            lang=lang,
            trace_id=trace_id,
        )

    raw_id = str(value.get("proposal") or "")
    try:
        proposal_id = uuid.UUID(raw_id)
    except (ValueError, AttributeError, TypeError):
        logger.warning("assistant: card action with unusable proposal id %r", raw_id)
        return _refusal(canned.failed(lang), lang)

    # Rendered from the row BEFORE the execution that may roll it back — a
    # rollback expires every instance in the session, and reading
    # ``proposal.params`` afterwards raises ``MissingGreenlet`` from a lazy
    # refresh. A plain string survives that; an ORM object does not.
    rows = await _submitted_rows(db, proposal_id=proposal_id, lang=lang)
    handoff.in_flight = _working_card(lang, rows=rows)

    try:
        proposal = await confirm_and_execute_for_person(
            db, proposal_id=proposal_id, person=person, trace_id=trace_id,
        )
    except ProposalError as exc:
        # Feishu re-posts a callback it judged slow, and the work here (an MCP
        # write, sometimes with its own external call) can outrun that deadline.
        # The retry then finds the proposal no longer pending and would show
        # 「没能提交」 for a submission that went through. Ask what actually
        # happened before calling it a failure.
        if exc.code == CODE_ALREADY_HANDLED:
            replayed = await _already_done_card(
                db, proposal_id=proposal_id, person=person, lang=lang, rows=rows,
            )
            if replayed is not None:
                return replayed
        # Expired / someone else's — the message explains which, and the card
        # carries it so the employee is not left guessing.
        logger.info("assistant: card confirm refused (%s)", exc.code)
        await _audit_refusal(
            db, person_id=person_id, reason=exc.code, what=what,
            target=str(proposal_id), trace_id=trace_id,
        )
        # Not ``exc.message``: that string is written for an API envelope and
        # exists only in English, so a colleague whose Helix language is Chinese
        # or Japanese read the one sentence in this channel that was not
        # localised — the one they meet when something went wrong.
        return _refusal(canned.proposal_refused(exc.code, lang), lang)
    except Exception:
        logger.exception("assistant: card confirm failed for %s", proposal_id)
        # Same reason the outer guard rolls back: a failed flush deactivates
        # the transaction, and the audit write below would raise on a session
        # nobody reset.
        await db.rollback()
        return _refusal(canned.offer_failed(lang), lang)

    # The link path audits ``assistant.proposal.confirmed``; the card path is
    # now the primary route, so without this the assistant's own trail loses
    # its main path and ``usage_stats.confirmed_proposals`` silently undercounts.
    await _audit_confirmed(db, person=person, proposal=proposal, trace_id=trace_id)

    if proposal.status != STATUS_EXECUTED:
        return _refusal(canned.offer_failed(lang), lang)
    return _submitted_card(lang, rows=rows)


def _open_chat_id(payload: dict[str, Any]) -> str | None:
    """按钮所在的那个会话。私聊里点一颗按钮,答案要回到同一个会话里去。

    和 :func:`_open_message_id` 一样读得防御性:拿不到就退回「这个人最近在用的
    单聊」,两个都没有时调用方还有一条不依赖会话的路可走。
    """
    context = (payload.get("event") or {}).get("context") or {}
    raw = context.get("open_chat_id")
    return raw if isinstance(raw, str) and raw else None


async def _help_action(
    db: AsyncSession,
    payload: dict[str, Any],
    *,
    person: IdentityPerson,
    what: str,
    value: dict[str, Any],
    lang: Lang,
    trace_id: str,
) -> dict[str, Any]:
    """帮助卡上的四颗按钮。

    两类,区别在于**有没有副作用**:

    * ``home`` / ``category`` 纯换一张卡。没有数据库写、没有外部调用,所以它们
      在 3 秒预算里绰绰有余,原地返回新卡即可。
    * ``ask`` / ``human`` 要跑一整轮或者起一张工单 —— 都可能几十秒。这里只把
      活儿交出去并**立刻**回一张卡(内容不变,底部多一行「已经发出去了」),
      真正的产物作为**另一条消息**到达。这比让飞书等到超时、或者把员工正在看
      的那张卡换成一个转圈的占位符都好。

    ``FOLLOWUP_KEY: False`` 是对 :func:`_deliver_followup` 说的:结果自己会走
    另一条消息过去,别再往私聊里补一份。
    """
    person_id = person.id
    #: 他**显式设过**的语言(没设过是 ``None``)—— 只决定语言那一行里勾哪个,
    #: 不影响这张卡用什么语言画。
    chosen = preferences.current_lang(person)
    open_id = str(
        ((payload.get("event") or {}).get("operator") or {}).get("open_id") or "",
    )
    #: 飞书对一次它判定为慢的回调会重投,而卡片回调**故意**不走 webhook 那道
    #: 重放闸(那批分支都是幂等的)。下面两颗按钮不是:一次重投会让同一个问题
    #: 问两遍、同一件事起两张工单。所以把事件 id 传下去,由后台那一段认领。
    event_id = str((payload.get("header") or {}).get("event_id") or "")
    category_key = str(value.get(help_card.KEY_CATEGORY) or "")
    category = starters.category(category_key)

    if what == help_card.ACTION_HOME:
        return {"card": help_card.raw(help_card.build_home_card(lang, chosen=chosen))}

    if what == help_card.ACTION_LANG:
        return await _switch_language(
            db, person=person, wanted=str(value.get(help_card.KEY_LANG) or ""),
            lang=lang, trace_id=trace_id,
        )

    if what == help_card.ACTION_CATEGORY:
        if category is None:
            # 卡片会在聊天记录里躺几个月,而这张表还会继续被编辑。指向一个已经
            # 不存在的分类时退回一级卡,而不是回一句错误 —— 员工要的是接着问,
            # 不是知道我们改过表。
            return {
                "toast": help_card.gone_toast(lang),
                "card": help_card.raw(help_card.build_home_card(lang, chosen=chosen)),
            }
        return {"card": help_card.raw(help_card.build_category_card(category, lang))}

    # 这两颗按钮动的是**这个人的私聊线程**:答案私聊发、工单引用的是他和助手的
    # 对话、「下一条消息直接成单」说的也是他在私聊里打的下一句。所以线程键优先
    # 取他的私聊指针,拿不到才退回按钮所在的那个会话。
    #
    # 顺序反过来会在一种真实情况下出错:员工把这张卡**转发**进一个群再点它 ——
    # 按钮还在,回调带回来的是群的 chat_id。那时候按群来记,会把他的私聊线程
    # 指针指到群上(菜单里的「新话题」于是清错线程),而「下一条消息成单」的
    # 标记也会记在一个永远不会被兑现的键上(消费只发生在私聊分支)—— 承诺了
    # 一件做不到的事。正常情况下两者是同一个 id,这一行因此是纯保险。
    chat_id = await session.active_chat(person_id=person_id) or _open_chat_id(payload)

    if what == help_card.ACTION_HUMAN:
        background.spawn(
            _help_human(
                person_id=person_id, open_id=open_id, chat_id=chat_id or "",
                lang=lang, trace_id=trace_id, event_id=event_id,
            ),
            name="assistant.help_human",
        )
        card = (
            help_card.build_category_card(category, lang) if category is not None
            else help_card.build_home_card(lang, chosen=chosen)
        )
        return {
            "toast": {"type": "info", "content": _pick(_TOAST_WORKING, lang)},
            "card": help_card.raw(card),
            FOLLOWUP_KEY: False,
        }

    # ── ask ────────────────────────────────────────────────────────────
    starter_key = str(value.get(help_card.KEY_STARTER) or "")
    question = starters.question(category_key, starter_key, lang)
    if category is None or question is None:
        return {
            "toast": help_card.gone_toast(lang),
            "card": help_card.raw(help_card.build_home_card(lang, chosen=chosen)),
        }
    if not chat_id:
        # 没有会话就没有线程,也就没有「接着聊」。这不该表现成一次静默失败:
        # 直接说清楚该怎么办,员工打一行字就能继续。
        logger.warning(
            "assistant: starter tap without a chat id person=%s", person_id,
        )
        return {
            "toast": help_card.gone_toast(lang),
            "card": help_card.raw(help_card.build_category_card(category, lang)),
        }

    from app.assistant.feishu_events import ask_starter  # noqa: PLC0415 — cycle

    background.spawn(
        ask_starter(
            person_id=person_id,
            open_id=open_id,
            chat_id=chat_id,
            question=question,
            lang=lang,
            trace_id=trace_id,
            starter_key=f"{category_key}.{starter_key}",
            event_id=event_id,
        ),
        name="assistant.help_ask",
    )
    return {
        "toast": help_card.ask_toast(lang),
        "card": help_card.raw(
            help_card.build_category_card(category, lang, sent=question),
        ),
        FOLLOWUP_KEY: False,
    }


async def _switch_language(
    db: AsyncSession,
    *,
    person: IdentityPerson,
    wanted: str,
    lang: Lang,
    trace_id: str,
) -> dict[str, Any]:
    """帮助卡上的语言自助。

    这一颗**直接写**,不走确认卡:改一个自己的显示偏好是可逆的、只影响自己,
    而为它加一次确认,等于让一个「这东西说的不是我的语言」的人多读一屏他读不
    懂的字。确认卡留给会提交申请、会动别人东西的那些动作。

    写完之后**清掉这个会话的语言记忆**:那一格记的是「他刚才用什么语言说话」,
    而他刚刚明确说了要用哪种 —— 不清的话,下一句「ok」会把他拉回旧语言,而他
    刚按过按钮,只会觉得这个开关坏了。
    """
    if wanted not in ("zh", "en", "ja"):
        logger.warning("assistant: language tap with unusable code %r", wanted)
        return _refusal(canned.failed(lang), lang)
    picked = cast("Lang", wanted)

    changed = await preferences.set_preferred_language(
        db, person=person, lang=picked, source="assistant_help_card",
        trace_id=trace_id,
    )
    if changed:
        await db.commit()
    chat_id = await session.active_chat(person_id=person.id)
    if chat_id:
        await session.forget_lang(person_id=person.id, chat_id=chat_id)
    return {
        "toast": help_card.language_switched_toast(picked),
        "card": help_card.raw(
            help_card.build_home_card(picked, chosen=picked),
        ),
    }


async def _help_human(
    *,
    person_id: Any,
    open_id: str,
    chat_id: str,
    lang: Lang,
    trace_id: str,
    event_id: str = "",
) -> str:
    """帮助卡上的「转人工」。

    和员工打字说「转人工」走的是同一个 :func:`escalation.offer_ticket` —— 一件
    事只有一条路,否则两条路迟早给出两种工单。对话还是空的时候该怎么办(不硬起
    一张没有内容的工单,而是先问一句)也在那一个函数里,原因同上。

    **自带 session**:调用方要在 3 秒内把响应还给飞书,所以这段是
    ``background.spawn`` 出来的,活得比那次请求久;用请求的 session 会在
    FastAPI 拆掉它之后炸在半路。人也在这里重新读一遍,不跨 session 复用实例。
    """
    from app.assistant import escalation  # noqa: PLC0415 — cycle
    from app.assistant.feishu_events import claim_event  # noqa: PLC0415 — cycle
    from app.core.database import async_session_factory  # noqa: PLC0415
    from app.identity.person_models import IdentityPerson as _Person  # noqa: PLC0415

    # 飞书重投同一次点击 ⇒ 同一件事两张确认卡。见 ``_help_action`` 里那段注释。
    if event_id and not await claim_event(event_id):
        logger.info("assistant: duplicate handoff tap %s", event_id)
        return "duplicate"

    async with async_session_factory() as db:
        person = await db.get(_Person, person_id)
        if person is None:
            logger.warning("assistant: handoff tap for unknown person %s", person_id)
            return "unbound"
        return await escalation.offer_ticket(
            db, person=person, chat_id=chat_id, open_id=open_id,
            lang=lang, trace_id=trace_id,
        )


async def _ticket_action(
    db: AsyncSession,
    *,
    person: IdentityPerson,
    handoff: _Handoff,
    what: str,
    ticket_id: str,
    lang: Lang,
    trace_id: str,
    digest: str = "",
) -> dict[str, Any]:
    """群里那张工单卡上的一次点击。

    ## 谁按的,重新问一次

    这张卡摆在**所有群成员**面前,所以「它在他面前」不是授权 —— 每一次点击都
    重新问 `can_act_on_ticket`(IT 管理员,或这张工单的处理人)。按下去被拒的
    人会看到「谁可以做这件事」,而不是一句「无权操作」:后者读起来像平台坏了。

    ## 返回的是**整张卡的新样子**

    飞书把回调的返回内容当成这条消息的新卡片渲染。所以每个分支都必须重画完整的
    卡 —— 返回一个两行的回执会把群里那张工单卡整个替换掉,而它是这个房间里唯一
    能看到状态的地方。
    """
    from app.assistant import ticket_chat  # noqa: PLC0415 — cycle
    from app.domains.support.models import SupportTicket  # noqa: PLC0415
    from app.domains.support.service import (  # noqa: PLC0415
        SupportService,
        can_act_on_ticket,
        role_names_for_person,
    )

    try:
        parsed = uuid.UUID(ticket_id)
    except (ValueError, AttributeError, TypeError):
        logger.warning("assistant: ticket card tap with unusable id %r", ticket_id)
        return _refusal(canned.failed(lang), lang)

    ticket = await db.get(SupportTicket, parsed)
    if ticket is None:
        return _refusal(canned.failed(lang), lang)
    chat = await ticket_chat.get_by_ticket(db, parsed)

    async def _rendered(toast: str, *, toast_type: str = "success") -> dict[str, Any]:
        """当前状态的完整卡片 + 一条浮层提示。

        ``_FOLLOWUP_NONE``:慢的那一下不再往操作者私聊补一份 —— 群里那张卡会
        由工单状态变化的后台刷新原地重画,而把整张工单卡塞进某个人的私聊是纯
        噪音。
        """
        requester = await db.get(IdentityPerson, ticket.requester_id)
        assignee = (
            await db.get(IdentityPerson, ticket.assignee_id)
            if ticket.assignee_id else None
        )
        card = ticket_card.build_ticket_card(
            ticket_number=ticket.ticket_number,
            ticket_id=ticket.id,
            title=ticket.title,
            status=ticket.status,
            priority=ticket.priority,
            requester_name=(
                (requester.display_name or requester.work_email) if requester else ""
            ) or "",
            assignee_name=(
                (assignee.display_name or assignee.work_email) if assignee else None
            ),
            lang=lang,
            portal_url=f"/me/support/tickets/{ticket.id}",
        )
        return {
            "toast": {"type": toast_type, "content": toast},
            "card": {"type": "raw", "data": card},
            FOLLOWUP_KEY: False,
        }

    # 🔴 **慢的那一下不能把这张卡换掉。** 飞书把回调的返回当成这条消息的新
    # 内容,而超预算时 ``handle_card_action`` 返回的是 ``handoff.in_flight``
    # 或者一张通用的「正在处理」卡 —— 后者会把群里唯一能看到工单状态的那张卡
    # 整个替换掉。所以在任何慢活(读群、跑模型)之前先把**当前这张卡**发布成
    # in-flight。答案卡的转人工分支为同一件事踩过一次。
    handoff.in_flight = await _rendered(
        _pick(_TOAST_WORKING, lang), toast_type="info",
    )

    if what == ticket_card.ACTION_REFRESH:
        # 只读:谁都能按。它存在的理由是「这张卡是不是已经过期了」这个疑问在
        # 群里没有别的办法回答。
        return await _rendered(_pick(_TOAST_REFRESHED, lang), toast_type="info")

    # ── 求助人那两颗按钮:授权判据不是「处理人」,是「这是你的工单」 ────
    #
    # 「问题解决了」和「还有问题」只有求助人能按 —— 前者是他对结果的确认,
    # 后者是他说这件事没完。让处理人替他按第一颗,等于自己给自己发合格证。
    if what in (ticket_card.ACTION_CLOSE, ticket_card.ACTION_REOPEN):
        return await _requester_verdict(
            db, person=person, ticket=ticket, chat=chat, what=str(what),
            lang=lang, trace_id=trace_id, rendered=_rendered,
        )

    roles = await role_names_for_person(db, person.id)
    if not can_act_on_ticket(
        ticket=ticket, person_id=person.id, role_names=roles,
    ):
        await _audit_refusal(
            db, person_id=person.id, reason="not_ticket_agent", what=what,
            target=str(parsed), trace_id=trace_id,
        )
        assignee = (
            await db.get(IdentityPerson, ticket.assignee_id)
            if ticket.assignee_id else None
        )
        return await _rendered(
            _pick(_TOAST_NOT_AGENT, lang).format(
                who=(assignee.display_name or assignee.work_email)
                if assignee else _pick(_TOAST_NOBODY, lang),
            ),
            toast_type="error",
        )

    if what == ticket_card.ACTION_RESOLVE:
        return await _propose_resolution(
            db, person=person, ticket=ticket, chat=chat, lang=lang,
            trace_id=trace_id, rendered=_rendered,
        )

    if what == ticket_card.ACTION_RESOLVE_CONFIRM:
        return await _confirm_resolution(
            db, person=person, ticket=ticket, chat=chat,
            digest=digest, lang=lang, trace_id=trace_id,
        )

    try:
        if what == ticket_card.ACTION_CLAIM:
            if ticket.assignee_id == person.id:
                return await _rendered(
                    _pick(_TOAST_ALREADY_YOURS, lang), toast_type="info",
                )
            await SupportService(db).claim(
                ticket=ticket, agent=person, trace_id=trace_id,
            )
            toast = _pick(_TOAST_CLAIMED, lang)
        else:
            await SupportService(db).set_attributes(
                ticket=ticket, actor=person, priority="high", trace_id=trace_id,
            )
            toast = _pick(_TOAST_PRIORITY, lang)
        if chat is not None:
            ticket_chat.touch(chat)
        await db.commit()
    except Exception:
        logger.exception("assistant: ticket card action %s failed", what)
        # 失败的 flush 会让整个事务失效,而下面还要读 ticket 去重画卡片。
        await db.rollback()
        return _refusal(canned.offer_failed(lang), lang)

    return await _rendered(toast)


async def _requester_verdict(
    db: AsyncSession,
    *,
    person: IdentityPerson,
    ticket: Any,
    chat: Any,
    what: str,
    lang: Lang,
    trace_id: str,
    rendered: Callable[..., Awaitable[dict[str, Any]]],
) -> dict[str, Any]:
    """求助人说「好了」或者「还有问题」。

    这两颗按钮的授权判据**不是**「处理人」,而是「这是你的工单」:前者是求助人
    对结果的确认,后者是他说这件事没完。让处理人替他按第一颗,等于自己给自己
    发合格证 —— 而工单的满意度和「真的解决了吗」这两个数从此都不可信。

    IT 管理员也放行:他们本来就能替员工关单(后台一直如此),而群里少这条会让
    一张已经在别处关掉的工单在房间里永远悬着。
    """
    from app.assistant import ticket_chat  # noqa: PLC0415 — cycle
    from app.domains.support.service import (  # noqa: PLC0415
        SupportService,
        role_names_for_person,
    )

    roles = await role_names_for_person(db, person.id)
    is_requester = ticket.requester_id == person.id
    if not is_requester and "global_admin" not in roles:
        await _audit_refusal(
            db, person_id=person.id, reason="not_requester", what=what,
            target=str(ticket.id), trace_id=trace_id,
        )
        return await rendered(_pick(_TOAST_NOT_REQUESTER, lang), toast_type="error")

    svc = SupportService(db)
    try:
        if what == ticket_card.ACTION_CLOSE:
            await svc.close(
                ticket=ticket, actor=person, is_agent=not is_requester,
                trace_id=trace_id,
            )
            toast = _pick(_TOAST_CLOSED, lang)
        elif ticket.status in ("resolved", "closed"):
            await svc.reopen(
                ticket=ticket, actor=person, is_agent=not is_requester,
                trace_id=trace_id,
            )
            toast = _pick(_TOAST_REOPENED, lang)
        else:
            # 还没被标记解决的单子上按「还有问题」——没有什么可重开的,但这句话
            # 本身是信息:记一条,别把它变成一次「无效操作」的拒绝。
            await svc.add_note(
                ticket=ticket, author=person,
                body=_pick(_NOTE_STILL_BROKEN, lang), trace_id=trace_id,
            )
            toast = _pick(_TOAST_NOTED, lang)
        if chat is not None:
            ticket_chat.touch(chat)
        await db.commit()
    except AppException as exc:
        logger.info("assistant: requester verdict refused (%s)", exc.code)
        await db.rollback()
        return await rendered(_pick(_TOAST_CANNOT, lang), toast_type="error")
    except Exception:
        logger.exception("assistant: requester verdict failed")
        await db.rollback()
        return _refusal(canned.offer_failed(lang), lang)

    # 关单之后紧接着发评价卡:这是这段处理的收尾,而员工正在看着这个房间。
    if what == ticket_card.ACTION_CLOSE and chat is not None:
        await offer_room_rating(db, chat=chat, ticket=ticket, lang=lang)
    return await rendered(toast)


async def _propose_resolution(
    db: AsyncSession,
    *,
    person: IdentityPerson,
    ticket: Any,
    chat: Any,
    lang: Lang,
    trace_id: str,
    rendered: Callable[..., Awaitable[dict[str, Any]]],
) -> dict[str, Any]:
    """「我弄完了」→ 先整理一段结案说明,发到群里让人看一眼。

    **不直接结案。** 那段字会成为工单上的正式记录,而一段没人读过就写上去的
    说明会被下一个人当成事实 —— 这条链路上所有 AI 产出的文本都按这条办。

    整理来自群里的讨论。没有讨论、或模型不可用时,如实说「你自己写一句」——
    而不是拿一句「已解决」凑数:那种记录等于没有记录。
    """
    from app.assistant import (  # noqa: PLC0415 — cycle
        channel,
        room_history,
        room_summary,
        ticket_chat,
    )

    if chat is None:
        return await rendered(_pick(_TOAST_TYPE_RESOLUTION, lang), toast_type="info")
    transcript = await room_history.read_room(db, chat_id=chat.feishu_chat_id)
    summary = None
    if transcript is not None and transcript.human_messages:
        summary = await room_summary.summarize(
            db, transcript=transcript, ticket_title=ticket.title, lang=lang,
        )
    if summary is None:
        await ticket_chat.post_note(
            db, chat=chat, text=_pick(_ROOM_TYPE_RESOLUTION, lang),
        )
        return await rendered(
            _pick(_TOAST_TYPE_RESOLUTION, lang), toast_type="info",
        )

    resolution = summary.render(lang)
    card = ticket_card.build_resolution_proposal_card(
        ticket_id=ticket.id,
        resolution=resolution,
        digest=_digest(resolution),
        lang=lang,
    )
    await channel.send_card(
        db, to=channel.ReplyTarget.chat(chat.feishu_chat_id), card=card,
        idempotency_key=f"ticket-resolution-{ticket.id}-{_digest(resolution)[:8]}",
    )
    # 这段说明**同时**记进工单:它是这次讨论的过程记录,而过程记录不该等结案
    # 才落地 —— 万一这个群明天就解散了。
    from app.domains.support.service import SupportService  # noqa: PLC0415

    await SupportService(db).add_note(
        ticket=ticket, author=person, body=resolution, trace_id=trace_id,
    )
    if chat is not None:
        ticket_chat.touch(chat)
    await db.commit()
    return await rendered(_pick(_TOAST_RESOLUTION_DRAFTED, lang), toast_type="info")


async def _confirm_resolution(
    db: AsyncSession,
    *,
    person: IdentityPerson,
    ticket: Any,
    chat: Any,
    digest: str,
    lang: Lang,
    trace_id: str,
) -> dict[str, Any]:
    """确认那段结案说明 → 真的结案。

    ``digest`` 必须对得上**卡片上显示的那段文本**。对不上就拒绝 —— 「你确认的和
    写进去的不是同一段」是这条链路上最难发现的失败,而助手的确认卡为此早就
    立了同一条规矩。
    """
    from app.assistant import ticket_chat  # noqa: PLC0415 — cycle
    from app.domains.support.models import MSG_NOTE  # noqa: PLC0415
    from app.domains.support.service import SupportService  # noqa: PLC0415

    resolution = await _latest_note(db, ticket_id=ticket.id, kind=MSG_NOTE)
    if not resolution or _digest(resolution) != digest:
        logger.info("assistant: resolution digest mismatch on %s", ticket.id)
        await _audit_refusal(
            db, person_id=person.id, reason="digest_mismatch",
            what=ticket_card.ACTION_RESOLVE_CONFIRM,
            target=str(ticket.id), trace_id=trace_id,
        )
        return _refusal(_pick(_TOAST_STALE_RESOLUTION, lang), lang)
    try:
        await SupportService(db).resolve(
            ticket=ticket, agent=person, resolution=resolution,
            trace_id=trace_id,
        )
        if chat is not None:
            ticket_chat.touch(chat)
        await db.commit()
    except AppException as exc:
        logger.info("assistant: resolve refused (%s)", exc.code)
        await db.rollback()
        return _refusal(_pick(_TOAST_CANNOT, lang), lang)
    except Exception:
        logger.exception("assistant: resolve from card failed")
        await db.rollback()
        return _refusal(canned.offer_failed(lang), lang)
    # 结案 = 该请求助人评价了。
    if chat is not None:
        await offer_room_rating(db, chat=chat, ticket=ticket, lang=lang)
    return {
        **_closing_card(
            _pick(_RESOLVED_TITLE, lang),
            _pick(_RESOLVED_BODY, lang).format(number=ticket.ticket_number),
            colour="green",
            status=card_design.STATUS_DONE,
        ),
        FOLLOWUP_KEY: False,
    }


async def _rate_from_room(
    db: AsyncSession,
    *,
    person: IdentityPerson,
    ticket_id: str,
    score: Any,
    lang: Lang,
    trace_id: str,
) -> dict[str, Any]:
    """群里的三档满意度。只有求助人本人能评。"""
    from app.domains.support.models import SupportTicket  # noqa: PLC0415
    from app.domains.support.service import SupportService  # noqa: PLC0415

    try:
        parsed = uuid.UUID(ticket_id)
        value = int(score)
    except (ValueError, AttributeError, TypeError):
        return _refusal(canned.failed(lang), lang)
    ticket = await db.get(SupportTicket, parsed)
    if ticket is None:
        return _refusal(canned.failed(lang), lang)
    if ticket.requester_id != person.id:
        await _audit_refusal(
            db, person_id=person.id, reason="not_requester",
            what=ticket_card.ACTION_RATE, target=ticket_id, trace_id=trace_id,
        )
        return _refusal(_pick(_TOAST_NOT_REQUESTER, lang), lang)
    try:
        await SupportService(db).rate_ticket(
            ticket=ticket, actor=person, score=value, trace_id=trace_id,
            via="feishu_group",
        )
        await db.commit()
    except AppException as exc:
        logger.info("assistant: rating refused (%s)", exc.code)
        await db.rollback()
        # 已经评过、或者工单还没解决 —— 两种都不是错误,如实说一句。
        return _refusal(_pick(_TOAST_CANNOT_RATE, lang), lang)
    except Exception:
        logger.exception("assistant: rating from room failed")
        await db.rollback()
        return _refusal(canned.offer_failed(lang), lang)
    return {
        "toast": {
            "type": "success", "content": _pick(_TOAST_CSAT_RECORDED, lang),
        },
        "card": {"type": "raw", "data": ticket_card.rated_card(
            score=value, lang=lang,
        )},
        FOLLOWUP_KEY: False,
    }


async def offer_room_rating(
    db: AsyncSession, *, chat: Any, ticket: Any, lang: Lang,
) -> None:
    """把评价卡发到群里。已经评过就不发。

    公开的,因为在**群外**结案(后台点了「解决」)时也该问这一句 —— 而那条路
    走的是 ``ticket_chat_coordinator`` 的后台刷新任务。让协调器自己拼一张卡等于
    第二份实现,而卡片的按钮 payload 与这里的处理函数必须一致。

    幂等键带工单 id:飞书一小时内的重复投递会被它挡住,而一张单子被反复
    resolve → reopen → resolve 时确实该再问一次。
    """
    from app.assistant import channel  # noqa: PLC0415

    if getattr(ticket, "satisfaction_score", None) is not None:
        return
    await channel.send_card(
        db,
        to=channel.ReplyTarget.chat(chat.feishu_chat_id),
        card=ticket_card.build_rating_card(ticket_id=ticket.id, lang=lang),
        idempotency_key=f"ticket-rate-{ticket.id}",
    )


async def _latest_note(
    db: AsyncSession, *, ticket_id: Any, kind: str,
) -> str | None:
    """这张工单上最后一条记录的正文。"""
    from sqlalchemy import select  # noqa: PLC0415

    from app.domains.support.models import SupportTicketMessage  # noqa: PLC0415

    return (await db.execute(
        select(SupportTicketMessage.body)
        .where(
            SupportTicketMessage.ticket_id == ticket_id,
            SupportTicketMessage.kind == kind,
        )
        .order_by(SupportTicketMessage.created_at.desc())
        .limit(1)
    )).scalars().first()


def _digest(text: str) -> str:
    """卡片上那段文本的指纹。

    和助手确认卡同一个用途:确认的那一下必须回显它,于是「中途换了内容」会被
    拒绝,而不是提交一份人没读过的东西。
    """
    import hashlib  # noqa: PLC0415

    return hashlib.sha256(text.encode("utf-8")).hexdigest()


async def _submitted_rows(
    db: AsyncSession, *, proposal_id: uuid.UUID, lang: Lang,
) -> str | None:
    """The 将要提交的内容 lines, re-rendered from the stored parameters.

    Feishu replaces the whole card with whatever the tap returns, so a receipt
    that carried only 「✅ 已提交」 deleted the parameter rows the employee had
    just approved — leaving a chat that records a submission with no record of
    WHAT was submitted. Same failure the answer card had with 转人工, one
    surface over.

    Best-effort: a derived action whose summariser trips must not cost the
    employee their receipt, so a failure degrades to the receipt without rows.
    """
    from app.assistant.actions import get_action  # noqa: PLC0415 — cycle
    from app.assistant.confirm_card import clip_value  # noqa: PLC0415 — cycle
    from app.assistant.models import AssistantActionProposal  # noqa: PLC0415

    try:
        proposal = await db.get(AssistantActionProposal, proposal_id)
        if proposal is None:
            return None
        action = get_action(proposal.action_key)
        if action is None:
            return None
        summary = action.summarise(dict(proposal.params or {}), lang)
        lines = [
            f"**{row.localized(lang)['label']}**: "
            f"{clip_value(row.localized(lang)['value'])}"
            for row in summary
        ]
    except Exception:  # noqa: BLE001 — a receipt is worth more than its rows
        logger.warning(
            "assistant: could not summarise proposal %s for its receipt",
            proposal_id, exc_info=True,
        )
        return None
    return "\n".join(lines) or None


async def _already_done_card(
    db: AsyncSession,
    *,
    proposal_id: uuid.UUID,
    person: IdentityPerson,
    lang: Lang,
    rows: str | None = None,
) -> dict[str, Any] | None:
    """The success card, when this tap lost only to its own earlier retry.

    ``confirmed`` counts as well as ``executed``, and that is the common case:
    Feishu re-posts because the FIRST call was SLOW, so at retry time the
    earlier tap is usually still running — ``_claim`` flips the row to
    ``confirmed`` before the work starts. Recognising only ``executed`` left
    the employee reading 「没能提交」 for the submission they were waiting on,
    which is the defect this function exists for. Reporting ``confirmed`` as
    submitted is safe: the conditional claim already succeeded, so exactly one
    execution is under way, and a failure resets the row to ``pending`` rather
    than leaving it here.

    Returns ``None`` when the proposal is not this person's or never got that
    far — in which case the caller falls through to the honest refusal.

    ``populate_existing=True`` because the caller has ALREADY loaded this row
    in this session, and the sessions here do not expire on commit: a plain
    ``get`` would hand back that first copy, still reading ``pending``, and the
    whole point of this branch is to find out what the row says NOW. Without
    it the only redelivery this recognises is one that was already terminal
    before the tap was handled — i.e. not the race it exists for.
    """
    from app.assistant.models import (  # noqa: PLC0415
        STATUS_CONFIRMED,
        AssistantActionProposal,
    )

    proposal = await db.get(
        AssistantActionProposal, proposal_id, populate_existing=True,
    )
    if proposal is None or str(proposal.person_id) != str(person.id):
        return None
    if proposal.status not in (STATUS_EXECUTED, STATUS_CONFIRMED):
        return None
    return _submitted_card(lang, rows=rows)


def _submitted_body(lang: Lang, rows: str | None) -> str:
    body = f"{_pick(_DONE_BODY, lang)}\n{_pick(_WHERE, lang)}"
    if not rows:
        return body
    return f"{body}\n\n**{_pick(_SUBMITTED_SECTION, lang)}**\n{rows}"


def _submitted_card(lang: Lang, *, rows: str | None = None) -> dict[str, Any]:
    return {
        "toast": {"type": "success", "content": _pick(_TOAST_DONE, lang)},
        "card": {
            "type": "raw",
            "data": _closing_card(
                _pick(_DONE_TITLE, lang),
                _submitted_body(lang, rows),
                colour="green",
                status=card_design.STATUS_DONE,
            ),
        },
    }


#: Audited on every refused tap. The read bridge already audits
#: ``assistant.tool.refused`` on the same reasoning: a refusal is the signal
#: that something reached past the boundary, and that is exactly what an
#: administrator should be able to notice and count. Without this, the
#: forwarded-card impersonation attempt this release blocks is invisible —
#: blocked and unrecorded is only half of the fix.
ACTION_CARD_REFUSED = "assistant.card.refused"


async def _audit_refusal(
    db: AsyncSession,
    *,
    person_id: Any | None,
    reason: str,
    what: str | None,
    target: str | None,
    trace_id: str,
) -> None:
    """Record a tap that was turned away.

    SAVEPOINT-isolated for the same reason ``_audit_confirmed`` is: a failed
    flush deactivates the whole transaction, and a refusal must still reach the
    employee as a card rather than as the API's 500 envelope, which Feishu
    cannot render.
    """
    from app.audit.service import AuditService  # noqa: PLC0415
    from app.core.txn import best_effort  # noqa: PLC0415

    async with best_effort(
        db, label="assistant card refusal audit", log=logger,
    ) as step:
        await AuditService(db).log(
            trace_id=trace_id,
            actor_type="user" if person_id is not None else "system",
            actor_user_id=person_id,
            action=ACTION_CARD_REFUSED,
            resource_type="assistant_card_action",
            resource_id=target,
            status="denied",
            detail={"reason": reason, "button": what},
        )
    if step.ok:
        await db.commit()


def _working_card(lang: Lang, *, rows: str | None = None) -> dict[str, Any]:
    """What replaces the buttons while a slow submission finishes.

    Carries no buttons of its own: the tap has been accepted and re-tapping
    would only race the work that is already running. ``rows`` carries the
    confirmation card's own 将要提交的内容 forward, so a submission that outran
    the budget does not cost the employee the record of what they approved.
    """
    body = _pick(_WORKING_BODY, lang)
    if rows:
        body = f"{body}\n\n**{_pick(_SUBMITTED_SECTION, lang)}**\n{rows}"
    return {
        "toast": {"type": "info", "content": _pick(_TOAST_WORKING, lang)},
        "card": {
            "type": "raw",
            "data": _closing_card(
                _pick(_WORKING_TITLE, lang), body, colour="blue",
            ),
        },
    }


def _refusal(message: str, lang: Lang) -> dict[str, Any]:
    return {
        "toast": {"type": "error", "content": _pick(_TOAST_FAILED, lang)},
        "card": {
            "type": "raw",
            "data": _closing_card(_pick(_TOAST_FAILED, lang), message, colour="red"),
        },
    }


def assert_still_actionable(proposal: Any) -> None:
    """Re-exported so the card path and the page path refuse identically."""
    assert_actionable(proposal)


async def _own_message(
    db: AsyncSession, message_id: uuid.UUID, person: IdentityPerson,
) -> Any | None:
    """The message, only if it belongs to this employee's own conversation.

    Thin alias over :func:`app.assistant.feedback.own_message`, kept because the
    escalate branch below reads better with the short name. The check itself is
    shared with the web page: a Feishu card is forwardable by default and a web
    client sends whatever id it likes, so both surfaces need it and neither may
    have its own answer.
    """
    from app.assistant.feedback import own_message  # noqa: PLC0415

    return await own_message(db, message_id, person)


async def _record_verdict(
    db: AsyncSession,
    *,
    person: IdentityPerson,
    message_id: str,
    helpful: bool,
    lang: Lang,
    trace_id: str,
) -> dict[str, Any]:
    """Store the verdict, then spend the buttons.

    The record itself lives in :mod:`app.assistant.feedback`, shared with the
    web page — who may rate what, and how a changed mind is stored, must not
    depend on which surface the employee tapped. What stays here is the card:
    the answer text is re-rendered from the stored message so it keeps saying
    what the employee came for, because replacing it with a thank-you would
    delete the answer they are reading.
    """
    from app.assistant import feedback as feedback_store  # noqa: PLC0415

    try:
        parsed = uuid.UUID(message_id)
    except (ValueError, AttributeError, TypeError):
        logger.warning("assistant: feedback tap with unusable message id %r", message_id)
        return _refusal(canned.failed(lang), lang)

    message = await feedback_store.record_verdict(
        db, person=person, message_id=parsed, helpful=helpful, trace_id=trace_id,
    )
    if message is None:
        await _audit_refusal(
            db, person_id=person.id, reason="not_your_message", what="feedback",
            target=message_id, trace_id=trace_id,
        )
        return _refusal(canned.failed(lang), lang)

    return {
        "toast": {
            "type": "success",
            "content": _pick(_TOAST_RATED, lang),
        },
        "card": {
            "type": "raw",
            "data": answer_card.rated_card(
                message.content, helpful=helpful, lang=lang,
                # Read back rather than dropped: rating an answer used to delete
                # the 参考文档 links it cited (migration 518).
                docs=message.sources,
                message_id=str(message.id),
            ),
        },
    }


async def _record_feedback_reason(
    db: AsyncSession,
    *,
    person: IdentityPerson,
    message_id: str,
    reason: str,
    lang: Lang,
    trace_id: str,
) -> dict[str, Any]:
    """Attach an optional structured reason to the employee's negative vote."""
    from app.assistant import feedback as feedback_store  # noqa: PLC0415

    try:
        parsed = uuid.UUID(message_id)
    except (ValueError, AttributeError, TypeError):
        return _refusal(canned.failed(lang), lang)
    message = await feedback_store.record_reason(
        db, person=person, message_id=parsed, reason=reason, trace_id=trace_id,
    )
    if message is None:
        return _refusal(canned.failed(lang), lang)
    return {
        "toast": {"type": "success", "content": _pick(_TOAST_RATED, lang)},
        "card": {
            "type": "raw",
            "data": answer_card.rated_card(
                message.content, helpful=False, lang=lang, docs=message.sources,
            ),
        },
    }


async def _escalate(
    db: AsyncSession,
    *,
    person: IdentityPerson,
    message_id: str,
    lang: Lang,
    trace_id: str,
    handoff: _Handoff,
    with_chat: bool = False,
    open_message_id: str | None = None,
) -> dict[str, Any]:
    """Hand the exchange to a human without making the employee retype it.

    The ticket carries the question and the answer that missed, which is the
    whole point: an escalation that says only "AI didn't help" costs IT the
    reconstruction the transcript already has.

    Every card this returns still SHOWS the answer. Feishu renders a callback's
    response as the new card, so the old receipt — a two-line 「已转人工处理」
    box — replaced the answer and its 参考文档 with nothing, and the employee
    scrolled back to find the reply they had just been reading gone. The
    receipt the employee needs to keep (the ticket number) arrives as its own
    message instead; the card only gains a line saying what happened.

    ``with_chat`` is the 转人工 button rather than 建工单: after the ticket is
    filed, classify the question, look up who owns that area, and open a Feishu
    group with the employee and that person in it. **The ticket is filed either
    way** — the group is the fast path, the ticket is the ledger, and a ledger
    must not depend on whether an external call succeeded. One code path for
    both buttons on purpose: all the rollback / MissingGreenlet discipline
    below is subtle, and a second copy of it would drift.
    """
    from sqlalchemy import select  # noqa: PLC0415

    from app.assistant.models import AssistantMessage  # noqa: PLC0415
    from app.core.coordinators.assistant_support_coordinator import (  # noqa: PLC0415
        create_assistant_ticket,
    )

    # An id we cannot read is refused, exactly as the rating branch refuses it.
    # Falling through to "escalate anyway" would file a ticket carrying neither
    # the question nor the answer — a blank row that costs IT the reconstruction
    # this whole path exists to spare them.
    try:
        parsed = uuid.UUID(message_id)
    except (ValueError, AttributeError, TypeError):
        logger.warning("assistant: escalate tap with unusable message id %r", message_id)
        return _refusal(canned.failed(lang), lang)

    message = await _own_message(db, parsed, person)
    if message is None:
        # Not this employee's conversation. Refusing beats filing a ticket
        # in their name that quotes someone else's exchange.
        await _audit_refusal(
            db, person_id=person.id, reason="not_your_message", what="escalate",
            target=message_id, trace_id=trace_id,
        )
        return _refusal(canned.failed(lang), lang)
    # Read off the row BEFORE anything can roll the transaction back: a
    # rollback expires every instance in the session, so re-reading
    # ``message.content`` in the failure branch below would raise
    # ``MissingGreenlet`` from a lazy refresh — inside the very branch whose
    # job is to still show the employee their answer.
    reply_rich = message.content or ""
    docs = message.sources
    service_request_id = message.service_request_id
    #: Same reason, one object over: the failure branches log WHO the
    #: escalation was for, and a rollback expires ``person`` too — reading
    #: ``person.id`` after it raised ``MissingGreenlet`` from the logging call
    #: itself, turning a handled failure into an unhandled one.
    person_id = person.id

    def _kept(note: str) -> dict[str, Any]:
        """The answer, unchanged, with one line about where it went."""
        return answer_card.settled_card(
            reply_rich, lang=lang, docs=docs, note=note,
        )

    # Published before the slow part starts. Ticket creation fans out to every
    # IT admin over Feishu and SMTP, so this branch routinely outruns the 3s
    # callback budget — which is exactly when the generic working card used to
    # land on top of the answer.
    handoff.in_flight = {
        "toast": {"type": "info", "content": _pick(_TOAST_WORKING, lang)},
        "card": {"type": "raw", "data": _kept(_pick(_NOTE_ESCALATING, lang))},
    }

    # Flattened for the ticket body: a support ticket is read in the
    # helpdesk UI and in email, neither of which renders Markdown.
    answer = to_feishu_text(reply_rich)
    asked = (
        await db.execute(
            select(AssistantMessage)
            .where(
                AssistantMessage.conversation_id == message.conversation_id,
                AssistantMessage.seq == max(message.seq - 1, 0),
            ),
        )
    ).scalars().first()
    question = to_feishu_text((asked.content if asked else "") or "")

    title = (question or _pick(_ESCALATED_TITLE, lang))[:120]
    body = "\n\n".join(
        part for part in (
            f"{_pick(_ESCALATED_Q, lang)}\n{question}" if question else "",
            f"{_pick(_ESCALATED_A, lang)}\n{answer}" if answer else "",
        ) if part
    ) or _pick(_ESCALATED_TITLE, lang)
    try:
        ticket = await create_assistant_ticket(
            db,
            requester=person,
            title=title,
            body=body,
            category="other",
            priority="normal",
            service_request_id=service_request_id,
            trace_id=trace_id,
        )
        await db.commit()
    except AppException as exc:
        # Feishu re-posts a callback it judged slow, and this branch is slow by
        # construction. The second post reaches the ticket dedup lock while the
        # first is still filing and is refused with ``SUPPORT_TICKET_IN_FLIGHT``
        # — a submission in PROGRESS, which the blanket handler below reported
        # as 「没能提交」. The employee then taps again, or files by hand, for a
        # ticket that was already on its way. Same defect ``_already_done_card``
        # fixes on the confirm path.
        in_flight = exc.code == _TICKET_IN_FLIGHT_CODE
        await db.rollback()
        if in_flight:
            logger.info("assistant: escalate tap raced its own retry for %s", person_id)
            return {
                "toast": {"type": "info", "content": _pick(_TOAST_ESCALATED, lang)},
                "card": {
                    "type": "raw",
                    "data": _kept(_pick(_NOTE_ESCALATE_IN_FLIGHT, lang)),
                },
                FOLLOWUP_KEY: _closing_card(
                    _pick(_ESCALATED_TITLE, lang),
                    _pick(_NOTE_ESCALATE_IN_FLIGHT, lang),
                    colour="blue",
                    status=card_design.STATUS_DONE,
                ),
            }
        logger.exception("assistant: escalation from card failed for %s", person_id)
        return _escalate_failed(_kept, lang)
    except Exception:
        logger.exception("assistant: escalation from card failed for %s", person_id)
        # A failed flush deactivates the WHOLE transaction, so returning the
        # refusal card without this leaves the caller a session that raises on
        # its next statement — the crash just moves one statement later. A
        # SAVEPOINT is the usual isolation, but ``SupportService.create``
        # commits its own unit of work, so a nested transaction around it
        # closes underneath the context manager; rolling the dead transaction
        # back is the form that fits a service which owns its boundary.
        await db.rollback()
        return _escalate_failed(_kept, lang)

    # ``ticket_number`` is the column; a getattr-with-default here would
    # quietly print an id if the field were ever renamed.
    number = ticket.ticket_number

    note = _pick(_NOTE_ESCALATED, lang).format(number=number)
    followup_body = _with_ticket(_pick(_ESCALATED_BODY, lang), number)
    if with_chat:
        # 工单已经落了,群是加分项。整段 best-effort:失败只是少一个群,员工
        # 不会掉进黑洞 —— 而反过来(建群失败把工单也回滚掉)才是真的坏。
        from app.assistant import handoff as handoff_mod  # noqa: PLC0415

        try:
            outcome = await handoff_mod.hand_off(
                db,
                person=person,
                message_id=parsed,
                question=question,
                answer=answer,
                feishu_message_id=open_message_id,
                lang=lang,
                ticket=ticket,
            )
            await db.commit()
        except Exception:
            logger.exception("assistant: handoff chat failed for %s", person_id)
            await db.rollback()
            outcome = None
        if outcome is not None and outcome.chat_created:
            note = _pick(_NOTE_HANDOFF_CHAT, lang).format(
                owner=outcome.owner_name or "IT", number=number,
            )
            followup_body = _with_ticket(
                _pick(_HANDOFF_BODY_CHAT, lang).format(
                    owner=outcome.owner_name or "IT",
                    label=outcome.category_label or "IT",
                ),
                number,
            )
        else:
            # 没群就说没群。拉不到人时说拉到了,是这条链路最坏的失败。
            note = _pick(_NOTE_HANDOFF_NO_CHAT, lang).format(number=number)
            followup_body = _with_ticket(_pick(_ESCALATED_BODY, lang), number)

    return {
        "toast": {"type": "success", "content": _pick(_TOAST_ESCALATED, lang)},
        "card": {
            "type": "raw",
            "data": _kept(note),
        },
        # The ticket number is the one thing here worth scrolling back TO, so
        # it also goes out as its own message when the tap outran the budget —
        # the in-place card above is the answer, and re-sending that whole
        # answer as a follow-up would post it twice.
        FOLLOWUP_KEY: _closing_card(
            _pick(_ESCALATED_TITLE, lang),
            followup_body,
            colour="blue",
            status=card_design.STATUS_DONE,
        ),
    }


def _escalate_failed(
    kept: Callable[[str], dict[str, Any]], lang: Lang,
) -> dict[str, Any]:
    """A failed 转人工 that still leaves the answer on screen.

    The generic red refusal card would replace it — costing the employee both
    the escalation AND the reply they were reading, for what is usually a
    transient failure they are about to retry.
    """
    return {
        "toast": {"type": "error", "content": _pick(_TOAST_FAILED, lang)},
        "card": {
            "type": "raw",
            "data": kept(_pick(_NOTE_ESCALATE_FAILED, lang)),
        },
        # Every branch whose in-place card is the whole answer needs one of
        # these. Without it a SLOW failure falls back to sending that card as
        # the follow-up — posting the entire answer a second time to tell the
        # employee something went wrong.
        FOLLOWUP_KEY: _closing_card(
            _pick(_TOAST_FAILED, lang),
            _pick(_NOTE_ESCALATE_FAILED, lang),
            colour="red",
        ),
    }


async def _audit_confirmed(
    db: AsyncSession,
    *,
    person: IdentityPerson,
    proposal: Any,
    trace_id: str,
) -> None:
    """Record the press on the same action name the link path uses.

    Best-effort, and SAVEPOINT-isolated rather than merely wrapped in a bare
    ``try``. That distinction is the savepoint guard's whole subject and it is
    right to insist: a failed flush deactivates the WHOLE transaction, so a
    plain swallow leaves the caller holding a session that raises on its next
    statement — here, FastAPI's own teardown.

    An own-session write would also be safe, but it commits somewhere the
    caller cannot see, which makes the audit invisible to anything reading
    through the caller's transaction. The savepoint keeps the row where the
    rest of this request put it, and keeps the session usable either way.
    """
    from app.assistant.admin_service import ACTION_PROPOSAL_CONFIRMED  # noqa: PLC0415
    from app.audit.service import AuditService  # noqa: PLC0415
    from app.core.txn import best_effort  # noqa: PLC0415

    async with best_effort(
        db, label="assistant card confirm audit", log=logger,
    ) as step:
        await AuditService(db).log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=person.id,
            action=ACTION_PROPOSAL_CONFIRMED,
            resource_type="assistant_action_proposal",
            resource_id=str(proposal.id),
            status="success",
            detail={
                "action_key": proposal.action_key,
                "params_hash": proposal.params_hash,
                "via": "feishu_card",
            },
        )
    # Outside the savepoint: committing inside it would close the context
    # manager underneath itself (the same shape ``propose_tools`` documents).
    if step.ok:
        await db.commit()