"""The funnel, and the two things it is not allowed to count.

Both of these were real behaviours of the numbers this replaces:

* a conversation that produced nothing counted as a success ("no ticket was
  created, therefore it was auto-resolved");
* a request nobody rated counted as a success by omission.

There is also a guard here that has nothing to do with counting: the system
prompt names a tool by string, and a renamed tool would leave the model calling
something that does not exist — which reaches the employee as the assistant
refusing to check on their request.
"""
from __future__ import annotations

import uuid

import pytest
from sqlalchemy.ext.asyncio import AsyncSession

from app.assistant import admin_service
from app.assistant.models import (
    AssistantActionProposal,
    AssistantObjectLink,
)
from app.assistant.prompt import SUBMISSIONS_TOOL_NAME, feishu_system_prompt
from app.core.coordinators import assistant_outcome_coordinator as oc
from app.core.coordinators.assistant_support_coordinator import (
    OUTCOME_SELF_SOLVED,
    OUTCOME_SYSTEM_VERIFIED,
)
from app.domains.support.models import (
    ORIGIN_ASSISTANT,
    REQUEST_ACTIVE,
    REQUEST_SELF_RESOLVED,
    SupportServiceRequest,
)
from tests.test_assistant.conftest import make_feishu_person

_async = pytest.mark.asyncio


# ── the prompt names a real tool ─────────────────────────────────────


def test_the_prompt_names_a_tool_that_is_actually_registered() -> None:
    """A string in a prompt is not a reference — nothing links it to the
    registry. Rename the tool and the model calls a name that resolves to
    nothing, which the employee experiences as being refused."""
    from app.mcp import tools as _tools  # noqa: F401 — force registration
    from app.mcp.server import mcp_server

    registered = set(mcp_server._tool_manager._tools)  # noqa: SLF001
    assert SUBMISSIONS_TOOL_NAME in registered, (
        f"the system prompt tells the model to call {SUBMISSIONS_TOOL_NAME!r}, "
        "which is not a registered MCP tool"
    )
    assert SUBMISSIONS_TOOL_NAME in feishu_system_prompt(lang="zh")


def test_the_prompt_says_submitting_is_not_finishing() -> None:
    """The specific failure being prevented: repeating an earlier "已提交" as
    though it meant "已完成"."""
    text = feishu_system_prompt(lang="zh")
    assert "provisioning" in text and "effective" in text
    assert "earlier in this conversation" in text


# ── counting ─────────────────────────────────────────────────────────


async def _proposal(db, person, *, status, attempts=0, error=None, miss=None):
    row = AssistantActionProposal(
        person_id=person.id, action_key="create_support_ticket", params={},
        params_hash=uuid.uuid4().hex * 2, status=status,
        attempt_count=attempts, last_error_code=error, link_miss_reason=miss,
    )
    db.add(row)
    await db.flush()
    return row


@_async
async def test_pressed_and_executed_are_counted_separately(
    db_session: AsyncSession,
) -> None:
    """Production held 14 proposals reading as unexecuted, two of which had
    been pressed. Without ``attempt_count`` those two were invisible, and
    "offered but nobody used it" absorbed them."""
    person = await make_feishu_person(db_session)
    await _proposal(db_session, person, status="pending")
    await _proposal(
        db_session, person, status="pending", attempts=1, error="ToolError",
    )
    await _proposal(db_session, person, status="executed", attempts=1)
    await db_session.commit()

    funnel = await admin_service.submission_funnel(db_session, days=30)
    assert funnel["proposals"]["offered"] == 3
    assert funnel["proposals"]["pressed"] == 2
    assert funnel["proposals"]["press_failed"] == 1
    assert funnel["proposals"]["executed"] == 1


@_async
async def test_a_link_miss_is_counted_and_not_hidden_inside_zero_submissions(
    db_session: AsyncSession,
) -> None:
    """A gap in the outcome contract and "nobody submitted anything" look
    identical on a funnel that only counts links."""
    person = await make_feishu_person(db_session)
    await _proposal(
        db_session, person, status="executed", attempts=1, miss="no_locator",
    )
    await db_session.commit()

    funnel = await admin_service.submission_funnel(db_session, days=30)
    assert funnel["submissions"]["linked"] == 0
    assert funnel["submissions"]["link_misses"] == {"no_locator": 1}


