"""The web page runs the same assistant as the chat.

The gap this closes is not subtle. ``/me/support`` — the front door of IT help
for anyone not in Feishu — called the model kernel bare: fourteen tools instead
of a hundred and fifty, no propose path, no server-side grounding, and no
transcript, so nothing said there reached the admin console, the knowledge-gap
queue, the funnel or the evaluation harness.

These tests pin the properties that make "same assistant" true rather than
claimed, and the handful of places the two surfaces are allowed to differ.
"""
from __future__ import annotations

import uuid

import pytest
from sqlalchemy import func, select

from app.assistant.models import (
    CHANNEL_FEISHU,
    CHANNEL_WEB,
    OUTCOME_ANSWERED,
    OUTCOME_FAILED,
    OUTCOME_UNAVAILABLE,
    AssistantConversation,
    AssistantMessage,
)
from app.assistant.web_turn import (
    run_web_turn,
    web_chat_id,
    web_transcript,
)
from tests.test_assistant.conftest import FakeCopilotResult, make_feishu_person

pytestmark = pytest.mark.asyncio


async def _person(db, *, email: str | None = None):
    return await make_feishu_person(
        db,
        subject=f"on_web_{uuid.uuid4().hex[:10]}",
        email=email or f"web.{uuid.uuid4().hex[:8]}@test.com",
        name="网页提问的员工",
    )


# ── it is the SAME turn ─────────────────────────────────────────────


async def test_the_model_gets_the_propose_tools_and_the_read_bridge(
    db_session, enabled_gate, kernel,
):
    """🔴 The whole gap in one assertion.

    The old endpoint passed no ``extra_tools`` at all, so the web assistant
    could neither reach the platform's read-only tool surface nor offer to
    submit anything — on the page where most employees ask for help first.
    """
    from app.assistant import mcp_bridge, propose_tools

    person = await _person(db_session)
    await run_web_turn(
        db_session, person=person, question="我想申请一个邮箱别名", trace_id="t-1",
    )

    names = {tool["name"] for tool in kernel["state"]["last_extra_tools"]}
    assert propose_tools.TOOL_ACTION_SEARCH in names
    assert propose_tools.TOOL_ACTION_PROPOSE in names
    assert mcp_bridge.TOOL_SEARCH in names
    assert mcp_bridge.TOOL_CALL in names


async def test_the_prompt_is_the_web_one_not_the_feishu_one(
    db_session, enabled_gate, kernel,
):
    """One prompt, two surfaces — and the surface clauses have to follow."""
    person = await _person(db_session)
    await run_web_turn(db_session, person=person, question="你好", trace_id="t-2")

    prompt = kernel["state"]["last_system_prompt"]
    assert "inside Feishu chat" not in prompt
    assert "/me/support" in prompt
    # The one that would actively mislead: the page HAS buttons.
    assert "there is no page here" not in prompt


async def test_the_server_grounds_the_question_before_the_model_sees_it(
    db_session, enabled_gate, kernel,
):
    """🔴 A prompt instruction is not a control plane.

    Production had the model answer 「知识库中没有…」 without calling the KB tool
    once, for a question the published article answered. The chat channel grew a
    server-side preflight for it; the web page never did.
    """
    from app.domains.support.models import KB_PUBLISHED, SupportKbArticle

    article = SupportKbArticle(
        title="如何连接公司 Wi-Fi",
        body="选择 SSID「PingCAP-Corp」，用工号和邮箱密码登录。",
        category="network",
        status=KB_PUBLISHED,
        source="manual",
    )
    db_session.add(article)
    await db_session.commit()

    person = await _person(db_session)
    turn = await run_web_turn(
        db_session, person=person, question="怎么连接公司 Wi-Fi？", trace_id="t-3",
    )

    prompt = kernel["state"]["last_system_prompt"]
    assert "KB EVIDENCE PREFETCH" in prompt
    assert "如何连接公司 Wi-Fi" in prompt
    # And the turn reports the retrieval it really did, so a grounded citation
    # does not read as fabricated in the transcript.
    from app.domains.support.copilot import KB_TOOL_NAME

    assert KB_TOOL_NAME in turn.tool_calls


