"""转人工:分类 → 找人 → 建群,以及每一种「没拉到人」都必须说实话。

用户的要求是:「转人工应该根据问题 AI 分一个类,如果是权限/软件/制度/入离职之类
的拉孟凡生和提问者建一个群,把聊天转过来;如果是云、安全、平台相关的拉冯帅」,
并且「这种哪些类型的可以在后台配置」。

这个文件钉住四件事:

1. **分类是闭集。** 模型只能从已存在的 `category_key` 里选;它发明一个键等同于
   「没分类出来」,走兜底路由。一个发明出来的键会静默变成「没人负责」。
2. **路由是数据,不是代码。** 谁负责哪一类由表决定,所以改派不需要发布。
3. **降级必须说实话。** 没指派负责人 / 建群失败 / 功能关掉 —— 三种都仍然建工单,
   而且**不能**说「已经拉了人」。拉不到人时说拉到了,是这条链路最坏的失败。
4. **幂等。** 建群是不可撤销的外部动作,而飞书会重投它判为慢的回调。
"""
from __future__ import annotations

import uuid

import pytest
from sqlalchemy.ext.asyncio import AsyncSession

from app.assistant import handoff
from app.assistant.models import (
    AssistantConversation,
    AssistantEscalationRoute,
    AssistantMessage,
)
from app.identity.person_models import IdentityPerson


async def _person(db: AsyncSession, name: str) -> IdentityPerson:
    person = IdentityPerson(
        id=uuid.uuid4(),
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
        display_name=name,
        work_email=f"h-{uuid.uuid4().hex[:8]}@pingcap.com",
        employment_status="active",
    )
    db.add(person)
    await db.flush()
    return person


async def _route(
    db: AsyncSession,
    *,
    key: str,
    owner: IdentityPerson | None = None,
    is_default: bool = False,
    enabled: bool = True,
) -> AssistantEscalationRoute:
    row = AssistantEscalationRoute(
        category_key=key,
        display_name={"zh": f"{key}-中", "en": f"{key}-en", "ja": f"{key}-ja"},
        hint=f"{key} 相关",
        owner_person_id=owner.id if owner else None,
        is_default=is_default,
        enabled=enabled,
        sort_order=10,
    )
    db.add(row)
    await db.flush()
    return row


async def _ticket(db: AsyncSession, person: IdentityPerson):
    """一张真工单。

    ``hand_off`` 现在收的是**工单对象**而不是一个裸 id:它要拿工单号给群命名,
    并且要把工单指派给被拉进群的那个人。传 ``uuid4()`` 的旧写法能通过是因为
    那时它只是被原样写进台账 —— 而「指派」正是那一版缺的一步。
    """
    from app.domains.support.models import SupportTicket

    ticket = SupportTicket(
        id=uuid.uuid4(),
        ticket_number=f"HELP-{uuid.uuid4().hex[:10]}",
        requester_id=person.id,
        title="转人工的那条问题",
        body="正文",
        category="other",
        priority="normal",
    )
    db.add(ticket)
    await db.flush()
    return ticket


async def _message(db: AsyncSession, person: IdentityPerson) -> AssistantMessage:
    conv = AssistantConversation(
        id=uuid.uuid4(), person_id=person.id, channel="feishu",
    )
    db.add(conv)
    await db.flush()
    msg = AssistantMessage(
        id=uuid.uuid4(), conversation_id=conv.id, seq=1,
        role="assistant", content="我没能回答这个问题。",
    )
    db.add(msg)
    await db.flush()
    return msg


