"""The employee assistant, rendered as a web page instead of a Feishu chat.

**What this file corrects.** ``/me/support`` has been the front door of IT help
since 2026-08-17 — the chat IS the page — and until now it called the model
kernel bare::

    result = await run_copilot_turn(db, person, history)

No propose tools, so it could not submit anything. No read bridge, so it could
reach fourteen tools instead of a hundred and fifty. No server-side grounding,
so it answered 「知识库里没有」 for questions the published article answered. No
transcript, so nothing an employee was told on this surface appeared in the
admin console, the knowledge-gap queue, the funnel, or the evaluation harness.
Meanwhile the Feishu bot — the same product, the same employees — had all five.

Every one of those gaps had already been diagnosed and fixed once. They were
fixed in :mod:`app.assistant.conversation`, on the channel that happened to be
under repair at the time, and the second surface silently kept the old
behaviour. This module exists so there is no second surface: it runs the same
preflight, the same tools, the same guards and the same transcript writer, and
the ONLY thing it does differently is where the finished offer is drawn
(:mod:`app.assistant.offer_delivery`).

**What it deliberately does NOT share with the browser.** The client sends one
message. It does not send history, and the history it displays is not what the
model is shown — the thread is the server's, keyed by person, exactly as in
Feishu. That is a change from the old endpoint, which took the whole transcript
from the page. With a write path now in reach, "the caller supplies what the
assistant remembers" is a surface where a crafted request could put words in the
assistant's mouth and then act on them; the corroborator would still refuse to
put an unproven id on an offer, but the cheaper fix is not to accept the input
at all.
"""

from __future__ import annotations

import contextlib
import json
import logging
import time
from dataclasses import dataclass, field
from datetime import UTC, datetime
from typing import Any

from sqlalchemy.ext.asyncio import AsyncSession

from app.assistant import messages as canned
from app.assistant import reply_language
from app.assistant import session as session_store
from app.assistant.models import (
    CHANNEL_WEB,
    OUTCOME_ANSWERED,
    OUTCOME_FAILED,
    OUTCOME_UNAVAILABLE,
    ROLE_ASSISTANT,
    ROLE_USER,
)
from app.core.locale import Lang, resolve_lang
from app.identity.person_models import IdentityPerson

logger = logging.getLogger(__name__)

#: How many un-confirmed offers a page will draw when it rehydrates. A
#: conversation cannot realistically accumulate more inside one hour, and the
#: cap keeps a read path from turning into an unbounded loop that also makes one
#: settings round-trip per row (``confirm_url``).
_MAX_REHYDRATED_OFFERS = 10


def web_chat_id(person: IdentityPerson | Any) -> str:
    """The thread key for one employee's web conversation.

    One thread per person, not per browser tab: two tabs are one employee
    talking to one assistant, and giving them separate memories would let the
    same question get two different answers depending on which tab it was typed
    in. Segmentation into conversations is by the working-memory TTL, which is
    what ``conversation._current_conversation`` already does for Feishu.
    """
    return f"web:{getattr(person, 'id', person)}"


@dataclass(frozen=True)
class WebTurnResult:
    """One web turn, as the page needs to render it."""

    outcome: str
    #: Markdown. Not flattened — this surface renders it, and flattening here
    #: was never a channel requirement, only a Feishu *text*-message one.
    reply: str
    #: Confirmation offers to draw beneath the reply. Real proposals with real
    #: tokens; the employee has confirmed nothing by being shown one.
    offers: list[dict[str, Any]] = field(default_factory=list)
    #: Sources the answer is allowed to cite, as ``[{title, url}]``.
    docs: list[dict[str, str]] = field(default_factory=list)
    tool_calls: list[str] = field(default_factory=list)
    provider: str | None = None
    latency_ms: int = 0
    #: The stored assistant message, so a rating attaches to this exact answer.
    message_id: str | None = None
    conversation_id: str | None = None
    #: Is the kernel configured and on. Stable across turns — the page gates its
    #: composer on it and must not disable itself because one turn erred.
    available: bool = True
    #: Did THIS turn fail. Orthogonal to ``available``; see ``CopilotResult``.
    turn_failed: bool = False
    suggest_escalate: bool = True
    #: Deterministic text the page must show, independent of what the model
    #: said — currently only the "nothing was submitted, use the portal"
    #: sentence, whose whole point is that a model must not be trusted to
    #: narrate its own failed side effect.
    notices: list[str] = field(default_factory=list)
    #: This turn was refused because another turn is already running for this
    #: employee. Not an error: the page re-enables its composer and says so.
    busy: bool = False