@_async
async def test_verified_and_taken_on_their_word_are_never_added_together(
    db_session: AsyncSession,
) -> None:
    person = await make_feishu_person(db_session)
    for outcome in (OUTCOME_SYSTEM_VERIFIED, OUTCOME_SELF_SOLVED, OUTCOME_SELF_SOLVED):
        db_session.add(
            SupportServiceRequest(
                requester_id=person.id, origin=ORIGIN_ASSISTANT,
                status=REQUEST_SELF_RESOLVED, outcome=outcome,
            )
        )
    await db_session.commit()

    funnel = await admin_service.submission_funnel(db_session, days=30)
    assert funnel["requests"]["closed_verified"] == 1
    assert funnel["requests"]["closed_on_word"] == 2
    assert "closed_total" not in funnel["requests"], (
        "a combined total invites exactly the conflation these two exist to prevent"
    )


@_async
async def test_needs_attention_is_visible_and_is_not_a_closure(
    db_session: AsyncSession,
) -> None:
    from app.core.coordinators.assistant_support_coordinator import (
        OUTCOME_ACTION_NEEDS_ATTENTION,
    )

    person = await make_feishu_person(db_session)
    db_session.add(
        SupportServiceRequest(
            requester_id=person.id, origin=ORIGIN_ASSISTANT,
            status=REQUEST_ACTIVE, outcome=OUTCOME_ACTION_NEEDS_ATTENTION,
        )
    )
    await db_session.commit()

    funnel = await admin_service.submission_funnel(db_session, days=30)
    assert funnel["requests"]["needs_attention"] == 1
    assert funnel["requests"]["closed_verified"] == 0
    assert funnel["requests"]["closed_on_word"] == 0


@_async
async def test_phases_are_broken_out_so_effective_is_not_the_only_number(
    db_session: AsyncSession,
) -> None:
    person = await make_feishu_person(db_session)
    for phase in (
        oc.PHASE_EFFECTIVE, oc.PHASE_EFFECTIVE,
        oc.PHASE_REJECTED, oc.PHASE_FAILED, oc.PHASE_PENDING_APPROVAL,
    ):
        proposal = await _proposal(db_session, person, status="executed", attempts=1)
        db_session.add(
            AssistantObjectLink(
                proposal_id=proposal.id, person_id=person.id,
                action_key="create_support_ticket",
                resource_type="support_ticket", resource_id=str(uuid.uuid4()),
                resolver_key="support_ticket", phase=phase,
            )
        )
    await db_session.commit()

    funnel = await admin_service.submission_funnel(db_session, days=30)
    assert funnel["submissions"]["linked"] == 5
    assert funnel["submissions"]["effective"] == 2
    assert funnel["submissions"]["rejected"] == 1
    assert funnel["submissions"]["failed"] == 1
    assert funnel["submissions"]["by_phase"][oc.PHASE_PENDING_APPROVAL] == 1


# ── 简报第 6 条剩下的五个指标 ────────────────────────────────────


async def _turn_audit(db, person, *, is_card_reply=None, resolved=None):
    from app.audit.models import AuditLog

    detail = {}
    if is_card_reply is not None:
        detail["is_card_reply"] = is_card_reply
    if resolved is not None:
        detail["card_context_resolved"] = resolved
    db.add(
        AuditLog(
            id=uuid.uuid4(), trace_id=uuid.uuid4(), actor_type="user",
            actor_user_id=person.id, action="assistant.message.received",
            status="success", detail=detail,
        )
    )


@_async
async def test_card_reply_resolution_is_two_numbers_not_one(
    db_session: AsyncSession,
) -> None:
    """"Nobody replies to cards" and "people reply but we cannot tell which
    card" look identical from outside and have opposite fixes — the first is
    discoverability, the second is plumbing."""
    person = await make_feishu_person(db_session)
    await _turn_audit(db_session, person, is_card_reply=True, resolved=True)
    await _turn_audit(db_session, person, is_card_reply=True, resolved=True)
    await _turn_audit(db_session, person, is_card_reply=True, resolved=False)
    await _turn_audit(db_session, person)  # an ordinary message
    await db_session.commit()

    f = await admin_service.submission_funnel(db_session, days=30)
    assert f["engagement"]["turns"] == 4
    assert f["engagement"]["card_replies"] == 3
    assert f["engagement"]["card_replies_resolved"] == 2
    assert f["engagement"]["card_reply_resolution_rate"] == round(2 / 3, 3)