class _FakeClient:
    """记下建群/发消息/转发都被怎么调用的。"""

    def __init__(self, *, fail_create: bool = False, fail_forward: bool = False):
        self.fail_create = fail_create
        self.fail_forward = fail_forward
        self.created: list[dict] = []
        self.texts: list[str] = []
        self.cards: list[dict] = []
        self.updates: list[str] = []
        self.invited: list[str] = []
        self.forwarded: list[str] = []

    async def resolve_open_id_by_email(self, email: str) -> str:
        return f"ou_{email.split('@')[0]}"

    async def create_group_chat(self, *, name, member_open_ids, description=""):
        if self.fail_create:
            raise RuntimeError("feishu says no")
        self.created.append(
            {"name": name, "members": member_open_ids, "description": description},
        )
        return "oc_fake_chat"

    async def send_text_to_chat(
        self, *, chat_id: str, text: str, idempotency_key: str | None = None,
    ) -> dict:
        self.texts.append(text)
        return {"message_id": f"om_t{len(self.texts)}"}

    async def send_card_to_chat(
        self, *, chat_id: str, card: dict, idempotency_key: str | None = None,
    ) -> dict:
        self.cards.append(card)
        return {"message_id": f"om_c{len(self.cards)}"}

    async def update_card_message(self, *, message_id: str, card: dict) -> None:
        self.updates.append(message_id)

    async def add_chat_members(self, *, chat_id: str, member_open_ids: list) -> None:
        self.invited.extend(member_open_ids)

    async def forward_message_to_chat(self, *, message_id: str, chat_id: str) -> None:
        if self.fail_forward:
            raise RuntimeError("too old to forward")
        self.forwarded.append(message_id)


@pytest.fixture
def fake_chat(monkeypatch):
    client = _FakeClient()

    async def _build(_db):
        return client

    from app.assistant import channel

    monkeypatch.setattr(channel, "build_client", _build)
    return client


@pytest.fixture
def no_llm(monkeypatch):
    """默认让分类不可用 —— 大多数用例测的是路由与降级,不是模型。

    分类不可用会走兜底路由,这本身就是一条必须成立的路径(AI 挂了不该让
    「找个人」这件事也挂掉)。
    """
    async def _none(_db, _system, _prompt, timeout=0.0):
        return None

    import app.domains.support.copilot as copilot

    monkeypatch.setattr(copilot, "_one_shot_llm", _none)


def _classifier(payload: str):
    async def _fake(_db, _system, _prompt, timeout=0.0):
        return payload

    return _fake


# ── 一、分类是闭集 ──────────────────────────────────────────────

@pytest.mark.asyncio
async def test_the_model_can_only_pick_an_existing_category(
    db_session: AsyncSession, monkeypatch,
) -> None:
    """模型发明一个键 = 没分类出来。

    一个发明出来的 category_key 在路由表里查不到,会静默变成「没人负责」——
    所以它必须在这一层就被判为无效,而不是被当成一个真的类别传下去。
    """
    import app.domains.support.copilot as copilot

    monkeypatch.setattr(
        copilot, "_one_shot_llm",
        _classifier('{"category": "quantum_computing", "confidence": 0.99, '
                    '"reason": "编的"}'),
    )
    routes = [await _route(db_session, key="cloud")]
    result = await handoff.classify(
        db_session, question="云账号进不去", answer="不知道", routes=routes,
    )
    assert result.category_key is None
    assert result.confidence == 0.0
    assert "unusable" in result.reason


@pytest.mark.asyncio
async def test_a_valid_category_is_taken(
    db_session: AsyncSession, monkeypatch,
) -> None:
    import app.domains.support.copilot as copilot

    monkeypatch.setattr(
        copilot, "_one_shot_llm",
        _classifier('{"category": "cloud", "confidence": 0.9, "reason": "云账号"}'),
    )
    routes = [await _route(db_session, key="cloud")]
    result = await handoff.classify(
        db_session, question="云账号进不去", answer="不知道", routes=routes,
    )
    assert result.category_key == "cloud"
    assert result.confidence == pytest.approx(0.9)


@pytest.mark.asyncio
async def test_a_fenced_json_reply_is_still_parsed(
    db_session: AsyncSession, monkeypatch,
) -> None:
    """模型爱加 ```json 围栏。围栏不该让分类失败。"""
    import app.domains.support.copilot as copilot

    monkeypatch.setattr(
        copilot, "_one_shot_llm",
        _classifier('```json\n{"category": "cloud", "confidence": 0.8, '
                    '"reason": "x"}\n```'),
    )
    routes = [await _route(db_session, key="cloud")]
    result = await handoff.classify(
        db_session, question="q", answer="a", routes=routes,
    )
    assert result.category_key == "cloud"