async def reset_web_thread(
    *, person: IdentityPerson, lang: Lang | None = None,
) -> tuple[bool, str]:
    """Forget this employee's web thread. Returns ``(cleared, message)``.

    The returned flag is honest about a Redis failure rather than claiming a
    reset that did not happen — the same contract, and the same reason, as the
    chat channel's 「新话题」.
    """
    lang = lang or resolve_lang(person)
    cleared = await session_store.clear_history(
        person_id=person.id, chat_id=web_chat_id(person),
    )
    return cleared, (canned.reset_done(lang) if cleared else canned.reset_failed(lang))


async def run_web_turn(
    db: AsyncSession,
    *,
    person: IdentityPerson,
    question: str,
    trace_id: str,
    lang: Lang | None = None,
) -> WebTurnResult:
    """One question from the web page in, one answer out, transcript written.

    Mirrors :func:`app.assistant.conversation.run_turn` step for step. Where the
    two differ, the difference is named in a comment — an unexplained divergence
    between these two functions is how the surfaces drifted apart the first time.
    """
    from app.assistant import conversation, corroborate, mcp_bridge  # noqa: PLC0415
    from app.assistant import propose_tools as proposals_tools  # noqa: PLC0415
    from app.assistant.offer_delivery import WebOfferDelivery  # noqa: PLC0415
    from app.assistant.prompt import employee_system_prompt  # noqa: PLC0415
    from app.assistant.reply_hygiene import (  # noqa: PLC0415
        add_missing_destination,
        to_web_markdown,
    )
    from app.domains.support.copilot import (  # noqa: PLC0415
        KB_TOOL_NAME,
        CopilotResult,
        run_copilot_turn,
    )

    lang = lang or resolve_lang(person)
    # Captured before anything can fail: a failed flush inside a tool expires
    # every instance in the session, so ``person.id`` read afterwards needs IO
    # and raises — including from the logging line in the error handler.
    person_id = person.id
    chat_id = web_chat_id(person)

    # Serialized per employee, as in chat. Two tabs a second apart would
    # otherwise read the same history, reply out of order, and — because ``seq``
    # comes from ``turn_count`` — collide on the transcript's unique key.
    owner = await session_store.acquire_turn_lock(person_id=person_id, chat_id=chat_id)
    if owner is None:
        return WebTurnResult(
            outcome=OUTCOME_FAILED,
            reply=canned.busy(lang),
            busy=True,
            suggest_escalate=False,
        )

    try:
        epoch = await session_store.current_epoch(person_id=person_id, chat_id=chat_id)
        prior = await session_store.load_history(db, person_id=person_id, chat_id=chat_id)
        continuing = bool(prior)

        pre = await conversation.build_preflight(db, question, person_id=person_id)
        history = (
            [{"role": ROLE_USER, "content": question}]
            if pre.drop_history
            else [*prior, {"role": ROLE_USER, "content": question}]
        )

        chat = await conversation._current_conversation(  # noqa: SLF001 — same package
            db,
            person=person,
            chat_id=chat_id,
            continuing=continuing,
            channel=CHANNEL_WEB,
        )
        from app.core.coordinators.assistant_support_coordinator import (  # noqa: PLC0415
            ensure_active_assistant_request,
        )

        service_request = await ensure_active_assistant_request(db, conversation=chat)
        # Snapshots before the model/tool loop, for the reason ``run_turn``
        # states at length: a failed tool flush rolls the session back and
        # expires every instance, and the failure path still has to write a
        # transcript.
        service_request_id = service_request.id
        conversation_id = chat.id
        await db.commit()

        seen = corroborate.Corroborator()
        delivery = WebOfferDelivery()
        bridged_calls: list[str] = []
        propose_dispatch = proposals_tools.make_dispatch(
            person=person,
            chat_id=chat_id,
            # No ``open_id`` exists here and none is needed: ``delivery`` is
            # what decides where the offer goes, and this one goes into the
            # response. A non-empty sentinel keeps the tool surface identical
            # to the chat channel's without being routable anywhere.
            open_id="web",
            lang=lang,
            trace_id=trace_id,
            corroborator=seen,
            conversation_id=conversation_id,
            delivery=delivery,
        )
        bridge_dispatch = mcp_bridge.make_dispatch(person=person, called=bridged_calls)
        extra_tools = [*proposals_tools.propose_tools(), *mcp_bridge.bridge_tools()]

        async def extra_dispatch(
            name: str,
            tool_input: dict[str, Any],
            session: Any,
            caller: Any,
        ) -> str:
            if name in (mcp_bridge.TOOL_SEARCH, mcp_bridge.TOOL_CALL):
                return str(await bridge_dispatch(name, tool_input, session, caller))
            return str(await propose_dispatch(name, tool_input, session, caller))

        kb_docs: list[dict[str, str]] = list(pre.docs)

        def _note_tool(name: str, raw: str) -> None:
            seen.observe(raw)
            if name != KB_TOOL_NAME:
                return
            try:
                for item in json.loads(raw) or []:
                    url = (item or {}).get("url")
                    if url and not any(d.get("url") == url for d in kb_docs):
                        kb_docs.append({"title": item.get("title") or url, "url": url})
            except Exception:  # noqa: BLE001
                # A tool result we cannot read costs the citation, never the
                # turn: anything escaping here is caught by the kernel's
                # catch-all, which throws the whole answer away.
                logger.debug("assistant: unusable KB result while collecting sources")

        started = time.monotonic()
        try:
            forced_reply = await conversation._forced_software_role_lookup(  # noqa: SLF001
                db=db,
                person=person,
                question=question,
                trace_id=trace_id,
                called=bridged_calls,
            )
            if forced_reply is not None:
                result: CopilotResult | None = CopilotResult(
                    available=True,
                    reply=forced_reply,
                    tool_calls=[],
                    provider="helix-catalogue",
                )
            else:
                result = await run_copilot_turn(
                    db,
                    person,
                    history,
                    extra_tools=extra_tools,
                    extra_dispatch=extra_dispatch,
                    system_prompt=employee_system_prompt(
                        lang, person=person, channel=CHANNEL_WEB,
                    )
                    + pre.system_context,
                    observer=_note_tool,
                )
        except Exception:
            logger.exception("assistant: web turn failed person=%s", person_id)
            # Rolled back for the reason ``run_turn`` documents: neither the
            # kernel nor its dispatcher rolls back, so a DBAPI error inside a
            # bridged tool leaves this session needing one — and the very next
            # statement is the transcript write.
            await db.rollback()
            with contextlib.suppress(Exception):
                # A rollback expires every instance, ``person`` included, and
                # the audit/transcript path reads it after this returns.
                await db.refresh(person)
            result = None
        latency_ms = int((time.monotonic() - started) * 1000)

        turn = _shape_turn(
            result,
            lang=lang,
            offers=delivery.offers,
            bridged_calls=bridged_calls,
            kb_prefetched=pre.kb_prefetched,
            prefetched_tools=pre.identifier_tools,
            kb_tool_name=KB_TOOL_NAME,
            bridge_meta_tool=mcp_bridge.TOOL_CALL,
        )
        reply = add_missing_destination(to_web_markdown(turn.reply), lang)

        if turn.outcome == OUTCOME_ANSWERED:
            # Only a real answer earns a place in working memory: replaying
            # "the assistant is unavailable" would poison every later turn.
            await session_store.save_history(
                db,
                person_id=person_id,
                chat_id=chat_id,
                history=[*history, {"role": ROLE_ASSISTANT, "content": reply}],
                expect_epoch=epoch,
            )

        message_id = await conversation._record(  # noqa: SLF001 — same package
            db,
            conversation_id=conversation_id,
            question=question,
            turn=conversation.TurnResult(
                turn.outcome,
                reply,
                turn.tool_calls,
                turn.provider,
                latency_ms,
                reply_rich=reply,
            ),
            trace_id=trace_id,
            sources=conversation._public_source_links(kb_docs),  # noqa: SLF001
            service_request_id=service_request_id,
        )

        return WebTurnResult(
            outcome=turn.outcome,
            reply=reply,
            offers=delivery.offers,
            docs=conversation._public_source_links(kb_docs),  # noqa: SLF001
            tool_calls=turn.tool_calls,
            provider=turn.provider,
            latency_ms=latency_ms,
            message_id=message_id,
            conversation_id=str(conversation_id),
            available=turn.available,
            turn_failed=turn.turn_failed,
            suggest_escalate=turn.suggest_escalate,
            notices=list(delivery.notices),
        )
    finally:
        await session_store.release_turn_lock(
            person_id=person_id, chat_id=chat_id, owner=owner,
        )