async def test_a_turn_is_written_to_the_transcript(db_session, enabled_gate, kernel):
    """🔴 Without this, nothing said on the web page exists to an administrator."""
    person = await _person(db_session)
    turn = await run_web_turn(
        db_session, person=person, question="我的电脑保修到什么时候", trace_id="t-4",
    )
    await db_session.commit()

    assert turn.message_id is not None
    conversation = (
        await db_session.execute(
            select(AssistantConversation).where(
                AssistantConversation.person_id == person.id,
            )
        )
    ).scalar_one()
    assert conversation.channel == CHANNEL_WEB
    assert conversation.chat_id == web_chat_id(person)

    rows = (
        (
            await db_session.execute(
                select(AssistantMessage)
                .where(AssistantMessage.conversation_id == conversation.id)
                .order_by(AssistantMessage.seq)
            )
        )
        .scalars()
        .all()
    )
    assert [r.role for r in rows] == ["user", "assistant"]
    assert rows[0].content == "我的电脑保修到什么时候"
    assert rows[1].content == FakeCopilotResult().reply


async def test_the_two_channels_are_separate_threads_for_one_person(
    db_session, enabled_gate, kernel,
):
    """🔴 One person, two surfaces, two conversations.

    ``_current_conversation`` matched on ``chat_id`` alone until this change.
    Sharing a row would interleave a phone conversation with a browser one in
    the same transcript, and hand each surface the other's working memory.
    """
    from app.assistant import conversation as conv

    person = await _person(db_session)
    await run_web_turn(db_session, person=person, question="网页这一句", trace_id="t-5")
    await db_session.commit()

    feishu_chat = await conv._current_conversation(
        db_session, person=person, chat_id="oc-web-parity", continuing=False,
        channel=CHANNEL_FEISHU,
    )
    await db_session.commit()

    channels = (
        (
            await db_session.execute(
                select(AssistantConversation.channel).where(
                    AssistantConversation.person_id == person.id,
                )
            )
        )
        .scalars()
        .all()
    )
    assert sorted(channels) == [CHANNEL_FEISHU, CHANNEL_WEB]
    assert feishu_chat.channel == CHANNEL_FEISHU


async def test_a_second_turn_remembers_the_first(db_session, enabled_gate, kernel):
    """Working memory is the server's, and it is actually used."""
    person = await _person(db_session)
    await run_web_turn(db_session, person=person, question="第一句", trace_id="t-6")
    await run_web_turn(db_session, person=person, question="第二句", trace_id="t-7")

    second_history = kernel["calls"][-1]
    assert [t["content"] for t in second_history][0] == "第一句"
    assert second_history[-1]["content"] == "第二句"


async def test_a_concurrent_turn_is_refused_rather_than_run(
    db_session, enabled_gate, kernel,
):
    """🔴 Two tabs, one employee.

    Without the lock both turns read the same history, reply out of order, and
    collide on the transcript's ``(conversation_id, seq)`` unique key — which
    the retry loop then burns three attempts on.
    """
    from app.assistant import session as session_store

    person = await _person(db_session)
    held = await session_store.acquire_turn_lock(
        person_id=person.id, chat_id=web_chat_id(person),
    )
    assert held is not None
    try:
        turn = await run_web_turn(
            db_session, person=person, question="同时问的第二句", trace_id="t-8",
        )
    finally:
        await session_store.release_turn_lock(
            person_id=person.id, chat_id=web_chat_id(person), owner=held,
        )

    assert turn.busy is True
    assert turn.message_id is None
    # It must not look like an outage: the page keeps its composer.
    assert turn.turn_failed is False
    assert kernel["calls"] == [], "a refused turn still called the model"


# ── offers ──────────────────────────────────────────────────────────