@pytest.mark.asyncio
async def test_an_unavailable_classifier_does_not_raise(
    db_session: AsyncSession, no_llm,
) -> None:
    """AI 挂了不该让「找个人」这件事也挂掉。"""
    routes = [await _route(db_session, key="cloud")]
    result = await handoff.classify(
        db_session, question="q", answer="a", routes=routes,
    )
    assert result.category_key is None


@pytest.mark.asyncio
async def test_low_confidence_falls_back_to_the_default_route(
    db_session: AsyncSession, monkeypatch, fake_chat: _FakeClient,
) -> None:
    """置信度不够就走兜底,但**分类结果照样记进台账**供复核。"""
    import app.domains.support.copilot as copilot

    monkeypatch.setattr(
        copilot, "_one_shot_llm",
        _classifier('{"category": "cloud", "confidence": 0.2, "reason": "拿不准"}'),
    )
    employee = await _person(db_session, "员工")
    cloud_owner = await _person(db_session, "云负责人")
    fallback_owner = await _person(db_session, "兜底负责人")
    await _route(db_session, key="cloud", owner=cloud_owner)
    await _route(db_session, key="other", owner=fallback_owner, is_default=True)
    msg = await _message(db_session, employee)

    result = await handoff.hand_off(
        db_session, person=employee, message_id=msg.id,
        question="云账号", answer="不知道", feishu_message_id="om_x",
        lang="zh", ticket=await _ticket(db_session, employee),
    )

    assert result.owner_name == "兜底负责人", "低置信度没有走兜底路由"
    row = await handoff.existing_handoff(db_session, message_id=msg.id)
    assert row is not None
    # 分类结果仍然要留痕 —— 否则「模型为什么老是拿不准」无从查。
    assert row.classified_confidence == pytest.approx(0.2)


# ── 二、建群与转发 ──────────────────────────────────────────────

@pytest.mark.asyncio
async def test_it_opens_a_group_with_both_people_and_forwards_the_answer(
    db_session: AsyncSession, no_llm, fake_chat: _FakeClient,
) -> None:
    employee = await _person(db_session, "提问的人")
    owner = await _person(db_session, "负责人")
    await _route(db_session, key="other", owner=owner, is_default=True)
    msg = await _message(db_session, employee)

    result = await handoff.hand_off(
        db_session, person=employee, message_id=msg.id,
        question="打印机坏了", answer="我不知道", feishu_message_id="om_answer",
        lang="zh", ticket=await _ticket(db_session, employee),
    )

    assert result.chat_created is True
    assert result.owner_name == "负责人"
    assert len(fake_chat.created) == 1
    # 两个人都在群里。只有一个人的群比没有群更糟 —— 员工会对着空房间等。
    assert len(fake_chat.created[0]["members"]) == 2
    # 开场白说清是谁在问、为什么拉群。
    assert "提问的人" in fake_chat.texts[0]
    # 那条回答被原样转发,而不是重新排版
    assert fake_chat.forwarded == ["om_answer"]


@pytest.mark.asyncio
async def test_when_forwarding_fails_the_answer_is_pasted_instead(
    db_session: AsyncSession, no_llm, fake_chat: _FakeClient,
) -> None:
    """群里**必须**有那段回答,否则负责人得去问「他到底看到了什么」。"""
    fake_chat.fail_forward = True
    employee = await _person(db_session, "员工")
    owner = await _person(db_session, "负责人")
    await _route(db_session, key="other", owner=owner, is_default=True)
    msg = await _message(db_session, employee)

    result = await handoff.hand_off(
        db_session, person=employee, message_id=msg.id,
        question="q", answer="这是助手当时的回答", feishu_message_id="om_old",
        lang="zh", ticket=await _ticket(db_session, employee),
    )

    assert result.chat_created is True
    assert any("这是助手当时的回答" in t for t in fake_chat.texts)


@pytest.mark.asyncio
async def test_with_no_feishu_message_id_the_answer_is_still_in_the_group(
    db_session: AsyncSession, no_llm, fake_chat: _FakeClient,
) -> None:
    """``open_message_id`` 是个没被实测过的可选字段 —— 拿不到也要有内容。"""
    employee = await _person(db_session, "员工")
    owner = await _person(db_session, "负责人")
    await _route(db_session, key="other", owner=owner, is_default=True)
    msg = await _message(db_session, employee)

    await handoff.hand_off(
        db_session, person=employee, message_id=msg.id,
        question="q", answer="回答正文", feishu_message_id=None,
        lang="zh", ticket=await _ticket(db_session, employee),
    )
    assert fake_chat.forwarded == []
    assert any("回答正文" in t for t in fake_chat.texts)


