"""HTTP surface for the IT support / help-desk domain (求助域).

Employee endpoints (``/api/v1/support/*``, any authenticated user — the
service scopes to the caller's own tickets): submit a free-form ticket,
track it, reply, close/reopen/cancel.

IT endpoints (``/api/v1/support/admin/*``): the workbench queue + claim /
assign / reply / resolve / reopen / attributes. Reads admit auditor +
management (``require_admin_read``); mutations require ``global_admin``.
"""

from __future__ import annotations

import logging
import re
import uuid
from datetime import UTC, datetime
from typing import Any

from fastapi import APIRouter, Depends, File, Form, Query, Request, UploadFile
from pydantic import BaseModel, Field
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.audit.service import AuditService
from app.auth.models import AuthUser
from app.core.database import get_db
from app.core.dependencies import (
    check_copilot_rate_limit,
    get_current_user,
    require_admin,
    require_admin_read,
    require_roles,
)
from app.core.exceptions import AppException
from app.core.rbac import AUDITOR_ROLE, GLOBAL_ADMIN_ROLE
from app.core.response import paginated_response, success_response
from app.domains.support.kb_service import SupportKbService
from app.domains.support.models import (
    CATEGORIES,
    CREATION_AGENT_ASSISTED,
    INITIATOR_AGENT,
    INITIATOR_EMPLOYEE,
    KB_PUBLISHED,
    KB_SOURCE_FEISHU_DOC,
    ORIGIN_SUPPORT_COPILOT,
    PRIORITIES,
    SOURCE_WEB,
    SupportDeflectionEvent,
    SupportKbArticle,
    SupportTicket,
)
from app.domains.support.schemas import (
    AssignBody,
    AttributesBody,
    CopilotChatBody,
    CopilotEscalateBody,
    CopilotFeedbackBody,
    CreateTicketBody,
    DeflectBody,
    KbCreateBody,
    KbStatusBody,
    KbUpdateBody,
    RateBody,
    ReplyBody,
    ResolveBody,
)
from app.domains.support.service import SupportService
from app.identity.person_models import IdentityPerson

logger = logging.getLogger(__name__)

router = APIRouter(prefix="/api/v1/support", tags=["support"])


# ── caller identity ─────────────────────────────────────────────────


async def _require_person(db: AsyncSession, person_id: uuid.UUID) -> IdentityPerson:
    """Resolve an identity row that every service entry point below requires.

    ``db.get`` is ``IdentityPerson | None``; the service layer takes a real
    row, so every call site was handing mypy an un-narrowed Optional (the
    ratchet regression that came with the help-desk endpoints). ``AuthUser``
    *is* ``IdentityPerson``, so for an authenticated caller this is a
    same-session identity-map hit and the miss branch is defence in depth —
    a purged row becomes a 404 instead of an ``AttributeError`` five frames
    deeper. Narrow once here rather than at ~20 call sites.
    """
    person = await db.get(IdentityPerson, person_id)
    if person is None:
        raise AppException(code="NOT_FOUND", message="用户不存在", status_code=404)
    return person


# ── serialization ───────────────────────────────────────────────────


async def _people_map(db: AsyncSession, ids: set[uuid.UUID]) -> dict[uuid.UUID, str]:
    ids = {i for i in ids if i is not None}
    if not ids:
        return {}
    rows = (
        await db.execute(
            select(IdentityPerson.id, IdentityPerson.display_name, IdentityPerson.work_email).where(
                IdentityPerson.id.in_(ids)
            )
        )
    ).all()
    return {pid: (name or email or str(pid)) for pid, name, email in rows}


def _iso(dt) -> str | None:
    return dt.isoformat() if dt else None


async def _chat_payload(db: AsyncSession, ticket_id: uuid.UUID) -> dict | None:
    """这张工单的飞书群(如果有)。

    详情页每次打开都会读一次 —— 一次按唯一索引的 SELECT,而它的产物是页面上
    那颗「进群」按钮。没有群时返回 ``None``,页面据此显示「拉个群聊」。
    """
    from app.assistant import ticket_chat  # noqa: PLC0415 — 层间调用

    row = await ticket_chat.get_by_ticket(db, ticket_id)
    if row is None or row.state != "open":
        return None
    return {
        "chat_id": row.feishu_chat_id,
        "url": ticket_chat.chat_applink(row.feishu_chat_id),
        "opened_reason": row.opened_reason,
    }


def _serialize(t: SupportTicket, people: dict[uuid.UUID, str], *, messages=None) -> dict:
    out = {
        "id": str(t.id),
        "ticket_number": t.ticket_number,
        "requester_id": str(t.requester_id),
        "service_request_id": str(t.service_request_id) if t.service_request_id else None,
        "requester_display": people.get(t.requester_id),
        "title": t.title,
        "body": t.body,
        "category": t.category,
        "priority": t.priority,
        "status": t.status,
        "source": t.source,
        "intent_initiator": t.intent_initiator,
        "creation_actor": t.creation_actor,
        "creation_mode": t.creation_mode,
        "origin_channel": t.origin_channel,
        "employee_confirmed_at": _iso(t.employee_confirmed_at),
        "assignee_id": str(t.assignee_id) if t.assignee_id else None,
        "assignee_display": people.get(t.assignee_id) if t.assignee_id else None,
        "resolution": t.resolution,
        "created_at": _iso(t.created_at),
        "updated_at": _iso(t.updated_at),
        "first_response_at": _iso(t.first_response_at),
        "first_response_due_at": _iso(t.first_response_due_at),
        "resolved_at": _iso(t.resolved_at),
        "resolution_due_at": _iso(t.resolution_due_at),
        "sla_paused_at": _iso(t.sla_paused_at),
        "agent_reminder_count": int(t.agent_reminder_count or 0),
        "requester_reminder_count": int(t.requester_reminder_count or 0),
        "closed_at": _iso(t.closed_at),
        # 3-point CSAT (mirror caliber): null until the requester rates.
        "satisfaction_score": t.satisfaction_score,
        "satisfaction_at": _iso(t.satisfaction_at),
    }
    if messages is not None:
        out["messages"] = [
            {
                "id": str(m.id),
                "author_id": str(m.author_id) if m.author_id else None,
                "author_display": people.get(m.author_id) if m.author_id else None,
                "kind": m.kind,
                "body": m.body,
                "created_at": _iso(m.created_at),
            }
            for m in messages
        ]
    return out


async def _serialize_list(db: AsyncSession, tickets: list[SupportTicket]) -> list[dict]:
    ids: set[uuid.UUID] = set()
    for t in tickets:
        ids.add(t.requester_id)
        if t.assignee_id:
            ids.add(t.assignee_id)
    people = await _people_map(db, ids)
    return [_serialize(t, people) for t in tickets]


def _serialize_kb(a: SupportKbArticle, *, include_body: bool = True) -> dict:
    out = {
        "id": str(a.id),
        "title": a.title,
        "category": a.category,
        "status": a.status,
        "source": a.source,
        "source_ticket_id": str(a.source_ticket_id) if a.source_ticket_id else None,
        "created_at": _iso(a.created_at),
        "updated_at": _iso(a.updated_at),
        "published_at": _iso(a.published_at),
        "embedded": a.embedding is not None,
        # Mirrored wiki articles carry the source-doc link (employees open
        # the Feishu doc itself); None for native manual/ai_ticket articles.
        "external_url": a.external_url,
        # Alternate phrasings of the same question (newline-separated) — part
        # of the retrieval text, so the admin editing an article must see them.
        "search_aliases": a.search_aliases or "",
        "language": a.language,
        # The source system's own category label (Feishu-helpdesk FAQ import);
        # ``category`` above is the six-bucket Helix one.
        "source_category": a.source_category,
    }
    if include_body:
        out["body"] = a.body
    return out


def _spawn_kb_embed(article: SupportKbArticle) -> None:
    """Fire-and-forget semantic-index a published article that isn't embedded
    yet — keeps the publish/edit response off the embedding-API latency path.
    No-op (cheap) when the article isn't published or is already embedded; the
    background task is itself idempotent and no-ops when the backend is off."""
    if article.status != KB_PUBLISHED or article.embedding is not None:
        return
    from app.core.background import spawn
    from app.domains.support.kb_service import run_kb_embed

    spawn(run_kb_embed(article.id), name="support.kb_embed")


# ── meta ────────────────────────────────────────────────────────────


@router.get("/meta", summary="Ticket form catalog (categories / priorities)")
async def get_meta(
    request: Request,
    _user: AuthUser = Depends(get_current_user),
):
    return success_response(
        data={"categories": list(CATEGORIES), "priorities": list(PRIORITIES)},
        request=request,
    )


# ── AI copilot (PR3) ────────────────────────────────────────────────
# Grounded IT assistant: answers from the employee's own data via
# RBAC-scoped read tools, escalates to a ticket when it can't resolve.
# Gated on an Anthropic credential (disabled-until-configured).