async def test_an_offer_comes_back_in_the_response_with_the_cards_own_rows(
    db_session, enabled_gate, kernel,
):
    """🔴 The web write path, end to end.

    The proposal is real — token, digest, stored row — and the rows shown are
    the ones ``action.summarise`` produces for the Feishu card. Two surfaces
    that summarise one proposal differently are two surfaces where an employee
    approved different words.
    """
    from app.assistant.models import AssistantActionProposal

    person = await _person(db_session)
    kernel["state"]["dispatch"] = [
        (
            "propose_helix_action",
            {
                "action": "submit_vpn_application",
                "arguments": {
                    "vpn_node": "CN4-HK",
                    "usage_description": "需要访问内网 Grafana",
                    "resource_checklist": "grafana.internal",
                },
            },
        ),
    ]
    kernel["state"]["result"] = FakeCopilotResult(reply="")

    turn = await run_web_turn(
        db_session, person=person, question="帮我申请 VPN", trace_id="t-9",
    )
    await db_session.commit()

    assert len(turn.offers) == 1
    offer = turn.offers[0]
    assert offer["action_key"] == "submit_vpn_application"
    assert offer["rows"], "an offer with no rows shows the employee nothing"
    assert any("CN4-HK" in row["value"] for row in offer["rows"])
    assert all(row["label"] for row in offer["rows"]), "an unlabelled value"
    assert offer["details_url"], "no link to the untruncated snapshot"
    assert offer["is_ticket"] is False
    assert offer["expires_in_minutes"] > 0

    stored = await db_session.get(
        AssistantActionProposal, uuid.UUID(offer["proposal_id"]),
    )
    assert stored is not None
    assert str(stored.person_id) == str(person.id)
    assert stored.status == "pending"
    # Nothing was submitted by being offered.
    assert stored.confirmed_at is None


async def test_the_ticket_guard_applies_on_the_web_page_too(
    db_session, enabled_gate, kernel,
):
    """🔴 A ticket asks a human to do by hand what the platform can do.

    Filing one when a real flow existed is a wrong answer even though the ticket
    itself succeeds — and this guard, like every other one, lived on the path
    only the chat channel used.
    """
    person = await _person(db_session)
    kernel["state"]["dispatch"] = [
        ("propose_create_support_ticket", {"title": "帮我开 VPN", "body": "要连内网。"}),
    ]
    turn = await run_web_turn(
        db_session, person=person, question="帮我开个 VPN", trace_id="t-9b",
    )
    assert turn.offers == [], "a ticket was filed without looking for a flow"


async def test_a_ticket_is_offered_once_the_model_has_actually_looked(
    db_session, enabled_gate, kernel,
):
    """The guard refuses ONCE, then lets a re-affirmed ticket through.

    Without this pair the test above would also pass if the web surface simply
    could not offer a ticket at all, which is the opposite of parity — and it
    pins the deliberate non-deadlock: a guard that could never be satisfied
    would eventually deny a legitimate escalation.
    """
    person = await _person(db_session)
    kernel["state"]["dispatch"] = [
        ("search_helix_actions", {"query": "办公椅 坏了 维修"}),
        # Turned away: the search showed candidates.
        (
            "propose_create_support_ticket",
            {"title": "办公椅坏了", "body": "椅子的气压杆降不下来。"},
        ),
        # Re-affirmed after reading the refusal. This one goes through.
        (
            "propose_create_support_ticket",
            {"title": "办公椅坏了", "body": "椅子的气压杆降不下来，没有对应的流程。"},
        ),
    ]
    kernel["state"]["result"] = FakeCopilotResult(reply="")
    turn = await run_web_turn(
        db_session, person=person, question="我的办公椅坏了", trace_id="t-9c",
    )
    assert len(turn.offers) == 1
    assert turn.offers[0]["is_ticket"] is True