# ── 三、降级必须说实话 ──────────────────────────────────────────

@pytest.mark.asyncio
async def test_no_owner_means_no_chat_and_it_says_so(
    db_session: AsyncSession, no_llm, fake_chat: _FakeClient,
) -> None:
    """一条启用了但还没指派负责人的路由是正常状态(刚建好、上一任刚离职)。

    此时不能拉一个空群,也不能假装拉到了人。
    """
    employee = await _person(db_session, "员工")
    await _route(db_session, key="other", owner=None, is_default=True)
    msg = await _message(db_session, employee)

    result = await handoff.hand_off(
        db_session, person=employee, message_id=msg.id,
        question="q", answer="a", feishu_message_id="om_x",
        lang="zh", ticket=await _ticket(db_session, employee),
    )

    assert result.chat_created is False
    assert result.degraded_reason == "no_owner"
    assert fake_chat.created == []
    row = await handoff.existing_handoff(db_session, message_id=msg.id)
    assert row is not None and row.degraded_reason == "no_owner"


@pytest.mark.asyncio
async def test_a_failed_chat_is_recorded_as_degraded(
    db_session: AsyncSession, no_llm, fake_chat: _FakeClient,
) -> None:
    fake_chat.fail_create = True
    employee = await _person(db_session, "员工")
    owner = await _person(db_session, "负责人")
    await _route(db_session, key="other", owner=owner, is_default=True)
    msg = await _message(db_session, employee)

    result = await handoff.hand_off(
        db_session, person=employee, message_id=msg.id,
        question="q", answer="a", feishu_message_id="om_x",
        lang="zh", ticket=await _ticket(db_session, employee),
    )

    assert result.chat_created is False
    assert result.degraded_reason == "chat_failed"


@pytest.mark.asyncio
async def test_the_feature_switch_degrades_without_removing_the_button(
    db_session: AsyncSession, no_llm, fake_chat: _FakeClient,
) -> None:
    """关掉之后按钮仍在,只是走降级 —— 按钮消失会让员工以为功能坏了。"""
    from app.integrations.models import PlatformSetting

    db_session.add(PlatformSetting(key=handoff.SETTING_ENABLED, value=False))
    await db_session.flush()
    employee = await _person(db_session, "员工")
    owner = await _person(db_session, "负责人")
    await _route(db_session, key="other", owner=owner, is_default=True)
    msg = await _message(db_session, employee)

    result = await handoff.hand_off(
        db_session, person=employee, message_id=msg.id,
        question="q", answer="a", feishu_message_id="om_x",
        lang="zh", ticket=await _ticket(db_session, employee),
    )
    assert result.degraded_reason == "disabled"
    assert fake_chat.created == []


@pytest.mark.asyncio
async def test_a_disabled_route_is_not_used(
    db_session: AsyncSession, no_llm, fake_chat: _FakeClient,
) -> None:
    """关掉的路由不参与 —— 它的负责人不该被拉进群。"""
    employee = await _person(db_session, "员工")
    off_owner = await _person(db_session, "已关闭这一类的负责人")
    await _route(db_session, key="cloud", owner=off_owner, enabled=False)
    msg = await _message(db_session, employee)

    result = await handoff.hand_off(
        db_session, person=employee, message_id=msg.id,
        question="q", answer="a", feishu_message_id="om_x",
        lang="zh", ticket=await _ticket(db_session, employee),
    )
    # 没有启用的路由、也没有兜底 ⇒ 降级
    assert result.degraded_reason == "no_owner"
    assert fake_chat.created == []