@dataclass(frozen=True)
class _Shaped:
    outcome: str
    reply: str
    tool_calls: list[str]
    provider: str | None
    available: bool
    turn_failed: bool
    suggest_escalate: bool


def _shape_turn(
    result: Any,
    *,
    lang: Lang,
    offers: list[dict[str, Any]],
    bridged_calls: list[str],
    kb_prefetched: bool,
    prefetched_tools: list[str],
    kb_tool_name: str,
    bridge_meta_tool: str,
) -> _Shaped:
    """Turn a kernel result into what the transcript and the page each need.

    Kept as its own function, and deliberately model-free, so the outcome rules
    can be tested without a provider. Every branch mirrors ``run_turn``'s, for
    the same stated reasons — the important one being that a provider outage
    must NOT be recorded as ``answered``, or it lands in the helpful-rate
    denominator and drags a quality metric down for a reason that has nothing to
    do with the answers.
    """
    from app.assistant.conversation import (  # noqa: PLC0415
        _claims_card_sent_without_delivery,
    )

    if result is None:
        return _Shaped(OUTCOME_FAILED, canned.failed(lang), [], None, True, True, True)
    if getattr(result, "turn_failed", False):
        return _Shaped(
            OUTCOME_FAILED, canned.failed(lang), [], result.provider, True, True, True,
        )
    if not result.available:
        return _Shaped(
            OUTCOME_UNAVAILABLE,
            canned.kernel_unavailable(lang),
            [],
            result.provider,
            False,
            False,
            True,
        )

    outcome = OUTCOME_ANSWERED
    reply = (result.reply or "").strip()
    if offers:
        # The offer states what will be submitted, that nothing is submitted
        # until confirmed, and when it expires. A narration above it is a second
        # paragraph about the thing the employee is already reading — so the
        # model was told to reply empty, and this is the stand-in. Recorded
        # rather than stored blank: ``load_history`` drops empty turns, so a
        # blank would show the model an unanswered request and it would offer
        # the same thing again.
        reply = canned.offer_shown(lang)
    elif _claims_card_sent_without_delivery(reply):
        # Never let model text manufacture evidence of an offer. 「创建吧」 can
        # make a model echo an earlier turn's card narration without calling a
        # proposal tool at all, and there is no request until one was made.
        reply = canned.nothing_submitted(lang)
        outcome = OUTCOME_FAILED
    elif not reply:
        reply = canned.failed(lang)
        outcome = OUTCOME_FAILED

    # The bridge's meta-tool name answers no question anyone has; report the
    # tools it actually ran.
    tool_calls = [n for n in (result.tool_calls or []) if n != bridge_meta_tool]
    if kb_prefetched:
        tool_calls.insert(0, kb_tool_name)
    tool_calls.extend(prefetched_tools)
    tool_calls.extend(bridged_calls)
    return _Shaped(
        outcome,
        reply,
        tool_calls,
        result.provider,
        True,
        # 🔴 Derived, not hardcoded. Two branches above turn an ANSWERED turn
        # into a FAILED one — an empty completion, and a model claiming an offer
        # it never made — and both used to report ``turn_failed=False``. The
        # page then drew 👍/👎 under a canned 「我这边出了点问题」, and a 👎 filed a
        # knowledge gap saying an article was missing. That is a work item
        # nobody can ever close, which is the exact thing the frontend's
        # ``degraded`` flag exists to prevent.
        outcome == OUTCOME_FAILED,
        bool(getattr(result, "suggest_escalate", True)),
    )