async def test_the_reply_under_an_offer_does_not_narrate_it(
    db_session, enabled_gate, kernel,
):
    """The offer says what will be submitted; a paragraph above it repeats
    what the employee is already reading."""
    person = await _person(db_session)
    kernel["state"]["dispatch"] = [
        (
            "propose_helix_action",
            {
                "action": "submit_vpn_application",
                "arguments": {
                    "vpn_node": "CN4-HK",
                    "usage_description": "需要访问内网 Grafana",
                    "resource_checklist": "grafana.internal",
                },
            },
        ),
    ]
    kernel["state"]["result"] = FakeCopilotResult(reply="我已经把确认卡片发给你了。")

    turn = await run_web_turn(
        db_session, person=person, question="帮我申请 VPN", trace_id="t-10",
    )
    assert turn.offers
    assert "卡片发给你" not in turn.reply
    # And it is not stored blank: an empty turn is dropped from working memory,
    # so the model would see an unanswered request and offer the same thing.
    assert turn.reply.strip()


async def test_claiming_an_offer_that_was_never_made_is_not_an_answer(
    db_session, enabled_gate, kernel,
):
    """🔴 Model text must never manufacture evidence of a side effect."""
    person = await _person(db_session)
    kernel["state"]["result"] = FakeCopilotResult(
        reply="我已经把确认卡片发给你了，等你确认。",
    )
    turn = await run_web_turn(
        db_session, person=person, question="那就帮我创建吧", trace_id="t-11",
    )
    from app.assistant import messages as canned

    assert turn.offers == []
    assert turn.outcome == OUTCOME_FAILED
    # The deterministic sentence, not the model's. And not the chat channel's
    # either: 「确认卡片没发出去」 names a card and a send, on a page where an
    # offer is drawn inline and nothing is ever sent anywhere.
    assert turn.reply == canned.nothing_submitted("zh")


# ── outcomes ────────────────────────────────────────────────────────


async def test_a_provider_outage_is_not_recorded_as_answered(
    db_session, enabled_gate, kernel,
):
    """🔴 An outage in the helpful-rate denominator makes a quality metric read
    low for a reason that has nothing to do with the answers."""
    person = await _person(db_session)
    kernel["state"]["result"] = FakeCopilotResult(reply="", turn_failed=True)
    turn = await run_web_turn(
        db_session, person=person, question="随便问问", trace_id="t-12",
    )
    assert turn.outcome == OUTCOME_FAILED
    assert turn.turn_failed is True
    # The capability is unchanged, so the page must keep its composer.
    assert turn.available is True


async def test_a_switched_off_kernel_says_so_rather_than_pretending(
    db_session, enabled_gate, kernel,
):
    person = await _person(db_session)
    kernel["state"]["result"] = FakeCopilotResult(reply="", available=False)
    turn = await run_web_turn(
        db_session, person=person, question="在吗", trace_id="t-13",
    )
    assert turn.outcome == OUTCOME_UNAVAILABLE
    assert turn.available is False


async def test_a_crashing_turn_still_leaves_a_transcript(
    db_session, enabled_gate, kernel,
):
    """「助手没有回答我」 is precisely the report that has to be checkable."""
    person = await _person(db_session)
    kernel["state"]["raises"] = True
    turn = await run_web_turn(
        db_session, person=person, question="崩了的那一句", trace_id="t-14",
    )
    await db_session.commit()

    assert turn.outcome == OUTCOME_FAILED
    count = (
        await db_session.execute(
            select(func.count())
            .select_from(AssistantMessage)
            .where(AssistantMessage.content == "崩了的那一句")
        )
    ).scalar()
    assert count == 1


# ── rendering ───────────────────────────────────────────────────────


async def test_markdown_survives_and_a_mention_marker_does_not(
    db_session, enabled_gate, kernel,
):
    """🔴 Two halves of one rule: this surface renders Markdown, and it has no
    Feishu delivery layer to resolve a mention token."""
    person = await _person(db_session)
    owner_id = str(uuid.uuid4())
    kernel["state"]["result"] = FakeCopilotResult(
        reply=f"负责人是 [[helix_mention:{owner_id}:李雷]]，**今天**就能处理。",
    )
    turn = await run_web_turn(
        db_session, person=person, question="谁负责这个", trace_id="t-15",
    )
    assert "@李雷" in turn.reply
    assert owner_id not in turn.reply, "an internal person id reached the employee"
    assert "**今天**" in turn.reply, "Markdown was flattened on a surface that renders it"