@router.get("/copilot/status", summary="Is the AI copilot available?")
async def copilot_status(
    request: Request,
    _user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    from app.domains.support.copilot import copilot_available

    return success_response(
        data={"available": await copilot_available(db)},
        request=request,
    )


@router.get("/copilot/starters", summary="Suggested ways to open a conversation")
async def copilot_starters(
    request: Request,
    lang: str | None = None,
    _user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """The same eight categories the Feishu help card offers, for this page.

    **One registry, two surfaces.** ``app.assistant.starters`` is where the
    openers live; serving them rather than duplicating them into the frontend
    bundle is what keeps 「这两个面写的不一样」 from happening on the next edit —
    and the openers are load-bearing copy, not chrome: each one is a promise
    that the assistant can actually answer it.

    ``lang`` is the PAGE's current language, which is not necessarily the
    reader's stored preference (they may be reading the English UI today). An
    unrecognised value degrades to the stored preference rather than 422 —
    a language tag is not something the employee typed.
    """
    from app.assistant import starters
    from app.core.locale import normalize_lang, resolve_lang

    person = await _require_person(db, _user.id)
    picked = normalize_lang(lang) or resolve_lang(person)
    return success_response(
        data={
            "lang": picked,
            "categories": [
                {
                    "key": category.key,
                    "label": category.title(picked),
                    "blurb": category.summary(picked),
                    "starters": [
                        {"key": item.key, "question": item.question(picked)}
                        for item in category.starters
                    ],
                }
                for category in starters.CATEGORIES
            ],
        },
        request=request,
    )


def _deterministic_turn(
    reply: str, *, suggest_escalate: bool = False,
) -> dict[str, Any]:
    """A turn the server answered itself, in the chat response's own shape.

    Same keys as a model turn so the page needs no second branch, and
    ``message_id`` is None because there is nothing here to rate: a canned
    sentence the server wrote is not an answer whose quality tells anyone
    anything about the assistant.
    """
    return {
        "available": True,
        "turn_failed": False,
        "outcome": "answered",
        "busy": False,
        "reply": reply,
        "tool_calls": [],
        "docs": [],
        "offers": [],
        "notices": [],
        "message_id": None,
        "conversation_id": None,
        "suggest_escalate": suggest_escalate,
    }


@router.post("/copilot/chat", summary="Ask the AI IT copilot")
async def copilot_chat(
    body: CopilotChatBody,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    _rl: None = Depends(check_copilot_rate_limit),
):
    """One question to the employee assistant, from the web page.

    Since #3413 this runs the SAME turn as the Feishu bot — same preflight
    grounding, same read bridge, same propose tools, same transcript — instead
    of calling the model kernel bare. See :mod:`app.assistant.web_turn` for what
    that changed and why it was not a second implementation.
    """
    from app.assistant import escalation, reply_language
    from app.assistant import messages as canned
    from app.assistant import session as session_store
    from app.assistant.web_turn import reset_web_thread, run_web_turn, web_chat_id

    person = await _require_person(db, user.id)
    question = body.question()
    if not question:
        raise AppException(
            code="COPILOT_MESSAGE_REQUIRED",
            message="请输入你的问题。",
            status_code=422,
            field="message",
        )

    # 回答一句话的语言,是那句话的语言 —— 与飞书那一侧同一个判定
    # (:mod:`app.assistant.reply_language`),否则同一个人在两个面上问同一句话
    # 会拿到两种语言的答案。下面那三条确定性分支也吃这个值。
    lang = await reply_language.for_turn(
        text=question, person=person, chat_id=web_chat_id(person),
    )

    # ── Deterministic channel controls, before any model hop ──────────────
    #
    # The rule is stated once, in ``session.py``: employees must be able to
    # reset the thread, ask what Helix can do, and reach a human even when the
    # model is unavailable or disagrees. The chat channel has had all three
    # since it shipped; the web page had only the reset, so 「转人工」 and
    # 「/help」 typed into the box were routed to the model like any other
    # sentence — and the second of those is precisely the question a broken
    # model cannot answer.
    if session_store.is_help_command(question):
        return success_response(
            data=_deterministic_turn(canned.help_guide(lang)), request=request,
        )
    if escalation.is_escalation_request(question):
        # NOT a second ticket path. The page has 转人工 right under the
        # composer and it carries the transcript; this says so and lights it
        # up, rather than inventing a parallel way to file the same ticket.
        return success_response(
            data=_deterministic_turn(
                canned.handoff_hint(lang), suggest_escalate=True,
            ),
            request=request,
        )
    # 「新话题」 typed rather than clicked. Deterministic and model-free, exactly
    # as in chat: a reset the employee asked for must never depend on the model
    # being available, or on it agreeing.
    if session_store.is_reset_command(question):
        cleared, said = await reset_web_thread(person=person, lang=lang)
        return success_response(
            data={**_deterministic_turn(said), "reset": cleared}, request=request,
        )

    turn = await run_web_turn(
        db,
        person=person,
        question=question,
        trace_id=str(getattr(request.state, "trace_id", "")),
        lang=lang,
    )
    # Not an error and not a failed turn: another tab is mid-answer. The page
    # keeps its composer and says so, rather than showing a red toast for a
    # request that was correctly refused.
    if turn.busy:
        return success_response(
            data={
                "available": True,
                "turn_failed": False,
                "busy": True,
                "reply": turn.reply,
                "tool_calls": [],
                "docs": [],
                "offers": [],
                "notices": [],
                "message_id": None,
                "conversation_id": None,
                "suggest_escalate": False,
            },
            request=request,
        )
    logger.info(
        "assistant: web turn person=%s outcome=%s tools=%s offers=%d chat=%s",
        person.id,
        turn.outcome,
        ",".join(turn.tool_calls) or "-",
        len(turn.offers),
        web_chat_id(person),
    )
    return success_response(
        data={
            "available": turn.available,
            # Separate from ``available`` on purpose: the panel gates its
            # composer on the capability, and must not disable it because one
            # turn hit a provider error.
            "turn_failed": turn.turn_failed,
            # The server's own verdict on this turn, so the page does not have
            # to re-derive it from two booleans. ``failed`` and ``unavailable``
            # both mean "there is nothing here worth rating".
            "outcome": turn.outcome,
            "busy": False,
            "reply": turn.reply,
            "tool_calls": turn.tool_calls,
            "docs": turn.docs,
            "offers": turn.offers,
            "notices": turn.notices,
            "message_id": turn.message_id,
            "conversation_id": turn.conversation_id,
            "suggest_escalate": turn.suggest_escalate,
        },
        request=request,
    )


@router.get("/copilot/thread", summary="What the assistant still remembers")
async def copilot_thread(
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """The current web conversation, so the page can show what the server holds.

    The page used to start empty on every mount while the server kept the
    thread for two hours. That is not cosmetic: the employee comes back, sees a
    blank box, types 「帮我提交」, and the assistant answers from a context that
    is nowhere on screen — with 「新话题」 and 转人工 both hidden because the page
    believed there was no conversation. Pending offers come back too, for the
    same reason: an unconfirmed proposal is live for an hour, and a page that
    forgets it leaves an offer the employee can neither act on nor cancel.
    """
    from app.assistant.web_turn import web_thread

    person = await _require_person(db, user.id)
    return success_response(data=await web_thread(db, person=person), request=request)


@router.post("/copilot/reset", summary="Start a new assistant topic")
async def copilot_reset(
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Forget the working memory for this employee's web thread.

    Reports whether it actually stuck. A Redis failure that is reported as
    success is the worst outcome here: the employee is told the thread is gone
    and the model keeps every word of it.
    """
    from app.assistant import reply_language
    from app.assistant.web_turn import reset_web_thread, web_chat_id

    person = await _require_person(db, user.id)
    # 用这段对话的语言说这句话。按钮上没有文字可判,所以走 ``for_card`` ——
    # 一段英文对话点「新话题」不该回一句中文。
    lang = await reply_language.for_card(
        person=person, chat_id=web_chat_id(person),
    )
    cleared, said = await reset_web_thread(person=person, lang=lang)
    return success_response(data={"reset": cleared, "message": said}, request=request)


@router.post("/copilot/feedback", summary="Rate one assistant answer")
async def copilot_feedback(
    body: CopilotFeedbackBody,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """👍 / 👎 on a stored answer, with an optional structured reason.

    The same record the Feishu card writes, through the same module — a verdict
    is a work item in the knowledge-gap queue, and a queue that only sees one of
    two surfaces is a queue that mis-ranks what to fix first.

    ⚠️ **A reason attaches to an existing verdict; it does not re-cast one.**
    ``record_verdict``'s upsert deliberately resets ``gap_status`` to ``open``
    and clears the reason columns, because tapping the other button is a change
    of mind. Running it again on the reason click — which the page sends right
    after the 👎 — reopened a gap an administrator had already claimed or
    resolved, left ``resolved_at`` and the resolution note dangling on an
    ``open`` row, and re-paged the reminder job. The Feishu twin calls
    ``record_reason`` alone for the same reason.
    """
    import uuid as _uuid

    from app.assistant import feedback as feedback_store

    person = await _require_person(db, user.id)
    try:
        message_id = _uuid.UUID(body.message_id)
    except (ValueError, AttributeError, TypeError):
        raise AppException(
            code="COPILOT_MESSAGE_ID_INVALID",
            message="This answer id is not usable.",
            status_code=422,
            field="message_id",
        ) from None

    # Ownership first, once. Deliberately 404, not 403: "that is not your
    # message" and "there is no such message" are the same fact to a caller who
    # may know neither, and telling them apart would confirm a stranger's id.
    if await feedback_store.own_message(db, message_id, person) is None:
        raise AppException(
            code="COPILOT_ANSWER_NOT_FOUND",
            message="No such assistant answer.",
            status_code=404,
        )

    trace_id = str(getattr(request.state, "trace_id", ""))
    wants_reason = bool(body.reason) and not body.helpful
    if wants_reason and body.reason not in feedback_store.REASON_CLASSIFICATIONS:
        raise AppException(
            code="COPILOT_REASON_UNKNOWN",
            message="Unknown feedback reason.",
            status_code=422,
            field="reason",
        )

    reason_stored = False
    if wants_reason:
        stored = await feedback_store.record_reason(
            db, person=person, message_id=message_id,
            reason=str(body.reason), trace_id=trace_id,
        )
        if stored is None:
            # No negative verdict to attach to — the page skipped the 👎, or a
            # positive one is on record. Cast it, then attach.
            await feedback_store.record_verdict(
                db, person=person, message_id=message_id,
                helpful=False, trace_id=trace_id,
            )
            stored = await feedback_store.record_reason(
                db, person=person, message_id=message_id,
                reason=str(body.reason), trace_id=trace_id,
            )
        reason_stored = stored is not None
    else:
        await feedback_store.record_verdict(
            db, person=person, message_id=message_id,
            helpful=body.helpful, trace_id=trace_id,
        )
    return success_response(
        data={"recorded": True, "reason_recorded": reason_stored}, request=request,
    )


@router.post("/copilot/escalate", status_code=201, summary="Escalate a copilot chat to a ticket")
async def copilot_escalate(
    body: CopilotEscalateBody,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    _rl: None = Depends(check_copilot_rate_limit),
):
    from app.assistant.web_turn import web_transcript
    from app.domains.support.copilot import draft_ticket_from_history

    person = await _require_person(db, user.id)
    # The stored transcript first: it is what the assistant actually said. The
    # page's own copy is the fallback for the only case that has no transcript —
    # the copilot being switched off, where the employee typed into a disabled
    # composer's escape hatch and no turn ever ran.
    import uuid as _uuid

    try:
        thread = _uuid.UUID(body.conversation_id) if body.conversation_id else None
    except (ValueError, AttributeError, TypeError):
        # A malformed id is a page bug, not an attack, and the honest recovery
        # is the most recent thread rather than a 422 over a field the employee
        # never typed.
        thread = None
    # Three sources, narrowest first, and the ORDER is the security property.
    history: list[dict[str, str]] = []
    if thread is not None:
        history = await web_transcript(db, person=person, conversation_id=thread)
    if not history:
        # The page named a thread that is not this person's — a stale tab, or a
        # borrowed id. Their own most recent thread is both the honest answer
        # and an unforgeable one.
        history = await web_transcript(db, person=person)
    if not history:
        # ⚠️ ONLY reached when this person has NO web transcript at all, which
        # is the case this fallback was written for: the copilot is switched
        # off, so no turn ever ran. Gating it any wider was a one-line way to
        # put a FABRICATED assistant conversation into a ticket an IT admin then
        # acts on — ``web_transcript`` returns [] for any id that is not the
        # caller's own, so "send a random uuid4" used to select the client's
        # array over the real transcript.
        history = [{"role": m.role, "content": m.content} for m in body.messages]
    title, ticket_body = draft_ticket_from_history(history)
    svc = SupportService(db)
    ticket = await svc.create(
        requester=person,
        title=title,
        body=ticket_body,
        category=body.category,
        priority=body.priority,
        source="ai",
        intent_initiator=INITIATOR_EMPLOYEE,
        creation_actor=INITIATOR_AGENT,
        creation_mode=CREATION_AGENT_ASSISTED,
        origin_channel=ORIGIN_SUPPORT_COPILOT,
        employee_confirmed_at=datetime.now(UTC),
        trace_id=str(getattr(request.state, "trace_id", "")),
    )
    await db.commit()
    people = await _people_map(db, {ticket.requester_id})
    return success_response(data=_serialize(ticket, people), request=request)


@router.get("/kb", summary="Browse / search published knowledge-base articles")
async def browse_kb(
    request: Request,
    _user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    q: str | None = Query(default=None, max_length=100),
    category: str | None = Query(default=None),
    page: int = Query(default=1, ge=1),
    page_size: int = Query(default=50, ge=1, le=100),
):
    """The employee knowledge base: every published article, whatever its
    source (manual, AI-drafted, mirrored Feishu wiki doc, imported helpdesk
    FAQ). With a query the result is RANKED by CJK-aware shingle similarity
    against each article's question forms (title + alternate phrasings) — see
    ``SupportKbService.browse_published``."""
    svc = SupportKbService(db)
    rows, total = await svc.browse_published(
        q=q, category=category, page=page, page_size=page_size,
    )
    return paginated_response(
        data=[_serialize_kb(a) for a in rows],
        total=total,
        page=page,
        page_size=page_size,
        request=request,
    )


# ── employee surface ────────────────────────────────────────────────


@router.post("/tickets", status_code=201, summary="Submit a help ticket")
async def create_ticket(
    body: CreateTicketBody,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    # Same bucket as reply/deflect: every create fans out a card+email to
    # EVERY global_admin, so a scripted title-rotating loop (the dedup is
    # per-title) floods the whole IT roster without this.
    _rl: None = Depends(check_copilot_rate_limit),
):
    person = await _require_person(db, user.id)
    svc = SupportService(db)
    deflection_event = None
    if body.deflect_event_id:
        try:
            candidate = await db.get(SupportDeflectionEvent, uuid.UUID(body.deflect_event_id))
            if candidate is not None and candidate.person_id == person.id:
                deflection_event = candidate
        except (ValueError, TypeError):
            pass
    ticket = await svc.create(
        requester=person,
        title=body.title,
        body=body.body,
        category=body.category,
        priority=body.priority,
        source=SOURCE_WEB,
        service_request_id=(deflection_event.service_request_id if deflection_event else None),
        intent_initiator=INITIATOR_EMPLOYEE,
        creation_actor=(INITIATOR_AGENT if deflection_event else INITIATOR_EMPLOYEE),
        creation_mode=(CREATION_AGENT_ASSISTED if deflection_event else "employee_direct"),
        origin_channel=(ORIGIN_SUPPORT_COPILOT if deflection_event else "web"),
        employee_confirmed_at=(datetime.now(UTC) if deflection_event else None),
        trace_id=str(getattr(request.state, "trace_id", "")),
    )
    # Record the escalation against the deflection session, if filed from one.
    if body.deflect_event_id:
        try:
            await svc.link_escalation(
                event_id=uuid.UUID(body.deflect_event_id),
                person=person,
                ticket=ticket,
            )
        except (ValueError, TypeError):
            pass  # bad id — never block ticket creation on funnel instrumentation
    await db.commit()
    # AI auto-triage runs fire-and-forget on its OWN session so the requester's
    # create response is never blocked on LLM latency; it fills category/priority
    # defaults only and no-ops when the copilot is disabled.
    from app.core.background import spawn
    from app.domains.support.copilot import run_ticket_triage

    spawn(run_ticket_triage(ticket.id), name="support.triage")
    people = await _people_map(db, {ticket.requester_id})
    return success_response(data=_serialize(ticket, people), request=request)


@router.post("/deflect", summary="Pre-file self-service: search KB + AI before filing")
async def deflect(
    body: DeflectBody,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    _rl: None = Depends(check_copilot_rate_limit),
):
    """Before filing a ticket, try to solve the problem: returns matching KB
    articles + an optional grounded AI answer, and opens a deflection event so
    the funnel can measure real deflection. Never fails the user — empty results
    just mean "go ahead and file"."""
    person = await _require_person(db, user.id)
    event, articles, answer = await SupportService(db).start_deflection(
        person=person,
        title=body.title,
        body=body.body,
    )
    await db.commit()
    return success_response(
        data={
            "event_id": str(event.id),
            "articles": [_serialize_kb(a, include_body=True) for a in articles],
            "ai_answer": answer,
            "ai_available": answer is not None,
        },
        request=request,
    )


@router.post(
    "/deflect/{event_id}/resolved", summary="Mark a deflection as solved (no ticket filed)"
)
async def deflect_resolved(
    event_id: str,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    person = await _require_person(db, user.id)
    try:
        eid = uuid.UUID(event_id)
    except (ValueError, TypeError):
        raise AppException(code="NOT_FOUND", message="记录不存在", status_code=404) from None
    event = await db.get(SupportDeflectionEvent, eid)
    if event is None:
        raise AppException(code="NOT_FOUND", message="记录不存在", status_code=404)
    await SupportService(db).mark_deflected(event=event, person=person)
    await db.commit()
    return success_response(data={"outcome": event.outcome}, request=request)


@router.get("/tickets", summary="My submitted tickets")
async def list_my_tickets(
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    status: str | None = Query(default=None),
    page: int = Query(default=1, ge=1),
    page_size: int = Query(default=20, ge=1, le=100),
):
    svc = SupportService(db)
    rows, total = await svc.list_for_requester(
        person_ids=[user.id],
        status=status,
        page=page,
        page_size=page_size,
    )
    return paginated_response(
        data=await _serialize_list(db, rows),
        total=total,
        page=page,
        page_size=page_size,
        request=request,
    )


async def _load_for_caller(
    db: AsyncSession,
    ticket_id: str,
    user: AuthUser,
    *,
    require_owner: bool,
) -> SupportTicket:
    try:
        tid = uuid.UUID(ticket_id)
    except ValueError as e:
        raise AppException(code="NOT_FOUND", message="工单不存在", status_code=404) from e
    svc = SupportService(db)
    ticket = await svc.get(tid)
    if ticket is None:
        raise AppException(code="NOT_FOUND", message="工单不存在", status_code=404)
    is_admin = (
        "global_admin" in (user.role_names or [])
        or "auditor" in (user.role_names or [])
        or "management" in (user.role_names or [])
    )
    # 这张工单的处理人也读得到它。2026-09-08:「转人工」把工单指派给路由负责人,
    # 而那个人常常不是 global_admin —— 群里那张卡上的「在 Helix 里打开」对他会
    # 直接 403。**能动手就该看得见**;反过来(能看不能动)才是安全的方向。
    is_assignee = ticket.assignee_id is not None and ticket.assignee_id == user.id
    if (
        require_owner
        and ticket.requester_id != user.id
        and not is_admin
        and not is_assignee
    ):
        raise AppException(code="FORBIDDEN", message="无权访问该工单", status_code=403)
    return ticket


@router.get("/tickets/{ticket_id}", summary="Ticket detail + timeline")
async def get_ticket(
    ticket_id: str,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    ticket = await _load_for_caller(db, ticket_id, user, require_owner=True)
    svc = SupportService(db)
    messages = await svc.list_messages(ticket.id)
    ids = {ticket.requester_id}
    if ticket.assignee_id:
        ids.add(ticket.assignee_id)
    ids.update(m.author_id for m in messages if m.author_id)
    people = await _people_map(db, ids)
    payload = _serialize(ticket, people, messages=messages)
    payload["feishu_chat"] = await _chat_payload(db, ticket.id)
    return success_response(data=payload, request=request)


@router.post("/tickets/{ticket_id}/chat", summary="Open a Feishu group for this ticket")
async def open_ticket_chat_endpoint(
    ticket_id: str,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    # 同 create_ticket 的桶:建群是不可撤销的外部动作,而这个端点在页面上是
    # 一颗按钮 —— 连点十次不该产生十次尝试。
    _rl: None = Depends(check_copilot_rate_limit),
):
    """把这张工单变成一个飞书群:求助人 + 现在负责它的人。

    ## 为什么求助人也能开

    「转人工」是助手答不上来时的路径,而**大多数工单不是从助手来的**。一个员工
    自己提交的工单,如果只能等 IT 在后台回复,那么「把人拉进来聊两句就完了」这
    条快路径对他不存在 —— 而那正是这套东西的价值所在。

    已经有群就返回同一个群(幂等):建群不可撤销,两个讲同一件事的房间会让参与
    的人各说各话。
    """
    from app.core.coordinators.ticket_chat_coordinator import (  # noqa: PLC0415
        open_ticket_chat,
        refusal_for,
    )
    from app.domains.support.service import (  # noqa: PLC0415
        can_act_on_ticket,
        role_names_for_person,
    )

    ticket = await _load_for_caller(db, ticket_id, user, require_owner=True)
    # 🔴 ``_load_for_caller`` 的 ``is_admin`` 里含 **auditor 与 management** ——
    # 两个**只读**角色。光靠它放行,等于让一个只读角色建一个飞书群、并把工单
    # 指派给别人(开群会 assign)。所以这里再问一次「你能动这张工单吗」:
    # 求助人本人,或者 ``can_act_on_ticket``(global_admin / 处理人)。
    #
    # 这条判据与 MCP 的 ``open_ticket_chat`` 逐字相同 —— 两个面对「谁能为这张
    # 工单拉群」给出不同答案,是这类改动最容易留下的洞,而这一次就留下了:
    # MCP 那一侧从一开始就检查了,HTTP 这一侧没有(合并后复审找到)。
    roles = await role_names_for_person(db, user.id)
    if ticket.requester_id != user.id and not can_act_on_ticket(
        ticket=ticket, person_id=user.id, role_names=roles,
    ):
        raise AppException(
            code="FORBIDDEN",
            message="只有求助人、这张工单的处理人或 IT 管理员可以拉群",
            status_code=403,
        )
    actor = await _require_person(db, user.id)
    outcome = await open_ticket_chat(
        db, ticket=ticket, actor=actor,
        reason="requester" if ticket.requester_id == user.id else "agent",
    )
    if not outcome.ok:
        await db.rollback()
        # 码 / 文案 / 状态都来自 coordinator 的那一张表 —— MCP 的孪生工具读的
        # 是同一张,所以两个面对同一次拒绝永远说同一句话。
        code, message, status = refusal_for(outcome.reason)
        raise AppException(code=code, message=message, status_code=status)
    await db.commit()
    return success_response(
        data={
            "chat_id": outcome.chat_id,
            "url": outcome.applink,
            "created": outcome.created,
        },
        request=request,
    )


@router.post(
    "/tickets/{ticket_id}/chat/disband",
    summary="Disband this ticket's Feishu group",
)
async def disband_ticket_chat_endpoint(
    ticket_id: str,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """解散这张工单的飞书群。**不可逆:聊天记录随它一起消失。**

    三道闸,顺序刻意:

    1. **工单必须已经结了**(closed / cancelled)。对一件还没有结论的事,
       任何人都不该解散它的房间 —— 所以这一条在授权之前判。
    2. **求助人,或这张工单的处理人**(管理员亦可)。
    3. 解散前确保工单上已经有一段过程记录 —— 这一条由群里的命令和每日收尾任务
       各自负责;从这里手动解散时,记录通常已经在了(结案说明)。

    平时不需要用它:每日收尾任务会按保留期自动清。它存在的理由是「现在就想清掉」
    和「自动清理失败了」这两种情况。
    """
    from app.assistant import ticket_chat as ticket_chat_mod  # noqa: PLC0415
    from app.core.locale import resolve_lang  # noqa: PLC0415
    from app.domains.support.service import (  # noqa: PLC0415
        can_act_on_ticket,
        role_names_for_person,
    )

    ticket = await _load_for_caller(db, ticket_id, user, require_owner=True)
    if ticket.status not in ("closed", "cancelled"):
        raise AppException(
            code="INVALID_STATE",
            message="工单还没有结束,群里的讨论还有用",
            status_code=409,
        )
    roles = await role_names_for_person(db, user.id)
    if ticket.requester_id != user.id and not can_act_on_ticket(
        ticket=ticket, person_id=user.id, role_names=roles,
    ):
        raise AppException(
            code="FORBIDDEN",
            message="只有求助人或这张工单的处理人可以解散群",
            status_code=403,
        )
    chat = await ticket_chat_mod.get_by_ticket(db, ticket.id)
    if chat is None or chat.state != "open":
        # 已经没了 —— 幂等地成功。一颗按了报错的「解散」按钮,在群已经没了的
        # 时候读起来像平台坏了。
        return success_response(data={"disbanded": False}, request=request)
    requester = await db.get(IdentityPerson, ticket.requester_id)
    ok = await ticket_chat_mod.disband(
        db, chat=chat, ticket=ticket, reason="manual",
        lang=resolve_lang(requester) if requester else "zh",
        actor_person_id=user.id,
        trace_id=str(getattr(request.state, "trace_id", "")),
    )
    if not ok:
        await db.rollback()
        raise AppException(
            code="FEISHU_API_ERROR",
            message="暂时没能解散这个群,每日收尾任务会再试一次",
            status_code=502,
        )
    await db.commit()
    return success_response(data={"disbanded": True}, request=request)


#: 开群失败的三种原因,各自对应一句给人看的话。「还没有人认领」和「飞书那边
#: 出问题了」是完全不同的下一步,合并成一句「建群失败」会把人送去问错的人。
@router.post("/tickets/{ticket_id}/reply", summary="Requester reply")
async def requester_reply(
    ticket_id: str,
    body: ReplyBody,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    # Rate-limit requester replies — each reply to a still-unassigned ticket
    # fans out a Feishu card to EVERY global_admin via _notify_it, so a scripted
    # reply loop on one's own ticket floods the whole IT roster. Same bucket as
    # create_ticket / the copilot mutations.
    _rl: None = Depends(check_copilot_rate_limit),
):
    ticket = await _load_for_caller(db, ticket_id, user, require_owner=True)
    if ticket.requester_id != user.id:
        raise AppException(code="FORBIDDEN", message="仅求助人可在此回复", status_code=403)
    person = await _require_person(db, user.id)
    svc = SupportService(db)
    await svc.add_reply(
        ticket=ticket,
        author=person,
        body=body.body,
        is_agent=False,
        trace_id=str(getattr(request.state, "trace_id", "")),
    )
    await db.commit()
    return success_response(data={"status": ticket.status}, request=request)


@router.post("/tickets/{ticket_id}/cancel", summary="Requester withdraw ticket")
async def cancel_ticket(
    ticket_id: str,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    ticket = await _load_for_caller(db, ticket_id, user, require_owner=True)
    if ticket.requester_id != user.id:
        raise AppException(code="FORBIDDEN", message="仅求助人可撤回", status_code=403)
    person = await _require_person(db, user.id)
    svc = SupportService(db)
    await svc.cancel(
        ticket=ticket, requester=person, trace_id=str(getattr(request.state, "trace_id", ""))
    )
    await db.commit()
    return success_response(data={"status": ticket.status}, request=request)


@router.post("/tickets/{ticket_id}/close", summary="Requester confirm-close")
async def close_ticket(
    ticket_id: str,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    ticket = await _load_for_caller(db, ticket_id, user, require_owner=True)
    if ticket.requester_id != user.id:
        raise AppException(code="FORBIDDEN", message="仅求助人可关闭", status_code=403)
    person = await _require_person(db, user.id)
    svc = SupportService(db)
    await svc.close(
        ticket=ticket,
        actor=person,
        is_agent=False,
        trace_id=str(getattr(request.state, "trace_id", "")),
    )
    await db.commit()
    return success_response(data={"status": ticket.status}, request=request)


# ── No-login CSAT magic link (from the resolved notification card) ──
# GET is side-effect-free (mail-client prefetch safe) and shows the three
# rating buttons; POST consumes the action token's jti exactly once.
# Mirrors provisioning's quick-complete pattern.


@router.get("/rate", include_in_schema=False)
async def rate_landing(
    token: str = Query(...),
    db: AsyncSession = Depends(get_db),
):
    from app.core.security import decode_action_token
    from app.domains.support.models import STATUS_CLOSED, STATUS_RESOLVED
    from app.domains.support.rating_pages import (
        rate_already_page,
        rate_choice_form,
        rate_invalid_page,
        rate_page,
    )

    try:
        payload = decode_action_token(token, expected_purpose="support_ticket_rate")
        ticket_id = uuid.UUID(str(payload["res"]))
    except Exception:  # noqa: BLE001 — any bad token renders the same page
        return rate_invalid_page()
    ticket = await db.get(SupportTicket, ticket_id)
    if ticket is None or str(ticket.requester_id) != str(payload.get("sub")):
        return rate_invalid_page()
    try:
        # Departed / force-logged-out requesters lose the link at the GET
        # too — the landing page shows ticket number/title/resolution
        # excerpt, and force-logout must sever every credential surface.
        from app.core.dependencies import enforce_action_token_subject

        await enforce_action_token_subject(db, payload)
    except Exception:  # noqa: BLE001
        return rate_invalid_page()
    if ticket.satisfaction_score is not None:
        return rate_already_page()
    if ticket.status not in (STATUS_RESOLVED, STATUS_CLOSED):
        return rate_page(
            title="暂不能评价",
            heading="工单还在处理中",
            tone="info",
            body_html=(
                "<p>这张工单已重新打开或仍在处理，解决后即可评价。</p>"
                "<p class='en'>This ticket was reopened / is still being "
                "worked on — you can rate it once it is resolved.</p>"
            ),
        )
    label = f"{ticket.ticket_number} · {ticket.title}"
    hint = (ticket.resolution or "").strip()[:80]
    return rate_page(
        title="评价 IT 服务",
        heading="评价 IT 服务",
        tone="info",
        icon="⭐",
        body_html=rate_choice_form(token, label, hint),
    )


@router.post("/rate", include_in_schema=False)
async def rate_submit(
    request: Request,
    token: str = Form(...),
    score: int = Form(...),
    db: AsyncSession = Depends(get_db),
):
    from app.core.dependencies import enforce_action_token_subject
    from app.core.security import (
        consume_action_jti,
        decode_action_token,
        release_action_jti,
    )
    from app.domains.support.rating_pages import (
        RATE_CHOICES,
        rate_already_page,
        rate_done_page,
        rate_invalid_page,
    )

    try:
        payload = decode_action_token(token, expected_purpose="support_ticket_rate")
        ticket_id = uuid.UUID(str(payload["res"]))
    except Exception:  # noqa: BLE001
        return rate_invalid_page()
    ticket = await db.get(SupportTicket, ticket_id)
    if ticket is None or str(ticket.requester_id) != str(payload.get("sub")):
        return rate_invalid_page()
    try:
        # Departed / force-logged-out requesters lose the link too.
        await enforce_action_token_subject(db, payload)
    except Exception:  # noqa: BLE001
        return rate_invalid_page()
    if ticket.satisfaction_score is not None:
        return rate_already_page()

    # Single-use gate AFTER validation: first click rates, replays get the
    # idempotent "already rated" page.
    if not await consume_action_jti(payload):
        return rate_already_page()
    person = await db.get(IdentityPerson, ticket.requester_id)
    if person is None:
        # Requester row is gone (hard delete) — nobody left to attribute the
        # rating to. Give the single-use jti back so the link isn't burned.
        await release_action_jti(payload)
        return rate_invalid_page()
    try:
        await SupportService(db).rate_ticket(
            ticket=ticket,
            actor=person,
            score=int(score),
            trace_id=str(getattr(request.state, "trace_id", "")),
            via="token",
        )
        await db.commit()
    except AppException as exc:
        if exc.code == "ALREADY_RATED":
            return rate_already_page()
        # Transient/state failure — give the link back so a retry works.
        await release_action_jti(payload)
        return rate_invalid_page()
    label = next((zh for s, _e, zh, _en in RATE_CHOICES if s == int(score)), str(score))
    return rate_done_page(label)


@router.post("/tickets/{ticket_id}/rate", summary="Requester CSAT rating (3-point)")
async def rate_ticket(
    ticket_id: str,
    body: RateBody,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    ticket = await _load_for_caller(db, ticket_id, user, require_owner=True)
    person = await _require_person(db, user.id)
    svc = SupportService(db)
    await svc.rate_ticket(
        ticket=ticket,
        actor=person,
        score=body.score,
        trace_id=str(getattr(request.state, "trace_id", "")),
        via="web",
    )
    await db.commit()
    return success_response(
        data={"satisfaction_score": ticket.satisfaction_score},
        request=request,
    )


@router.post("/tickets/{ticket_id}/reopen", summary="Requester reopen a resolved ticket")
async def requester_reopen(
    ticket_id: str,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    ticket = await _load_for_caller(db, ticket_id, user, require_owner=True)
    if ticket.requester_id != user.id:
        raise AppException(code="FORBIDDEN", message="仅求助人可重新打开", status_code=403)
    person = await _require_person(db, user.id)
    svc = SupportService(db)
    await svc.reopen(
        ticket=ticket,
        actor=person,
        is_agent=False,
        trace_id=str(getattr(request.state, "trace_id", "")),
    )
    await db.commit()
    return success_response(data={"status": ticket.status}, request=request)


# ── IT workbench surface ────────────────────────────────────────────

admin_router = APIRouter(prefix="/api/v1/support/admin", tags=["support-admin"])


@admin_router.get("/queue", summary="IT support queue")
async def list_queue(
    request: Request,
    _user: AuthUser = Depends(require_admin_read()),
    db: AsyncSession = Depends(get_db),
    status: str | None = Query(default=None),
    assignee_id: str | None = Query(default=None),
    category: str | None = Query(default=None),
    priority: str | None = Query(default=None),
    q: str | None = Query(default=None, max_length=100),
    page: int = Query(default=1, ge=1),
    page_size: int = Query(default=20, ge=1, le=100),
):
    svc = SupportService(db)
    assignee_uuid = None
    if assignee_id:
        try:
            assignee_uuid = uuid.UUID(assignee_id)
        except ValueError:
            assignee_uuid = None
    rows, total = await svc.list_queue(
        status=status,
        assignee_id=assignee_uuid,
        category=category,
        priority=priority,
        q=q,
        page=page,
        page_size=page_size,
    )
    return paginated_response(
        data=await _serialize_list(db, rows),
        total=total,
        page=page,
        page_size=page_size,
        request=request,
    )


@admin_router.get("/stats", summary="IT queue per-status counts")
async def queue_stats(
    request: Request,
    _user: AuthUser = Depends(require_admin_read()),
    db: AsyncSession = Depends(get_db),
):
    svc = SupportService(db)
    return success_response(data=await svc.queue_stats(), request=request)


async def _require_agent(
    db: AsyncSession, user: AuthUser, ticket: SupportTicket,
) -> IdentityPerson:
    """这张工单的处理人,或者 IT 管理员。返回执行者的 person 行。

    ## 为什么这些端点的门从「global_admin」放宽到了这里

    「转人工」会把工单**指派**给路由负责人,而那个人往往不是 `global_admin`:
    他在飞书群里正处理这件事,却对自己名下的这张工单没有任何写权限 —— 群里
    那些按钮全都是按了就被拒。「把人拉进房间」和「把活交给他」必须是同一件事。

    判据在支持域里只有一份(`service.can_act_on_ticket`),飞书群里的按钮走的
    是同一个函数 —— 两个面对「谁能动这张工单」给出不同答案,是这类改动最容易
    留下的洞。

    ⚠️ 读面**没有**跟着放宽:队列列表、统计仍然是 `require_admin_read`。放宽
    的是「对我名下的这一张工单能不能动手」,不是「能不能看见所有工单」。
    """
    from app.domains.support.service import (  # noqa: PLC0415
        can_act_on_ticket,
        role_names_for_person,
    )

    person = await _require_person(db, user.id)
    roles = await role_names_for_person(db, user.id)
    if not can_act_on_ticket(
        ticket=ticket, person_id=person.id, role_names=roles,
    ):
        raise AppException(
            code="FORBIDDEN",
            message="只有 IT 管理员或这张工单的处理人可以操作它",
            status_code=403,
        )
    return person


async def _load_admin(db: AsyncSession, ticket_id: str) -> SupportTicket:
    try:
        tid = uuid.UUID(ticket_id)
    except ValueError as e:
        raise AppException(code="NOT_FOUND", message="工单不存在", status_code=404) from e
    # All admin callers of ``_load_admin`` are state-changing POSTs, so take
    # a row lock to serialize concurrent transitions on the same ticket.
    ticket = await SupportService(db).get(tid, for_update=True)
    if ticket is None:
        raise AppException(code="NOT_FOUND", message="工单不存在", status_code=404)
    return ticket


@admin_router.post("/tickets/{ticket_id}/claim", summary="Claim a ticket")
async def claim_ticket(
    ticket_id: str,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    ticket = await _load_admin(db, ticket_id)
    person = await _require_agent(db, user, ticket)
    await SupportService(db).claim(
        ticket=ticket, agent=person, trace_id=str(getattr(request.state, "trace_id", ""))
    )
    await db.commit()
    return success_response(data={"status": ticket.status}, request=request)


@admin_router.post("/tickets/{ticket_id}/assign", summary="Assign a ticket")
async def assign_ticket(
    ticket_id: str,
    body: AssignBody,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    ticket = await _load_admin(db, ticket_id)
    actor = await _require_agent(db, user, ticket)
    try:
        assignee = await db.get(IdentityPerson, uuid.UUID(body.assignee_id))
    except ValueError as e:
        raise AppException(code="VALIDATION_ERROR", message="无效的指派人", status_code=422) from e
    if assignee is None:
        raise AppException(code="NOT_FOUND", message="指派人不存在", status_code=404)
    await SupportService(db).assign(
        ticket=ticket,
        assignee=assignee,
        actor=actor,
        trace_id=str(getattr(request.state, "trace_id", "")),
    )
    await db.commit()
    return success_response(data={"status": ticket.status}, request=request)


@admin_router.post("/tickets/{ticket_id}/attributes", summary="Set category / priority")
async def set_attributes(
    ticket_id: str,
    body: AttributesBody,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    ticket = await _load_admin(db, ticket_id)
    actor = await _require_agent(db, user, ticket)
    await SupportService(db).set_attributes(
        ticket=ticket,
        actor=actor,
        category=body.category,
        priority=body.priority,
        trace_id=str(getattr(request.state, "trace_id", "")),
    )
    await db.commit()
    return success_response(
        data={"category": ticket.category, "priority": ticket.priority},
        request=request,
    )


@admin_router.post("/tickets/{ticket_id}/reply", summary="Agent reply")
async def agent_reply(
    ticket_id: str,
    body: ReplyBody,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    ticket = await _load_admin(db, ticket_id)
    agent = await _require_agent(db, user, ticket)
    await SupportService(db).add_reply(
        ticket=ticket,
        author=agent,
        body=body.body,
        is_agent=True,
        trace_id=str(getattr(request.state, "trace_id", "")),
    )
    await db.commit()
    return success_response(data={"status": ticket.status}, request=request)


@admin_router.post("/tickets/{ticket_id}/resolve", summary="Resolve a ticket")
async def resolve_ticket(
    ticket_id: str,
    body: ResolveBody,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    ticket = await _load_admin(db, ticket_id)
    agent = await _require_agent(db, user, ticket)
    await SupportService(db).resolve(
        ticket=ticket,
        agent=agent,
        resolution=body.resolution,
        trace_id=str(getattr(request.state, "trace_id", "")),
    )
    await db.commit()
    return success_response(data={"status": ticket.status}, request=request)


@admin_router.post("/tickets/{ticket_id}/reopen", summary="Agent reopen")
async def agent_reopen(
    ticket_id: str,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    ticket = await _load_admin(db, ticket_id)
    agent = await _require_agent(db, user, ticket)
    await SupportService(db).reopen(
        ticket=ticket,
        actor=agent,
        is_agent=True,
        trace_id=str(getattr(request.state, "trace_id", "")),
    )
    await db.commit()
    return success_response(data={"status": ticket.status}, request=request)


@admin_router.get(
    "/tickets/{ticket_id}/feishu-chat",
    summary="Feishu 1:1 chat applink for the requester",
)
async def requester_feishu_chat(
    ticket_id: str,
    request: Request,
    _user: AuthUser = Depends(require_admin_read()),
    db: AsyncSession = Depends(get_db),
):
    """Hand the workbench a Feishu deeplink that opens a 1:1 with the requester.

    The 处理记录 「在飞书里联系」 button needs a p2p chat applink, and the only
    identifier that applink accepts for a person is the Feishu ``open_id``
    (``?openId=``; ``?openChatId=`` is the group form). Helix stores the SSO
    ``union_id`` on ``identity_external_identities``, not the open_id, so it
    has to come from the contact API — the same
    ``resolve_open_id_by_email`` every notification push already uses.

    Resolved on click, NOT stamped onto the ticket-detail payload: detail is
    fetched on every page view and after every mutation, and each resolve is
    a Feishu round-trip (the client's id cache is per-instance, so per
    request). One click = one lookup is the cheaper trade.

    Read-only, so ``require_admin_read`` — auditor / management can see the
    workbench and this reveals nothing beyond the directory they already read.
    """
    try:
        tid = uuid.UUID(ticket_id)
    except ValueError as e:
        raise AppException(code="NOT_FOUND", message="工单不存在", status_code=404) from e
    # No ``_load_admin`` here — that takes ``FOR UPDATE`` for the mutating
    # POSTs, and a lock on a read that then makes an external call is the
    # exact "transaction across an external API" shape to avoid.
    ticket = await SupportService(db).get(tid)
    if ticket is None:
        raise AppException(code="NOT_FOUND", message="工单不存在", status_code=404)
    requester = await db.get(IdentityPerson, ticket.requester_id)
    email = ((requester.work_email if requester else None) or "").strip()
    if not email:
        raise AppException(
            code="FEISHU_USER_NOT_FOUND",
            message="求助人没有企业邮箱，无法定位飞书账号",
            status_code=404,
        )

    from app.approval.feishu_client import FeishuApprovalClient
    from app.integrations.compat import get_feishu_config

    cfg = await get_feishu_config(session=db)
    client = FeishuApprovalClient(
        app_id=str(cfg.get("app_id") or ""),
        app_secret=str(cfg.get("app_secret") or ""),
    )
    try:
        open_id = await client.resolve_open_id_by_email(email)
    except AppException as e:
        # The client raises FEISHU_USER_NOT_FOUND/400 for "this address has no
        # Feishu user". That is a dead end for THIS ticket, not a bad request
        # from the browser — re-shape it to a 404 so the button can say
        # "对方没有飞书账号" instead of surfacing a generic failure toast.
        if e.code == "FEISHU_USER_NOT_FOUND":
            raise AppException(
                code="FEISHU_USER_NOT_FOUND",
                message="没有找到求助人的飞书账号",
                status_code=404,
            ) from e
        raise
    return success_response(
        data={
            "url": f"https://applink.feishu.cn/client/chat/open?openId={open_id}",
            "person_display": requester.display_name if requester else None,
        },
        request=request,
    )


# ── IT-side AI assist + knowledge base + analytics (PR4) ────────────


async def _load_kb(db: AsyncSession, kb_id: str) -> SupportKbArticle:
    try:
        aid = uuid.UUID(kb_id)
    except ValueError as e:
        raise AppException(code="NOT_FOUND", message="文章不存在", status_code=404) from e
    art = await db.get(SupportKbArticle, aid)
    if art is None:
        raise AppException(code="NOT_FOUND", message="文章不存在", status_code=404)
    return art


@admin_router.post("/tickets/{ticket_id}/ai-suggest", summary="AI-suggest a resolution")
async def ai_suggest(
    ticket_id: str,
    request: Request,
    user: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    from app.domains.support.copilot import suggest_resolution_for_ticket

    ticket = await _load_admin(db, ticket_id)
    result = await suggest_resolution_for_ticket(db, ticket, agent_id=user.id)
    # Persist the ai-suggest ledger row (adoption is marked when the agent's
    # reply lands close to the draft — see _mark_suggestion_adoption).
    await db.commit()
    return success_response(data=result, request=request)


@admin_router.post(
    "/tickets/{ticket_id}/draft-kb",
    status_code=201,
    summary="Draft a KB article from a resolved ticket",
)
async def draft_kb(
    ticket_id: str,
    request: Request,
    user: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    from app.domains.support.copilot import draft_kb_from_ticket
    from app.domains.support.models import KB_SOURCE_AI_TICKET

    ticket = await _load_admin(db, ticket_id)
    draft = await draft_kb_from_ticket(db, ticket)
    actor = await _require_person(db, user.id)
    art = await SupportKbService(db).create(
        title=draft["title"],
        body=draft["body"],
        category=draft["category"],
        author=actor,
        source=KB_SOURCE_AI_TICKET,
        source_ticket_id=ticket.id,
        publish=False,
        trace_id=str(getattr(request.state, "trace_id", "")),
    )
    await db.commit()
    return success_response(data=_serialize_kb(art), request=request)


@admin_router.get("/kb", summary="List KB articles (any status)")
async def kb_list(
    request: Request,
    _user: AuthUser = Depends(require_admin_read()),
    db: AsyncSession = Depends(get_db),
    status: str | None = Query(default=None),
    category: str | None = Query(default=None),
    source: str | None = Query(default=None, max_length=16),
    language: str | None = Query(default=None, max_length=8),
    q: str | None = Query(default=None, max_length=100),
    page: int = Query(default=1, ge=1),
    page_size: int = Query(default=20, ge=1, le=100),
):
    rows, total = await SupportKbService(db).list(
        status=status,
        category=category,
        source=source,
        language=language,
        q=q,
        page=page,
        page_size=page_size,
    )
    return paginated_response(
        data=[_serialize_kb(a, include_body=False) for a in rows],
        total=total,
        page=page,
        page_size=page_size,
        request=request,
    )


@admin_router.post("/kb", status_code=201, summary="Create a KB article")
async def kb_create(
    body: KbCreateBody,
    request: Request,
    user: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    actor = await _require_person(db, user.id)
    art = await SupportKbService(db).create(
        title=body.title,
        body=body.body,
        category=body.category,
        author=actor,
        publish=body.publish,
        search_aliases=body.search_aliases,
        language=body.language,
        trace_id=str(getattr(request.state, "trace_id", "")),
    )
    await db.commit()
    _spawn_kb_embed(art)
    return success_response(data=_serialize_kb(art), request=request)


# ── Feishu wiki KB mirror (source='feishu_doc') ─────────────────────
# The wiki stays the ONLY authoring surface; these endpoints configure
# which wiki subtree(s) to mirror and expose the sync state. Content
# lands in support_kb_articles via app/domains/support/feishu_docs.py.


class FeishuKbRootBody(BaseModel):
    """One human-readable Feishu knowledge source.

    A token is an implementation detail.  Administrators configure the link
    they copied from Feishu plus a durable description, and the server derives
    the token used by the sync worker.
    """

    url: str = Field(default="", max_length=512)
    # Legacy callers historically sent pasted wiki URLs in ``root_tokens``.
    # Accept that shape here and normalize before enforcing the 64-char token.
    token: str = Field(default="", max_length=512)
    description: str = Field(default="", max_length=200)


class FeishuKbConfigBody(BaseModel):
    root_tokens: list[str] = Field(default_factory=list, max_length=20)
    roots: list[FeishuKbRootBody] | None = Field(default=None, max_length=20)
    # 制度中心 integration: mirror published policy entries' wiki docs too.
    include_policy_notices: bool = True


async def _feishu_kb_config_payload(db: AsyncSession) -> dict:
    from app.domains.support.feishu_docs import (
        SETTING_INCLUDE_POLICY,
        SETTING_LAST_SYNC,
        SETTING_ROOT_TOKENS,
    )
    from app.integrations.service import get_platform_setting

    roots = await get_platform_setting(SETTING_ROOT_TOKENS, [], session=db) or []
    include_policy = bool(
        await get_platform_setting(SETTING_INCLUDE_POLICY, True, session=db),
    )
    last_sync = await get_platform_setting(SETTING_LAST_SYNC, None, session=db)
    root_configs = []
    for raw in roots:
        if isinstance(raw, dict):
            token = str(raw.get("token") or raw.get("root_token") or "").strip()
            description = str(raw.get("description") or "").strip()
            url = str(raw.get("url") or "").strip()
        else:
            token, description, url = str(raw).strip(), "", ""
        if token:
            root_configs.append(
                {
                    "token": token,
                    "description": description,
                    # Existing token-only settings are upgraded on read into a
                    # copyable/openable URL without a data migration.
                    "url": url or f"https://pingcap.feishu.cn/wiki/{token}",
                }
            )
    return {
        "root_tokens": [r["token"] for r in root_configs],
        "roots": root_configs,
        "include_policy_notices": include_policy,
        "configured": bool(roots) or include_policy,
        "last_sync": last_sync,
    }


@admin_router.get("/kb/feishu-config", summary="Feishu wiki KB mirror config + last sync state")
async def kb_feishu_config(
    request: Request,
    _user: AuthUser = Depends(require_admin_read()),
    db: AsyncSession = Depends(get_db),
):
    return success_response(data=await _feishu_kb_config_payload(db), request=request)


@admin_router.post("/kb/feishu-config", summary="Set the wiki root node tokens to mirror")
async def kb_set_feishu_config(
    body: FeishuKbConfigBody,
    request: Request,
    user: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    from app.domains.support.feishu_docs import (
        SETTING_INCLUDE_POLICY,
        SETTING_ROOT_TOKENS,
    )
    from app.integrations.models import PlatformSetting

    raw_roots = (
        body.roots
        if body.roots is not None
        else [
            FeishuKbRootBody(
                token=token,
                description="Legacy configured knowledge source",
            )
            for token in body.root_tokens
        ]
    )
    configs: list[dict[str, str]] = []
    tokens: list[str] = []
    for item in raw_roots:
        raw = str(item.url or item.token or "")
        description = str(item.description or "").strip()[:200]
        tok = raw.strip()
        # Accept a pasted wiki URL (…/wiki/<token>) or a bare node token.
        if "/wiki/" in tok:
            tok = tok.rsplit("/wiki/", 1)[1].split("?")[0].strip("/")
        if not tok:
            continue
        if not description:
            raise AppException(
                code="VALIDATION_ERROR",
                message="请为每个知识源填写说明，方便日后辨认其用途。",
                status_code=422,
            )
        # Feishu node tokens are URL-safe (observed alnum today, but '-'/'_'
        # are plausible in the wire format) — charset check, not isalnum.
        if not re.fullmatch(r"[A-Za-z0-9_-]{5,64}", tok):
            raise AppException(
                code="VALIDATION_ERROR",
                message=f"无效的知识库节点 token: {raw[:80]}",
                status_code=422,
            )
        if tok not in tokens:
            tokens.append(tok)
            configs.append(
                {
                    "token": tok,
                    "description": description,
                    "url": str(item.url or f"https://pingcap.feishu.cn/wiki/{tok}").strip(),
                }
            )
    await db.merge(PlatformSetting(key=SETTING_ROOT_TOKENS, value=configs))
    await db.merge(
        PlatformSetting(
            key=SETTING_INCLUDE_POLICY,
            value=bool(body.include_policy_notices),
        )
    )
    await AuditService(db).log(
        trace_id=str(getattr(request.state, "trace_id", "")),
        actor_type="user",
        action="support.kb.feishu_config_updated",
        status="success",
        actor_user_id=user.id,
        resource_type="platform_setting",
        resource_id=SETTING_ROOT_TOKENS,
        detail={"root_tokens": tokens, "include_policy_notices": bool(body.include_policy_notices)},
    )
    await db.commit()
    return success_response(data=await _feishu_kb_config_payload(db), request=request)


@admin_router.post("/kb/feishu-sync", status_code=202, summary="Run the wiki KB mirror sync now")
async def kb_feishu_sync_now(
    request: Request,
    user: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    """Fire-and-forget: the walk + per-doc embedding can take minutes on a
    first run, far beyond the HTTP budget — progress lands in
    ``last_sync`` (poll the config endpoint). Serialized against the cron
    run by a distributed lock inside the runner."""
    from app.core.background import spawn
    from app.domains.support.feishu_docs import (
        SETTING_INCLUDE_POLICY,
        SETTING_ROOT_TOKENS,
        run_feishu_kb_sync_once,
    )
    from app.integrations.service import get_platform_setting

    roots = await get_platform_setting(SETTING_ROOT_TOKENS, [], session=db) or []
    include_policy = bool(
        await get_platform_setting(SETTING_INCLUDE_POLICY, True, session=db),
    )
    if not roots and not include_policy:
        # Still allow a manual run when mirrored rows exist: turning
        # everything off is an ACTIVE state and the run performs the
        # decommission sweep (archive out of search) immediately instead
        # of waiting for the 6h cron.
        has_rows = (
            await db.execute(
                select(func.count())
                .select_from(SupportKbArticle)
                .where(SupportKbArticle.source == KB_SOURCE_FEISHU_DOC)
            )
        ).scalar() or 0
        if not has_rows:
            raise AppException(
                code="FEISHU_KB_NOT_CONFIGURED",
                message="请先配置要镜像的飞书知识库根节点(或开启制度中心镜像)",
                status_code=422,
            )
    await AuditService(db).log(
        trace_id=str(getattr(request.state, "trace_id", "")),
        actor_type="user",
        action="support.kb.feishu_sync_triggered",
        status="success",
        actor_user_id=user.id,
        resource_type="platform_setting",
        resource_id="support.feishu_kb",
    )
    await db.commit()
    spawn(run_feishu_kb_sync_once(), name="support-feishu-kb-sync")
    return success_response(data={"started": True}, request=request)


@admin_router.delete("/kb/{kb_id}", status_code=200, summary="Delete a KB article")
async def kb_delete(
    kb_id: str,
    request: Request,
    user: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    """Hard-delete. ``archive`` already removes an article from retrieval; this
    is for the other case — a bad or duplicated import that should leave no row
    behind. Mirrored wiki rows are refused (the next sync would recreate them)."""
    art = await _load_kb(db, kb_id)
    actor = await _require_person(db, user.id)
    await SupportKbService(db).delete(
        article=art,
        actor=actor,
        trace_id=str(getattr(request.state, "trace_id", "")),
    )
    await db.commit()
    return success_response(data={"deleted": True, "id": kb_id}, request=request)


# ── 飞书服务台知识库导入(source='helpdesk_faq')────────────────────
# 与上面的 wiki 镜像不同,这批问答是**交接**过来的:导入之后 Helix 就是编辑
# 面,管理员照常改、发布、删除。导入幂等(按问题文本 upsert),重传同一份
# 导出只更新内容、不会造出第二份。


@admin_router.post("/kb/import-helpdesk-faq", summary="Import a Feishu-helpdesk KB export")
async def kb_import_helpdesk_faq(
    request: Request,
    user: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
    file: UploadFile = File(...),
    publish: bool = Form(default=True),
    dry_run: bool = Form(default=False),
):
    from app.core.distributed_lock import LockBusy, distributed_lock
    from app.core.upload import read_upload_bounded
    from app.domains.support.helpdesk_faq_import import (
        MAX_UPLOAD_BYTES,
        import_faq_rows,
        parse_faq_workbook,
    )

    if not (file.filename or "").lower().endswith(".xlsx"):
        await file.close()
        raise AppException(
            code="UNSUPPORTED_FAQ_FILE",
            message="只支持飞书服务台导出的 .xlsx 文件",
            status_code=415,
        )
    payload = await read_upload_bounded(
        file,
        MAX_UPLOAD_BYTES,
        code="FAQ_WORKBOOK_TOO_LARGE",
        message=f"文件超过 {MAX_UPLOAD_BYTES // (1024 * 1024)} MB 上限",
        close=True,
    )
    parsed = parse_faq_workbook(payload)
    actor = await _require_person(db, user.id)
    try:
        # De-dup is a read-then-write over the whole source, so two admins
        # uploading at once would each miss the other's inserts and double the
        # library. One importer at a time; a second click gets a clear 409.
        async with distributed_lock("support_kb_faq_import", ttl=300):
            result = await import_faq_rows(
                db,
                parsed,
                actor_id=actor.id,
                publish=publish,
                dry_run=dry_run,
                trace_id=str(getattr(request.state, "trace_id", "")),
            )
            if dry_run:
                await db.rollback()
            else:
                await db.commit()
    except LockBusy as exc:
        raise AppException(
            code="FAQ_IMPORT_IN_PROGRESS",
            message="已有一个知识库导入正在进行,请稍后再试",
            status_code=409,
        ) from exc

    reembed_ids = result.pop("reembed_ids", [])
    if reembed_ids and not dry_run:
        # One drain, not 124 per-article spawns: the embeddings backend is
        # called serially anyway and drain_embeddings already commits per row.
        from app.core.background import spawn
        from app.domains.support.kb_service import run_kb_drain

        spawn(run_kb_drain(), name="support.kb_faq_embed")
    result["queued_for_embedding"] = len(reembed_ids)
    return success_response(data=result, request=request)


@admin_router.get("/kb/{kb_id}", summary="Get a KB article")
async def kb_get(
    kb_id: str,
    request: Request,
    _user: AuthUser = Depends(require_admin_read()),
    db: AsyncSession = Depends(get_db),
):
    art = await _load_kb(db, kb_id)
    return success_response(data=_serialize_kb(art), request=request)


@admin_router.post("/kb/{kb_id}", summary="Update a KB article")
async def kb_update(
    kb_id: str,
    body: KbUpdateBody,
    request: Request,
    user: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    art = await _load_kb(db, kb_id)
    actor = await _require_person(db, user.id)
    await SupportKbService(db).update(
        article=art,
        actor=actor,
        title=body.title,
        body=body.body,
        category=body.category,
        search_aliases=body.search_aliases,
        language=body.language,
        trace_id=str(getattr(request.state, "trace_id", "")),
    )
    await db.commit()
    _spawn_kb_embed(art)
    return success_response(data=_serialize_kb(art), request=request)


@admin_router.post("/kb/{kb_id}/status", summary="Publish / draft / archive a KB article")
async def kb_status(
    kb_id: str,
    body: KbStatusBody,
    request: Request,
    user: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    art = await _load_kb(db, kb_id)
    actor = await _require_person(db, user.id)
    await SupportKbService(db).set_status(
        article=art,
        status=body.status,
        actor=actor,
        trace_id=str(getattr(request.state, "trace_id", "")),
    )
    await db.commit()
    _spawn_kb_embed(art)
    return success_response(data=_serialize_kb(art), request=request)


@admin_router.post(
    "/kb/backfill-embeddings", summary="Backfill semantic embeddings for published KB"
)
async def kb_backfill_embeddings(
    request: Request,
    _user: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    """Embed published articles that have no embedding yet (bounded per call —
    this one runs inside the request, so it stays a single batch). ``pending`` in
    the response is the backlog left after the batch: call again to continue.
    No-op (embedded=0) when the embeddings backend isn't configured."""
    result = await SupportKbService(db).backfill_embeddings(limit=100)
    result.update(await SupportKbService(db).embedding_stats())
    await db.commit()
    return success_response(data=result, request=request)


class KbDebugQueryBody(BaseModel):
    question: str = Field(min_length=1, max_length=4000)
    employee_id: uuid.UUID


@admin_router.post("/kb/debug/query", summary="Simulate an employee KB answer with retrieval trace")
async def kb_debug_query(
    body: KbDebugQueryBody,
    request: Request,
    user: AuthUser = Depends(require_roles(GLOBAL_ADMIN_ROLE, AUDITOR_ROLE)),
    db: AsyncSession = Depends(get_db),
):
    """Same employee-assistant decision path, previewed with no side effects."""
    from app.assistant.conversation import simulate_employee_turn

    person = await _require_person(db, body.employee_id)
    result = await simulate_employee_turn(
        db,
        person=person,
        question=body.question,
    )
    await AuditService(db).log(
        trace_id=str(getattr(request.state, "trace_id", "")),
        actor_type="user",
        action="support.kb.debug_query",
        status="success",
        actor_user_id=user.id,
        resource_type="support_kb",
        resource_id="debug",
        detail={
            "question": body.question[:4000],
            "simulated_person_id": str(person.id),
            "article_ids": [str(item.get("id")) for item in result.docs if item.get("id")],
            "proposed_actions": [item.get("action_key") for item in result.proposed_actions],
        },
    )
    await db.commit()
    return success_response(
        data={
            "question": body.question,
            "answer": result.reply or "Helix 没有找到可确认的答案，建议转人工处理。",
            "retrieval": {
                "strategy": "same_employee_assistant_path",
                "documents_considered": None,
                "documents_returned": len(result.docs),
            },
            "sources": result.docs,
            "policy_conflicts": result.policy_conflicts,
            "tool_calls": result.tool_calls,
            "provider": result.provider,
            "proposed_actions": result.proposed_actions,
            "simulated_person": {"id": str(person.id), "display_name": person.display_name},
        },
        request=request,
    )


@admin_router.post(
    "/kb/bulk-draft-from-history", summary="Bulk-draft KB articles from resolved tickets"
)
async def kb_bulk_draft(
    request: Request,
    _user: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    """Seed the KB from resolved/closed tickets that don't yet have an article
    drafted from them — each becomes an ``ai_ticket`` DRAFT for admin review.
    Bounded per call (row cap AND wall-clock budget); idempotent (a ticket
    already sourced is skipped).

    Run mutex (#2230): the de-dup query reads COMMITTED articles, so a
    second run started while the first is still drafting sees the same
    candidate set and drafts every ticket twice —
    ``SupportKbArticle.source_ticket_id`` has no unique constraint to
    catch it. A retry click on a request that is visibly "hanging" (it
    can legitimately take minutes) is the natural way to trigger that.
    """
    from app.core.distributed_lock import LockBusy, distributed_lock

    try:
        async with distributed_lock("support_kb_bulk_draft", ttl=300):
            result = await SupportKbService(db).bulk_draft_from_history(limit=20)
            await db.commit()
    except LockBusy:
        raise AppException(
            code="BULK_DRAFT_IN_PROGRESS",
            message="批量起草正在运行中，请等它结束后再试",
            status_code=409,
        ) from None
    return success_response(data=result, request=request)


@admin_router.get(
    "/entrance-metrics", summary="Dual-entrance weekly ticket metrics (Helix vs Feishu helpdesk)"
)
async def entrance_metrics(
    request: Request,
    _user: AuthUser = Depends(require_admin_read()),
    db: AsyncSession = Depends(get_db),
    weeks: int = Query(default=12, ge=1, le=26),
):
    """The切流判据 series: weekly new tickets per entrance + Helix share.
    Feishu side reads the 12h-stale mirror (good enough for a weekly
    caliber); cross-domain read goes through the coordinator."""
    from app.core.coordinators.support_entrance_coordinator import (
        dual_entrance_weekly,
    )

    return success_response(
        data=await dual_entrance_weekly(db, weeks=weeks),
        request=request,
    )


@admin_router.get("/analytics", summary="Help-desk operational analytics")
async def support_analytics(
    request: Request,
    _user: AuthUser = Depends(require_admin_read()),
    db: AsyncSession = Depends(get_db),
):
    from app.core.coordinators.support_entrance_coordinator import unified_ticket_analytics

    svc = SupportService(db)
    by_status = await svc.queue_stats()

    by_category_rows = (
        await db.execute(
            select(SupportTicket.category, func.count()).group_by(SupportTicket.category)
        )
    ).all()
    by_category = {c: int(n) for c, n in by_category_rows}

    total = (await db.execute(select(func.count()).select_from(SupportTicket))).scalar() or 0
    ai_sourced = (
        await db.execute(
            select(func.count()).select_from(SupportTicket).where(SupportTicket.source == "ai")
        )
    ).scalar() or 0

    # Average resolution time (hours) over a bounded set of resolved tickets,
    # computed in Python to dodge the timestampdiff-binding pitfall.
    resolved = (
        await db.execute(
            select(SupportTicket.created_at, SupportTicket.resolved_at)
            .where(SupportTicket.resolved_at.is_not(None))
            .order_by(SupportTicket.resolved_at.desc())
            .limit(500)
        )
    ).all()
    avg_hours = None
    if resolved:
        deltas = [(r - c).total_seconds() / 3600.0 for c, r in resolved if c and r and r >= c]
        if deltas:
            avg_hours = round(sum(deltas) / len(deltas), 1)

    kb_rows = (
        await db.execute(
            select(SupportKbArticle.status, func.count()).group_by(SupportKbArticle.status)
        )
    ).all()
    kb_by_status = {s: int(n) for s, n in kb_rows}
    kb_ai_sourced = (
        await db.execute(
            select(func.count())
            .select_from(SupportKbArticle)
            .where(SupportKbArticle.source == "ai_ticket")
        )
    ).scalar() or 0

    # Real deflection funnel: of resolved deflection sessions (the employee
    # made a choice), what share ended without a ticket. This is the honest
    # deflection rate — tickets actually prevented — distinct from the proxy
    # signals above (AI-sourced share, KB growth).
    deflect_rows = (
        await db.execute(
            select(SupportDeflectionEvent.outcome, func.count()).group_by(
                SupportDeflectionEvent.outcome
            )
        )
    ).all()
    deflect_by_outcome = {o: int(n) for o, n in deflect_rows}
    deflected = deflect_by_outcome.get("deflected", 0)
    escalated = deflect_by_outcome.get("escalated", 0)
    decided = deflected + escalated
    deflection_rate = round(deflected / decided, 3) if decided else None

    # ai-suggest ledger: adoption = an agent reply landed textually close to
    # a recent draft — the honest "AI saves IT time" number.
    from app.domains.support.models import SupportAiSuggestion

    sug_total = (
        await db.execute(select(func.count()).select_from(SupportAiSuggestion))
    ).scalar() or 0
    sug_adopted = (
        await db.execute(
            select(func.count())
            .select_from(SupportAiSuggestion)
            .where(
                SupportAiSuggestion.adopted_at.is_not(None),
            )
        )
    ).scalar() or 0

    # CSAT — mirror caliber (NULL = not rated, excluded; rate = share of
    # 满意(3) among rated), so the number is directly comparable with the
    # Feishu-helpdesk analytics page during the dual-entrance period.
    sat_rows = (
        await db.execute(
            select(SupportTicket.satisfaction_score, func.count())
            .where(SupportTicket.satisfaction_score.is_not(None))
            .group_by(SupportTicket.satisfaction_score)
        )
    ).all()
    sat_by_score = {int(s): int(n) for s, n in sat_rows}
    sat_scored = sum(sat_by_score.values())
    satisfaction_avg = (
        round(sum(s * n for s, n in sat_by_score.items()) / sat_scored, 2) if sat_scored else None
    )

    satisfaction_rate = round(sat_by_score.get(3, 0) / sat_scored, 3) if sat_scored else None

    unified = await unified_ticket_analytics(db, days=30)
    return success_response(
        data={
            "tickets_total": int(total),
            "by_status": by_status,
            "by_category": by_category,
            "ai_sourced_tickets": int(ai_sourced),
            "avg_resolution_hours": avg_hours,
            "kb_by_status": kb_by_status,
            "kb_ai_sourced": int(kb_ai_sourced),
            "deflection_sessions": int(sum(deflect_by_outcome.values())),
            "deflection_deflected": deflected,
            "deflection_escalated": escalated,
            "deflection_expired": deflect_by_outcome.get("expired", 0),
            "deflection_rate": deflection_rate,
            "satisfaction_scored": int(sat_scored),
            "satisfaction_avg": satisfaction_avg,
            "satisfaction_rate": satisfaction_rate,
            "satisfaction_by_score": sat_by_score,
            "ai_suggest_total": int(sug_total),
            "ai_suggest_adopted": int(sug_adopted),
            "ai_suggest_adoption_rate": (round(sug_adopted / sug_total, 3) if sug_total else None),
            # Unified migration-period population: internal Helix tickets plus
            # the Feishu mirror. Keep the legacy internal fields above for
            # queue consumers, while the cockpit reads this explicit contract.
            "unified": unified,
        },
        request=request,
    )


@admin_router.get("/operations-analytics", summary="Unified IT service operations cockpit")
async def support_operations_cockpit(
    request: Request,
    days: int = Query(default=30, ge=1, le=366),
    source: str | None = Query(default=None, pattern="^(helix|feishu)$"),
    _user: AuthUser = Depends(require_admin_read()),
    db: AsyncSession = Depends(get_db),
):
    from app.core.coordinators.support_operations_analytics import (
        support_operations_analytics,
    )

    return success_response(
        data=await support_operations_analytics(db, days=days, source=source),
        request=request,
    )


# ── AI copilot admin config (enable + provider) ─────────────────────
# The Qwen key lives on the existing AI/LLM connector; this just flips
# the master switch + (optionally) pins the provider.


@admin_router.get("/copilot/config", summary="Copilot enable state + resolved backend")
async def copilot_config(
    request: Request,
    _user: AuthUser = Depends(require_admin_read()),
    db: AsyncSession = Depends(get_db),
):
    from app.domains.support.copilot import (
        SETTING_ENABLED,
        SETTING_PROVIDER,
        _resolve_llm,
        copilot_available,
    )
    from app.integrations.service import get_platform_setting

    cfg = await _resolve_llm(db)
    return success_response(
        data={
            "enabled": bool(await get_platform_setting(SETTING_ENABLED, False, session=db)),
            "available": await copilot_available(db),
            "provider": cfg.provider if cfg else None,
            "model": cfg.model if cfg else None,
            "provider_setting": await get_platform_setting(SETTING_PROVIDER, None, session=db),
        },
        request=request,
    )


class CopilotConfigBody(BaseModel):
    enabled: bool
    provider: str | None = None  # "bailian" | "anthropic" | null(auto)


@admin_router.post("/copilot/config", summary="Enable/disable copilot + pin provider")
async def set_copilot_config(
    body: CopilotConfigBody,
    request: Request,
    user: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    from app.domains.support.copilot import (
        SETTING_ENABLED,
        SETTING_PROVIDER,
        _resolve_llm,
        copilot_available,
    )
    from app.integrations.models import PlatformSetting

    await db.merge(PlatformSetting(key=SETTING_ENABLED, value=bool(body.enabled)))
    if body.provider in ("bailian", "anthropic"):
        await db.merge(PlatformSetting(key=SETTING_PROVIDER, value=body.provider))
    elif body.provider in (None, "", "auto"):
        await db.merge(PlatformSetting(key=SETTING_PROVIDER, value=None))
    await db.commit()
    from app.integrations.service import get_platform_setting

    cfg = await _resolve_llm(db)
    return success_response(
        data={
            "enabled": bool(await get_platform_setting(SETTING_ENABLED, False, session=db)),
            "available": await copilot_available(db),
            "provider": cfg.provider if cfg else None,
            "model": cfg.model if cfg else None,
        },
        request=request,
    )