"""The administrator's conversation-audit surface.

Two obligations. It must answer "what did the assistant tell this employee, and
what did it look at" — including for turns that failed, which is exactly the
report someone would come here to check. And it must not become a way for the
wider admin tier to read employees' own words: the gate here is deliberately
narrower than the platform's usual admin-read tier.
"""
from __future__ import annotations

import uuid

import pytest
from httpx import AsyncClient
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.assistant.feishu_events import handle_message_event
from app.assistant.mcp_bridge import TOOL_CALL, make_dispatch
from app.assistant.models import AssistantConversation
from app.audit.models import AuditLog
from tests.test_assistant.conftest import FakeCopilotResult, message_event

pytestmark = pytest.mark.asyncio

LIST_URL = "/api/v1/assistant/admin/conversations"
STATS_URL = "/api/v1/assistant/admin/stats"


async def _have_a_conversation(db, *, event_id="e1", text="我的电脑呢"):
    return await handle_message_event(
        db, message_event(text=text, event_id=event_id), trace_id=f"trace-{event_id}",
    )


# ── authorization ───────────────────────────────────────────────────


async def test_admin_can_list(
    client: AsyncClient, db_session: AsyncSession, admin_user,
):
    _user, token = admin_user
    response = await client.get(LIST_URL, headers={"Authorization": f"Bearer {token}"})
    assert response.status_code == 200


async def test_auditor_can_list(
    client: AsyncClient, db_session: AsyncSession, auditor_user,
):
    _user, token = auditor_user
    response = await client.get(LIST_URL, headers={"Authorization": f"Bearer {token}"})
    assert response.status_code == 200


async def test_employee_cannot_read_anyones_transcripts(
    client: AsyncClient, db_session: AsyncSession, employee_user,
):
    _user, token = employee_user
    response = await client.get(LIST_URL, headers={"Authorization": f"Bearer {token}"})
    assert response.status_code == 403


async def test_anonymous_is_refused(client: AsyncClient, db_session: AsyncSession):
    assert (await client.get(LIST_URL)).status_code in (401, 403)


# ── reading the record ──────────────────────────────────────────────


async def test_list_shows_the_conversation_with_the_person_resolved(
    client: AsyncClient, db_session: AsyncSession, admin_user,
    enabled_gate, feishu_person, sent, kernel,
):
    await _have_a_conversation(db_session)
    _user, token = admin_user

    response = await client.get(LIST_URL, headers={"Authorization": f"Bearer {token}"})

    rows = response.json()["data"]
    assert len(rows) == 1
    assert rows[0]["person_email"] == feishu_person.work_email
    assert rows[0]["turn_count"] == 1
    assert response.json()["meta"]["total"] == 1


async def test_transcript_returns_both_sides_in_order_with_evidence(
    client: AsyncClient, db_session: AsyncSession, admin_user,
    enabled_gate, feishu_person, sent, kernel,
):
    kernel["state"]["result"] = FakeCopilotResult(tool_calls=["list_my_assets"])
    await _have_a_conversation(db_session)
    conversation = (
        await db_session.execute(select(AssistantConversation))
    ).scalars().one()
    _user, token = admin_user

    response = await client.get(
        f"{LIST_URL}/{conversation.id}", headers={"Authorization": f"Bearer {token}"},
    )

    body = response.json()["data"]
    assert [m["seq"] for m in body["messages"]] == [0, 1]
    assert body["messages"][0]["role"] == "user"
    assert body["messages"][1]["tool_calls"] == ["list_my_assets"]
    assert body["messages"][1]["trace_id"] == "trace-e1"


async def test_a_failed_turn_is_visible_too(
    client: AsyncClient, db_session: AsyncSession, admin_user,
    enabled_gate, feishu_person, sent, kernel,
):
    """"The assistant never answered me" has to be checkable."""
    kernel["state"]["raises"] = True
    await _have_a_conversation(db_session)
    _user, token = admin_user

    response = await client.get(
        LIST_URL, params={"outcome": "failed"},
        headers={"Authorization": f"Bearer {token}"},
    )

    assert len(response.json()["data"]) == 1


async def test_filtering_by_outcome_excludes_the_others(
    client: AsyncClient, db_session: AsyncSession, admin_user,
    enabled_gate, feishu_person, sent, kernel,
):
    await _have_a_conversation(db_session)
    _user, token = admin_user

    matched = await client.get(
        LIST_URL, params={"outcome": "answered"},
        headers={"Authorization": f"Bearer {token}"},
    )
    unmatched = await client.get(
        LIST_URL, params={"outcome": "failed"},
        headers={"Authorization": f"Bearer {token}"},
    )

    assert len(matched.json()["data"]) == 1
    assert unmatched.json()["data"] == []