async def web_transcript(
    db: AsyncSession,
    *,
    person: IdentityPerson,
    conversation_id: Any = None,
    limit: int = 40,
) -> list[dict[str, str]]:
    """One web conversation of this employee's, oldest first.

    Read from the stored transcript rather than from the page, because this is
    what a ticket quotes. A page can be told anything; the transcript is what
    the assistant actually said, and an escalation that quotes something else
    hands a human a conversation that never happened.

    ``conversation_id`` says WHICH thread — the page passes back the id its last
    answer carried. It selects; it does not authorise: the query still filters
    on this person and on the web channel, so a borrowed id reads nothing.
    Without it this falls back to the most recent thread, which is right while
    the employee is mid-conversation and wrong after a reload following a long
    gap — the page would be empty and the ticket would quote yesterday.
    """
    from sqlalchemy import select  # noqa: PLC0415

    from app.assistant.models import AssistantConversation, AssistantMessage  # noqa: PLC0415

    chat_id = web_chat_id(person)
    query = (
        select(AssistantConversation.id)
        .where(
            AssistantConversation.person_id == person.id,
            AssistantConversation.channel == CHANNEL_WEB,
            AssistantConversation.chat_id == chat_id,
        )
        .order_by(AssistantConversation.last_message_at.desc())
        .limit(1)
    )
    if conversation_id is not None:
        query = query.where(AssistantConversation.id == conversation_id)
    conversation_id = (await db.execute(query)).scalar_one_or_none()
    if conversation_id is None:
        return []
    rows = (
        (
            await db.execute(
                select(AssistantMessage)
                .where(AssistantMessage.conversation_id == conversation_id)
                # Newest first, then reversed: the tail is what a long thread
                # needs kept, and LIMIT on an ascending order keeps the head.
                .order_by(AssistantMessage.seq.desc())
                .limit(limit)
            )
        )
        .scalars()
        .all()
    )
    return [
        {"role": row.role, "content": row.content}
        for row in reversed(rows)
        if row.role in (ROLE_USER, ROLE_ASSISTANT) and (row.content or "").strip()
    ]