async def test_the_stored_transcript_is_what_an_escalation_would_quote(
    db_session, enabled_gate, kernel,
):
    person = await _person(db_session)
    await run_web_turn(db_session, person=person, question="第一个问题", trace_id="t-16")
    await db_session.commit()

    quoted = await web_transcript(db_session, person=person)
    assert [t["content"] for t in quoted] == [
        "第一个问题",
        FakeCopilotResult().reply,
    ]


async def test_a_reset_actually_forgets_the_thread(db_session, enabled_gate, kernel):
    from app.assistant.web_turn import reset_web_thread

    person = await _person(db_session)
    await run_web_turn(db_session, person=person, question="记住这句", trace_id="t-17")
    cleared, said = await reset_web_thread(person=person)
    assert cleared is True
    assert said

    await run_web_turn(db_session, person=person, question="重置之后", trace_id="t-18")
    assert [t["content"] for t in kernel["calls"][-1]] == ["重置之后"]


async def test_outcome_answered_is_the_only_thing_worth_remembering(
    db_session, enabled_gate, kernel,
):
    """🔴 Replaying 「助手暂时不可用」 as context poisons every later turn."""
    person = await _person(db_session)
    kernel["state"]["result"] = FakeCopilotResult(reply="", available=False)
    await run_web_turn(db_session, person=person, question="第一句", trace_id="t-19")

    kernel["state"]["result"] = FakeCopilotResult()
    await run_web_turn(db_session, person=person, question="第二句", trace_id="t-20")
    assert [t["content"] for t in kernel["calls"][-1]] == ["第二句"]


async def test_an_answered_turn_reports_the_outcome_it_stored(
    db_session, enabled_gate, kernel,
):
    person = await _person(db_session)
    turn = await run_web_turn(
        db_session, person=person, question="一切正常的一句", trace_id="t-21",
    )
    await db_session.commit()
    assert turn.outcome == OUTCOME_ANSWERED
    stored = await db_session.get(AssistantMessage, uuid.UUID(turn.message_id))
    assert stored is not None
    assert stored.outcome == OUTCOME_ANSWERED
    assert stored.latency_ms is not None


async def test_a_crash_after_an_offer_still_shows_the_offer(
    db_session, enabled_gate, kernel,
):
    """A committed proposal is real, and hiding it would be the lie.

    Everything upstream of the crash already happened: parameters validated,
    the active-employee gate passed, the row written and COMMITTED. The employee
    caused a pending proposal to exist. Dropping it from the response would
    leave them with a failure message and an offer they can neither see nor
    cancel, expiring silently an hour later. The offer states everything it
    would submit, so showing it under a failure discloses nothing extra — and
    the reply is honest that the turn itself did not finish.
    """
    from app.assistant.models import AssistantActionProposal

    person = await _person(db_session)
    kernel["state"]["dispatch"] = [
        (
            "propose_helix_action",
            {
                "action": "submit_vpn_application",
                "arguments": {
                    "vpn_node": "CN4-HK",
                    "usage_description": "需要访问内网 Grafana",
                    "resource_checklist": "grafana.internal",
                },
            },
        ),
    ]
    kernel["state"]["raises"] = True

    turn = await run_web_turn(
        db_session, person=person, question="帮我申请 VPN", trace_id="t-22",
    )
    await db_session.commit()

    assert turn.outcome == OUTCOME_FAILED
    assert len(turn.offers) == 1
    stored = await db_session.get(
        AssistantActionProposal, uuid.UUID(turn.offers[0]["proposal_id"]),
    )
    assert stored is not None
    assert stored.status == "pending"