async def test_filtering_by_tool_finds_the_conversation_that_used_it(
    client: AsyncClient, db_session: AsyncSession, admin_user,
    enabled_gate, feishu_person, sent, kernel,
):
    kernel["state"]["result"] = FakeCopilotResult(tool_calls=["list_my_assets"])
    await _have_a_conversation(db_session)
    _user, token = admin_user

    hit = await client.get(
        LIST_URL, params={"tool": "list_my_assets"},
        headers={"Authorization": f"Bearer {token}"},
    )
    miss = await client.get(
        LIST_URL, params={"tool": "list_my_software_seats"},
        headers={"Authorization": f"Bearer {token}"},
    )

    assert len(hit.json()["data"]) == 1
    assert miss.json()["data"] == []


async def test_a_multi_turn_conversation_appears_once(
    client: AsyncClient, db_session: AsyncSession, admin_user,
    enabled_gate, feishu_person, sent, kernel,
):
    """Filtering joins on messages; three matches must not become three rows."""
    await _have_a_conversation(db_session, event_id="e1")
    await _have_a_conversation(db_session, event_id="e2")
    await _have_a_conversation(db_session, event_id="e3")
    _user, token = admin_user

    response = await client.get(
        LIST_URL, params={"outcome": "answered"},
        headers={"Authorization": f"Bearer {token}"},
    )

    assert len(response.json()["data"]) == 1
    assert response.json()["meta"]["total"] == 1


async def test_the_list_total_matches_the_rows_it_returns(
    client: AsyncClient, db_session: AsyncSession, admin_user,
    enabled_gate, feishu_person, sent, kernel,
):
    """The count query does not join; the row query does.

    Today a mismatch is impossible: the FK to ``identity_persons`` plus the
    platform's soft-delete-only policy mean every conversation always has a
    person row (an attempt to orphan one raises 1452). The row query still uses
    an OUTER join so the two can never disagree if that ever changes — this
    test pins the property, and the comment records why it cannot currently
    fail.
    """
    await _have_a_conversation(db_session, event_id="e1")
    await _have_a_conversation(db_session, event_id="e2")
    _user, token = admin_user

    body = (
        await client.get(LIST_URL, headers={"Authorization": f"Bearer {token}"})
    ).json()

    assert body["meta"]["total"] == len(body["data"])
    assert all(row["person_email"] for row in body["data"])


async def test_a_wildcard_in_the_tool_filter_is_not_a_wildcard(
    client: AsyncClient, db_session: AsyncSession, admin_user,
    enabled_gate, feishu_person, sent, kernel,
):
    """The filter is an operator-typed string, not a LIKE pattern."""
    kernel["state"]["result"] = FakeCopilotResult(tool_calls=["list_my_assets"])
    await _have_a_conversation(db_session)
    _user, token = admin_user

    response = await client.get(
        LIST_URL, params={"tool": "list_my_%"},
        headers={"Authorization": f"Bearer {token}"},
    )

    assert response.json()["data"] == []


async def test_unknown_conversation_is_a_404(
    client: AsyncClient, db_session: AsyncSession, admin_user,
):

    _user, token = admin_user
    response = await client.get(
        f"{LIST_URL}/{uuid.uuid4()}", headers={"Authorization": f"Bearer {token}"},
    )
    assert response.status_code == 404


# ── counters ────────────────────────────────────────────────────────


async def test_stats_counts_turns_and_outcomes(
    client: AsyncClient, db_session: AsyncSession, admin_user,
    enabled_gate, feishu_person, sent, kernel,
):
    await _have_a_conversation(db_session)
    _user, token = admin_user

    body = (
        await client.get(STATS_URL, headers={"Authorization": f"Bearer {token}"})
    ).json()["data"]

    assert body["conversations"] == 1
    assert body["people"] == 1
    assert body["outcomes"]["answered"] == 1


async def test_a_refused_tool_call_is_audited_and_counted(
    client: AsyncClient, db_session: AsyncSession, admin_user, feishu_person,
):
    """A refusal that only reached the log is a refusal nobody can audit."""
    dispatch = make_dispatch(person=feishu_person)
    await dispatch(
        TOOL_CALL, {"name": "revoke_api_key", "arguments": {}},
        db_session, feishu_person,
    )

    rows = (
        await db_session.execute(
            select(AuditLog).where(AuditLog.action == "assistant.tool.refused"),
        )
    ).scalars().all()
    assert len(rows) == 1
    assert rows[0].actor_user_id == feishu_person.id
    assert rows[0].detail["tool"] == "revoke_api_key"

    _user, token = admin_user
    body = (
        await client.get(STATS_URL, headers={"Authorization": f"Bearer {token}"})
    ).json()["data"]
    assert body["refused_tool_calls"] == 1


async def test_stats_window_is_bounded(
    client: AsyncClient, db_session: AsyncSession, admin_user,
):
    _user, token = admin_user
    too_wide = await client.get(
        STATS_URL, params={"days": 4000}, headers={"Authorization": f"Bearer {token}"},
    )
    assert too_wide.status_code == 422