# ── 四、幂等 ─────────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_two_taps_get_one_group(
    db_session: AsyncSession, no_llm, fake_chat: _FakeClient,
) -> None:
    """建群是不可撤销的外部动作,而飞书会重投它判为慢的回调。"""
    employee = await _person(db_session, "员工")
    owner = await _person(db_session, "负责人")
    await _route(db_session, key="other", owner=owner, is_default=True)
    msg = await _message(db_session, employee)

    first = await handoff.hand_off(
        db_session, person=employee, message_id=msg.id,
        question="q", answer="a", feishu_message_id="om_x",
        lang="zh", ticket=await _ticket(db_session, employee),
    )
    assert first.chat_created is True
    prior = await handoff.existing_handoff(db_session, message_id=msg.id)
    assert prior is not None, "第一次没有留下台账行,幂等就无从判断"


@pytest.mark.asyncio
async def test_the_ledger_records_what_the_model_decided(
    db_session: AsyncSession, monkeypatch, fake_chat: _FakeClient,
) -> None:
    """没有这张台账,「分类准不准」就只能靠人凭印象说。"""
    import app.domains.support.copilot as copilot

    monkeypatch.setattr(
        copilot, "_one_shot_llm",
        _classifier('{"category": "cloud", "confidence": 0.95, "reason": "云账号问题"}'),
    )
    employee = await _person(db_session, "员工")
    owner = await _person(db_session, "云负责人")
    route = await _route(db_session, key="cloud", owner=owner)
    msg = await _message(db_session, employee)
    ticket = await _ticket(db_session, employee)

    await handoff.hand_off(
        db_session, person=employee, message_id=msg.id,
        question="云账号进不去", answer="不知道", feishu_message_id="om_x",
        lang="zh", ticket=ticket,
    )

    row = await handoff.existing_handoff(db_session, message_id=msg.id)
    assert row is not None
    assert row.classified_category == "cloud"
    assert row.classified_confidence == pytest.approx(0.95)
    assert row.classified_reason == "云账号问题"
    assert row.route_id == route.id
    assert row.ticket_id == ticket.id
    assert row.degraded_reason is None


# ── 五、路由显示名 ──────────────────────────────────────────────

@pytest.mark.asyncio
async def test_the_group_name_uses_the_readers_language(
    db_session: AsyncSession, no_llm, fake_chat: _FakeClient,
) -> None:
    """群名是员工和负责人都看得见的东西。"""
    employee = await _person(db_session, "员工")
    owner = await _person(db_session, "负责人")
    await _route(db_session, key="cloud", owner=owner, is_default=True)
    msg = await _message(db_session, employee)

    await handoff.hand_off(
        db_session, person=employee, message_id=msg.id,
        question="q", answer="a", feishu_message_id=None,
        lang="en", ticket=await _ticket(db_session, employee),
    )
    assert "cloud-en" in fake_chat.created[0]["name"]


@pytest.mark.asyncio
async def test_only_enabled_routes_are_offered_to_the_model(
    db_session: AsyncSession,
) -> None:
    """喂给模型的类别清单就是它能选的闭集 —— 关掉的不该在里面。"""
    on = await _route(db_session, key="cloud")
    await _route(db_session, key="security", enabled=False)
    routes = await handoff.load_routes(db_session)
    assert [r.id for r in routes] == [on.id]


@pytest.mark.asyncio
async def test_an_asker_with_a_blank_work_email_gets_no_group(
    db_session: AsyncSession, no_llm, fake_chat: _FakeClient,
) -> None:
    """两个人都在群里才算「把人拉进来了」。

    ``work_email`` 在库里是 NOT NULL,但**空字符串是合法值**(它通过 NOT NULL,
    而 ``if not email`` 判它为假),而空邮箱的人在生产上真实存在过(见 memory
    里飞书那条)。这里空的是**提问者**:没有他的群是一个只有负责人的房间,比
    没有群更糟。

    为什么是提问者、而不是负责人 —— 这是一次**变异幸存**逼出来的发现。第一版
    这条测试给的是「负责人邮箱为空」,结果把 ``len(open_ids) < 2`` 改成 ``< 1``
    测试照样绿:负责人那一侧在更早的 ``not owner.work_email`` 就已经降级成
    ``no_owner``,根本走不到建群那一步。**真正能走到这道闸的只有提问者这一侧。**
    """
    employee = await _person(db_session, "员工")
    owner = await _person(db_session, "负责人")
    await _route(db_session, key="other", owner=owner, is_default=True)
    employee.work_email = ""
    await db_session.flush()
    msg = await _message(db_session, employee)

    result = await handoff.hand_off(
        db_session, person=employee, message_id=msg.id,
        question="q", answer="a", feishu_message_id="om_x",
        lang="zh", ticket=await _ticket(db_session, employee),
    )

    assert result.chat_created is False, "只有一个人也建了群"
    assert fake_chat.created == []
    assert result.degraded_reason == "chat_failed"