@_async
async def test_a_json_true_is_matched_and_not_silently_zero(
    db_session: AsyncSession,
) -> None:
    """The trap this codebase has hit before: comparing a JSON boolean by
    casting a Python ``True`` through SQLAlchemy's JSON type re-encodes it as
    the JSON *string* ``"true"`` and matches nothing — the count comes back 0
    and reads as "no card replies", which is a plausible answer."""
    person = await make_feishu_person(db_session)
    await _turn_audit(db_session, person, is_card_reply=True, resolved=True)
    await db_session.commit()

    f = await admin_service.submission_funnel(db_session, days=30)
    assert f["engagement"]["card_replies"] == 1, (
        "a JSON true did not match — the count would read as 'nobody replies'"
    )


@_async
async def test_no_data_reports_none_and_never_zero_percent(
    db_session: AsyncSession,
) -> None:
    """A 0% rate and "nothing happened yet" are different findings, and showing
    the first for the second invents a problem."""
    f = await admin_service.submission_funnel(db_session, days=30)
    assert f["engagement"]["card_reply_resolution_rate"] is None
    assert f["engagement"]["escalation_rate"] is None
    assert f["knowledge_gaps"]["reopen_rate"] is None
    assert f["timing"]["median_hours_to_result"] is None
    assert f["timing"]["p90_hours_to_result"] is None


@_async
async def test_time_to_result_measures_only_what_settled(
    db_session: AsyncSession,
) -> None:
    """Averaging in the ones still in flight would make the number IMPROVE
    every time something gets stuck."""
    from datetime import UTC, datetime, timedelta

    from app.domains.support.models import ORIGIN_ASSISTANT, REQUEST_ACTIVE

    person = await make_feishu_person(db_session)
    now = datetime.now(UTC)

    async def _settled(hours: int | None, phase: str):
        req = SupportServiceRequest(
            requester_id=person.id, origin=ORIGIN_ASSISTANT,
            status=REQUEST_ACTIVE, created_at=now - timedelta(hours=48),
        )
        db_session.add(req)
        proposal = await _proposal(db_session, person, status="executed", attempts=1)
        db_session.add(
            AssistantObjectLink(
                proposal_id=proposal.id, person_id=person.id,
                service_request_id=req.id, action_key="create_support_ticket",
                resource_type="support_ticket", resource_id=str(uuid.uuid4()),
                resolver_key="support_ticket", phase=phase,
                phase_at=(now - timedelta(hours=48 - hours)) if hours else None,
            )
        )

    await _settled(2, oc.PHASE_EFFECTIVE)
    await _settled(10, oc.PHASE_REJECTED)
    # Still in flight — and it HAS a ``phase_at``, because ``resolve_and_cache``
    # stamps that on every read, not only when something settles. An earlier
    # version of this test gave it ``phase_at=None`` and therefore passed
    # without pinning anything: the query's own null-check was doing the work.
    # In production this row exists and would contribute a "time to result"
    # that is really "time since somebody last looked".
    await _settled(1, oc.PHASE_PENDING_APPROVAL)
    await db_session.commit()

    f = await admin_service.submission_funnel(db_session, days=30)
    assert f["timing"]["settled_count"] == 2, (
        "an in-flight submission was counted in time-to-result"
    )
    assert f["timing"]["median_hours_to_result"] == 6.0


@_async
async def test_gap_reopen_rate_comes_off_the_audit_spine(
    db_session: AsyncSession,
) -> None:
    """A closure a real employee later contradicted. Derived from audit rather
    than a counter, so it can be recomputed over history."""
    from app.audit.models import AuditLog

    person = await make_feishu_person(db_session)
    for action, n in (("assistant.knowledge_gap.resolve", 4),
                      ("assistant.knowledge_gap.reopen", 1)):
        for _ in range(n):
            db_session.add(
                AuditLog(
                    id=uuid.uuid4(), trace_id=uuid.uuid4(), actor_type="user",
                    actor_user_id=person.id, action=action, status="success",
                    detail={},
                )
            )
    await db_session.commit()

    f = await admin_service.submission_funnel(db_session, days=30)
    assert f["knowledge_gaps"]["resolved"] == 4
    assert f["knowledge_gaps"]["reopened"] == 1
    assert f["knowledge_gaps"]["reopen_rate"] == 0.25