async def web_thread(
    db: AsyncSession, *, person: IdentityPerson, limit: int = 40,
) -> dict[str, Any]:
    """This employee's current web conversation, as the page has to draw it.

    The page used to start empty on every mount while the server kept the
    thread for two hours. That is not a cosmetic mismatch: the employee comes
    back to a blank box, types 「帮我提交」, and the assistant answers from a
    context that is nowhere on screen — and the two controls that would have
    let them see or clear it (「新话题」 and 转人工) were both hidden behind
    ``items.length > 0``. The page has to be able to show what the server
    remembers.

    Pending offers are rehydrated too, and for the same reason: a proposal the
    employee has not confirmed is live for an hour, and a page that forgets it
    leaves them an offer they can neither act on nor cancel.
    """
    from sqlalchemy import select  # noqa: PLC0415

    from app.assistant.actions import get_action  # noqa: PLC0415
    from app.assistant.confirm_card import confirm_url  # noqa: PLC0415
    from app.assistant.models import (  # noqa: PLC0415
        STATUS_PENDING,
        AssistantActionProposal,
        AssistantConversation,
        AssistantMessage,
    )
    from app.assistant.offer_delivery import offer_payload  # noqa: PLC0415
    from app.assistant.proposals import (  # noqa: PLC0415
        ACTION_TOKEN_PURPOSE,
        PROPOSAL_TTL_MINUTES,
    )
    from app.core.security import create_action_token  # noqa: PLC0415

    # 这一段对话在用哪种语言 —— 页面刷新之后重画的那些待确认卡片,不该因为
    # 这一次是「恢复」而换一种语言(见 :mod:`app.assistant.reply_language`)。
    chat_id = web_chat_id(person)
    lang = await reply_language.for_card(person=person, chat_id=chat_id)
    conversation_id = (
        await db.execute(
            select(AssistantConversation.id)
            .where(
                AssistantConversation.person_id == person.id,
                AssistantConversation.channel == CHANNEL_WEB,
                AssistantConversation.chat_id == chat_id,
            )
            .order_by(AssistantConversation.last_message_at.desc())
            .limit(1)
        )
    ).scalar_one_or_none()
    if conversation_id is None:
        return {"conversation_id": None, "messages": [], "offers": []}

    # MESSAGES are shown only while the SERVER still has the thread. Once
    # working memory has expired the next question opens a new conversation, so
    # replaying the old one would show the employee a context the assistant no
    # longer has — the same mismatch one direction over.
    #
    # ⚠️ OFFERS are not gated on it. ``load_history`` returns [] on any Redis
    # fault (it degrades rather than refusing), and a pending proposal lives in
    # the DATABASE with its own hour-long expiry — so gating both on Redis meant
    # a blip hid a live, confirmable offer behind a page that claimed there was
    # no conversation at all. That is precisely what this reader was written to
    # stop happening.
    remembered = bool(
        await session_store.load_history(db, person_id=person.id, chat_id=chat_id)
    )

    rows = (
        (
            await db.execute(
                select(AssistantMessage)
                .where(AssistantMessage.conversation_id == conversation_id)
                .order_by(AssistantMessage.seq.desc())
                .limit(limit)
            )
        )
        .scalars()
        .all()
    )
    messages = [] if not remembered else [
        {
            "role": row.role,
            "content": row.content,
            "message_id": str(row.id) if row.role == ROLE_ASSISTANT else None,
            "docs": row.sources or [],
            "outcome": row.outcome,
        }
        for row in reversed(rows)
        if row.role in (ROLE_USER, ROLE_ASSISTANT) and (row.content or "").strip()
    ]

    # ⚠️ ``expires_at`` is part of the filter, not just ``status``. Expiry is
    # LAZY in this repo — ``expire_if_lapsed`` runs on the proposal read path,
    # there is no sweeper — so a proposal that died an hour ago is still
    # ``pending`` in the table. Returning it drew an enabled 确认提交 with a full
    # sixty minutes on the clock over a call that answers 410, which is the
    # "enabled button on something the confirm call will refuse" failure the
    # proposal router's own docstring argues against.
    now = datetime.now(UTC)
    pending = (
        (
            await db.execute(
                select(AssistantActionProposal)
                .where(
                    AssistantActionProposal.conversation_id == conversation_id,
                    AssistantActionProposal.person_id == person.id,
                    AssistantActionProposal.status == STATUS_PENDING,
                    AssistantActionProposal.expires_at > now.replace(tzinfo=None),
                )
                .order_by(AssistantActionProposal.created_at.asc())
                .limit(_MAX_REHYDRATED_OFFERS)
            )
        )
        .scalars()
        .all()
    )
    offers: list[dict[str, Any]] = []
    for proposal in pending:
        action = get_action(proposal.action_key)
        if action is None:
            # The flow was retired since. Nothing can submit it, so showing an
            # offer for it would be a button that only ever fails.
            continue
        # A fresh read-only token for the snapshot page — minted for the
        # offer's REMAINING life, not for a fresh full TTL. Renewing the full
        # hour on every page mount would keep the snapshot page readable
        # indefinitely after the offer itself was dead, which is the exact
        # invariant ``create_proposal`` records having had to fix once already
        # ("a 24h token on a 60-minute proposal").
        expires_at = proposal.expires_at
        if expires_at is not None and expires_at.tzinfo is None:
            expires_at = expires_at.replace(tzinfo=UTC)
        remaining = (
            max(1, int((expires_at - now).total_seconds() // 60))
            if expires_at is not None
            else PROPOSAL_TTL_MINUTES
        )
        token = create_action_token(
            purpose=ACTION_TOKEN_PURPOSE,
            subject_id=str(person.id),
            resource_id=str(proposal.id),
            ttl_minutes=remaining,
        )
        offers.append(
            offer_payload(
                action=action,
                params=dict(proposal.params or {}),
                url=await confirm_url(
                    db, token=token, digest=proposal.params_hash,
                ),
                lang=lang,
                proposal_id=str(proposal.id),
                expires_in_minutes=remaining,
            )
        )
    return {
        "conversation_id": str(conversation_id),
        "messages": messages,
        "offers": offers,
    }