@pytest.mark.asyncio
async def test_a_group_that_was_created_is_never_reported_as_absent(
    db_session: AsyncSession, no_llm, fake_chat: _FakeClient, monkeypatch,
) -> None:
    """群建出来了,之后哪一步失败都不能再报「没有群」。

    这条补的是复审里发现的真缺陷:开场白/转发原来和建群在同一个 try 里,任一
    失败就 `return None` —— 于是员工被告知「没拉到人」,而飞书里躺着一个两个人
    都在的群没人提起。**孤儿群比没有群更让人困惑。** 群本身才是有价值的产物,
    开场白和转发是锦上添花。
    """
    async def _boom(*, chat_id: str, text: str) -> None:
        raise RuntimeError("intro failed")

    monkeypatch.setattr(fake_chat, "send_text_to_chat", _boom)
    employee = await _person(db_session, "员工")
    owner = await _person(db_session, "负责人")
    await _route(db_session, key="other", owner=owner, is_default=True)
    msg = await _message(db_session, employee)

    result = await handoff.hand_off(
        db_session, person=employee, message_id=msg.id,
        question="q", answer="a", feishu_message_id="om_x",
        lang="zh", ticket=await _ticket(db_session, employee),
    )

    assert len(fake_chat.created) == 1, "群没建出来,这条测的前提就不成立"
    assert result.chat_created is True, "群建好了却报成没有群"
    assert result.degraded_reason is None


@pytest.mark.asyncio
async def test_pulling_the_owner_in_also_hands_them_the_ticket(
    db_session: AsyncSession, no_llm, fake_chat: _FakeClient,
) -> None:
    """把人拉进房间和把活交给他,必须是同一个动作。

    2026-09-08 之前它们不是:群拉起来了、负责人在里面聊着,而工单在队列里仍然
    「无人认领」—— SLA 的首次响应永远不会被记上,看板上它是一条没人管的记录,
    而实际上有人正在处理它。
    """
    employee = await _person(db_session, "员工")
    owner = await _person(db_session, "负责人")
    await _route(db_session, key="other", owner=owner, is_default=True)
    msg = await _message(db_session, employee)
    ticket = await _ticket(db_session, employee)
    assert ticket.assignee_id is None

    result = await handoff.hand_off(
        db_session, person=employee, message_id=msg.id,
        question="q", answer="a", feishu_message_id="om_x",
        lang="zh", ticket=ticket,
    )

    assert result.chat_created is True
    assert ticket.assignee_id == owner.id, "拉了人却没把工单交给他"
    assert ticket.status == "triaging"


@pytest.mark.asyncio
async def test_a_failed_chat_leaves_the_ticket_unassigned(
    db_session: AsyncSession, no_llm, monkeypatch,
) -> None:
    """没有群就没有接手的人 —— 工单留在队列里,那是一个人看得见的状态。

    反过来(建群失败却把工单指派出去)更糟:队列上它有主了,而那个人根本不知道
    这件事存在。
    """
    from app.assistant import channel

    client = _FakeClient(fail_create=True)

    async def _build(_db):
        return client

    monkeypatch.setattr(channel, "build_client", _build)

    employee = await _person(db_session, "员工")
    owner = await _person(db_session, "负责人")
    await _route(db_session, key="other", owner=owner, is_default=True)
    msg = await _message(db_session, employee)
    ticket = await _ticket(db_session, employee)

    result = await handoff.hand_off(
        db_session, person=employee, message_id=msg.id,
        question="q", answer="a", feishu_message_id="om_x",
        lang="zh", ticket=ticket,
    )

    assert result.chat_created is False
    assert result.degraded_reason == "chat_failed"
    assert ticket.assignee_id is None