"""Assistant conversation kernel: memory, isolation, transcript, resilience.

The properties asserted here are the ones an employee and an administrator each
depend on: my thread is mine and remembers me, a slow or broken model still
produces a usable reply, and everything that was said is recoverable afterwards.
"""

from __future__ import annotations

import uuid
from datetime import UTC, datetime

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

from app.assistant import conversation, session
from app.assistant.feishu_events import handle_message_event
from app.assistant.models import (
    OUTCOME_ANSWERED,
    OUTCOME_FAILED,
    OUTCOME_UNAVAILABLE,
    ROLE_ASSISTANT,
    ROLE_USER,
    AssistantConversation,
    AssistantMessage,
)
from tests.test_assistant.conftest import (
    UNION_ID,
    FakeCopilotResult,
    make_feishu_person,
    message_event,
)

pytestmark = pytest.mark.asyncio

OTHER_UNION_ID = "on_assistant_union_0002"
OTHER_OPEN_ID = "ou_assistant_open_0002"


def test_software_role_product_extraction_is_narrow():
    assert conversation._software_role_product("Netsuite这个软件的负责人是谁") == "Netsuite"
    assert conversation._software_role_product("谁是 Netsuite 的负责人") == "Netsuite"
    assert conversation._software_role_product("Netsuite 的 IT 联系人是谁") == "Netsuite"
    assert conversation._software_role_product("怎么申请 Netsuite 坐席") is None


def test_buyback_intent_and_policy_source_filter_are_deliberately_narrow():
    assert conversation._is_buyback_question("电脑回购的政策是什么")
    assert conversation._is_buyback_question("Can I buy out my laptop?")
    assert conversation._is_buyback_question("What is the laptop buyback policy?")
    assert not conversation._is_buyback_question("电脑坏了怎么维修")
    assert conversation._supports_current_buyback_lifecycle("终端硬件生命周期：36个月。")
    # The production mirror contains this legacy wording. A document cannot
    # corroborate the current workflow merely because it also mentions 36.
    assert not conversation._supports_current_buyback_lifecycle(
        "不满2年，回购按原价/36个月计算",
    )


def test_buyback_intent_handles_large_untrusted_input_in_linear_time():
    repeated = "购买" * 100_000
    assert not conversation._is_buyback_question(repeated)
    assert conversation._is_buyback_question(f"{repeated}电脑")


async def _ask(db, text: str, *, event_id: str, union_id=UNION_ID, open_id=None) -> str:
    from tests.test_assistant.conftest import OPEN_ID

    return await handle_message_event(
        db,
        message_event(
            text=text,
            event_id=event_id,
            union_id=union_id,
            open_id=open_id or OPEN_ID,
        ),
        trace_id=f"trace-{event_id}",
    )


# ── remembered context ──────────────────────────────────────────────


async def test_second_question_carries_the_first_exchange(
    db_session: AsyncSession,
    enabled_gate,
    feishu_person,
    sent,
    kernel,
):
    await _ask(db_session, "我的电脑保修到什么时候", event_id="e1")
    await _ask(db_session, "那续保要找谁", event_id="e2")

    second_prompt = kernel["calls"][1]
    assert [t["role"] for t in second_prompt] == [ROLE_USER, ROLE_ASSISTANT, ROLE_USER]
    assert second_prompt[0]["content"] == "我的电脑保修到什么时候"
    assert second_prompt[-1]["content"] == "那续保要找谁"


async def test_buyback_policy_turn_does_not_reuse_stale_chat_policy(
    db_session: AsyncSession,
    enabled_gate,
    feishu_person,
    sent,
    kernel,
    monkeypatch,
):
    """A prior assistant answer is never evidence for a current policy turn."""

    async def grounded(_db):
        return conversation.BuybackGrounding(
            system_context="CURRENT BUYBACK POLICY: 36-month (3-year) lifecycle.",
            docs=[],
            conflicting_docs=[],
        )

    monkeypatch.setattr(conversation, "_ground_buyback_policy", grounded)
    await _ask(db_session, "电脑回购", event_id="buyback-old")
    await _ask(db_session, "电脑回购的政策是什么", event_id="buyback-policy")

    policy_turn = kernel["calls"][1]
    assert policy_turn == [{"role": ROLE_USER, "content": "电脑回购的政策是什么"}]
    assert "36-month (3-year) lifecycle" in kernel["state"]["last_system_prompt"]


async def test_the_turn_runs_on_the_chat_prompt_not_the_web_one(
    db_session: AsyncSession,
    enabled_gate,
    feishu_person,
    sent,
    kernel,
):
    """A Feishu turn must not inherit the web copilot's instructions.

    The web prompt tells the model to point at a '转人工 / 开工单' button, which
    in chat sends the employee to a control that isn't there — the reason
    answers read as "go do it yourself" instead of "tell me and I'll submit it".
    """
    await _ask(db_session, "怎么申请邮件别名", event_id="p1")

    prompt = kernel["state"]["last_system_prompt"]
    assert prompt is not None
    assert "Feishu chat" in prompt
    assert "转人工" not in prompt


async def test_how_to_question_gets_server_prefetched_kb_evidence(
    db_session: AsyncSession,
    enabled_gate,
    feishu_person,
    cards,
    kernel,
    monkeypatch,
):
    """A model cannot skip KB and then claim the exact guide is absent."""

    async def grounded(_db, _question):
        return conversation.KnowledgeGrounding(
            system_context=(
                "\nKB EVIDENCE PREFETCH — THIS TURN\n"
                "- 如何以群组名义发送邮件: 请从发件人下拉框选择群组邮箱。\n"
            ),
            docs=[
                {
                    "id": "kb-group-mail",
                    "title": "如何以群组名义发送邮件",
                    "url": "https://kb.example/group-mail",
                    "source": "feishu_doc",
                    "updated_at": "",
                }
            ],
        )

    monkeypatch.setattr(conversation, "_ground_kb_question", grounded)
    await _ask(db_session, "如何以群组名义发送邮件", event_id="kb-prefetch")

    assert "KB EVIDENCE PREFETCH" in kernel["state"]["last_system_prompt"]
    row = (
        (
            await db_session.execute(
                select(AssistantMessage).where(AssistantMessage.role == ROLE_ASSISTANT),
            )
        )
        .scalars()
        .one()
    )
    assert row.sources == [
        {"title": "如何以群组名义发送邮件", "url": "https://kb.example/group-mail"},
    ]
    # The server prefetch itself is grounding.  It must make the delivery use
    # an answer card, otherwise the employee gets neither the source link nor
    # the feedback/escalation controls merely because the model skipped a
    # redundant search_knowledge_base tool call.
    assert "https://kb.example/group-mail" in str(cards[-1]["card"])


async def test_terminal_request_starts_a_new_message_level_lifecycle(
    db_session: AsyncSession,
    enabled_gate,
    feishu_person,
    sent,
    kernel,
):
    from app.core.coordinators.assistant_support_coordinator import (
        record_assistant_request_verdict,
    )

    await _ask(db_session, "第一个问题", event_id="request-lifecycle-1")
    first_answer = (
        (
            await db_session.execute(
                select(AssistantMessage)
                .where(AssistantMessage.role == ROLE_ASSISTANT)
                .order_by(AssistantMessage.seq.asc())
            )
        )
        .scalars()
        .first()
    )
    assert first_answer and first_answer.service_request_id
    first_request_id = first_answer.service_request_id
    await record_assistant_request_verdict(
        db_session,
        service_request_id=first_request_id,
        helpful=True,
    )
    await db_session.commit()

    await _ask(db_session, "第二个问题", event_id="request-lifecycle-2")
    answers = (
        (
            await db_session.execute(
                select(AssistantMessage)
                .where(AssistantMessage.role == ROLE_ASSISTANT)
                .order_by(AssistantMessage.seq.asc())
            )
        )
        .scalars()
        .all()
    )
    assert len(answers) == 2
    assert answers[1].service_request_id
    assert answers[1].service_request_id != first_request_id


async def test_model_cannot_claim_a_confirmation_card_without_sending_one(
    db_session: AsyncSession,
    enabled_gate,
    feishu_person,
    sent,
    kernel,
):
    """A copied card-success sentence must not become a false side effect."""
    kernel["state"]["result"] = FakeCopilotResult(
        reply="（我已经把确认卡片发给你，等你在卡片上确认。）",
    )

    await _ask(db_session, "创建吧", event_id="card-claim-without-send")

    assert sent
    assert "没发出去" in sent[-1]["text"]
    assert "还没提交任何东西" in sent[-1]["text"]
    row = (
        (
            await db_session.execute(
                select(AssistantMessage).where(
                    AssistantMessage.role == ROLE_ASSISTANT,
                ),
            )
        )
        .scalars()
        .one()
    )
    assert row.outcome == OUTCOME_FAILED


async def test_one_persons_history_never_reaches_another(
    db_session: AsyncSession,
    enabled_gate,
    feishu_person,
    sent,
    kernel,
):
    await make_feishu_person(
        db_session,
        subject=OTHER_UNION_ID,
        email="other@test.com",
        name="Other",
    )
    await _ask(db_session, "我的工资卡号是多少", event_id="e1")
    await _ask(
        db_session,
        "刚才那个问题的答案是什么",
        event_id="e2",
        union_id=OTHER_UNION_ID,
        open_id=OTHER_OPEN_ID,
    )

    prompt_seen_by_second_person = kernel["calls"][1]
    assert len(prompt_seen_by_second_person) == 1
    assert "工资卡号" not in str(prompt_seen_by_second_person)


async def test_reset_command_clears_the_thread_without_calling_the_model(
    db_session: AsyncSession,
    enabled_gate,
    feishu_person,
    sent,
    kernel,
):
    await _ask(db_session, "第一个问题", event_id="e1")
    outcome = await _ask(db_session, "新话题", event_id="e2")
    await _ask(db_session, "第二个问题", event_id="e3")

    assert outcome == "reset"
    assert len(kernel["calls"]) == 2  # the reset itself never reached the model
    assert len(kernel["calls"][1]) == 1


async def test_help_command_is_deterministic_and_does_not_reach_the_model(
    db_session: AsyncSession,
    enabled_gate,
    feishu_person,
    sent,
    kernel,
):
    outcome = await _ask(db_session, "/help", event_id="help")

    assert outcome == "help"
    assert kernel["calls"] == []
    assert "/new" in sent[-1]["text"]
    assert "/reset" in sent[-1]["text"]


async def test_unavailable_kernel_does_not_poison_later_context(
    db_session: AsyncSession,
    enabled_gate,
    feishu_person,
    sent,
    kernel,
):
    kernel["state"]["result"] = FakeCopilotResult(available=False, reply="")
    await _ask(db_session, "第一个问题", event_id="e1")
    kernel["state"]["result"] = FakeCopilotResult()
    await _ask(db_session, "第二个问题", event_id="e2")

    assert len(kernel["calls"][1]) == 1


# ── resilience ──────────────────────────────────────────────────────


async def test_model_failure_still_answers_with_a_next_step(
    db_session: AsyncSession,
    enabled_gate,
    feishu_person,
    sent,
    kernel,
):
    kernel["state"]["raises"] = True

    outcome = await _ask(db_session, "我的电脑呢", event_id="e1")

    assert outcome == OUTCOME_FAILED
    assert len(sent) == 1
    assert "helix.pingcap.net" in sent[0]["text"]


async def test_disabled_kernel_points_at_the_portal(
    db_session: AsyncSession,
    enabled_gate,
    feishu_person,
    sent,
    kernel,
):
    kernel["state"]["result"] = FakeCopilotResult(available=False, reply="")

    outcome = await _ask(db_session, "我的电脑呢", event_id="e1")

    assert outcome == OUTCOME_UNAVAILABLE
    assert "helix.pingcap.net" in sent[0]["text"]


async def test_empty_model_reply_never_reaches_the_employee(
    db_session: AsyncSession,
    enabled_gate,
    feishu_person,
    sent,
    kernel,
):
    kernel["state"]["result"] = FakeCopilotResult(reply="   ")

    await _ask(db_session, "我的电脑呢", event_id="e1")

    assert sent[0]["text"].strip()


async def test_slow_turn_sends_a_holding_note_then_the_answer(
    db_session: AsyncSession,
    enabled_gate,
    feishu_person,
    sent,
    kernel,
    monkeypatch,
):
    monkeypatch.setattr(conversation, "HOLDING_NOTE_AFTER_SECONDS", 0.05)
    kernel["state"]["delay"] = 0.3

    await _ask(db_session, "我的电脑呢", event_id="e1")

    assert len(sent) == 2
    assert "稍等" in sent[0]["text"]
    assert sent[1]["text"] == "你的电脑保修到 2027-03-01。"


async def test_fast_turn_sends_only_the_answer(
    db_session: AsyncSession,
    enabled_gate,
    feishu_person,
    sent,
    kernel,
    monkeypatch,
):
    monkeypatch.setattr(conversation, "HOLDING_NOTE_AFTER_SECONDS", 5.0)

    await _ask(db_session, "我的电脑呢", event_id="e1")

    assert len(sent) == 1


async def test_a_second_message_while_busy_is_told_to_wait(
    db_session: AsyncSession,
    enabled_gate,
    feishu_person,
    sent,
    kernel,
):
    held = await session.acquire_turn_lock(
        person_id=feishu_person.id,
        chat_id="oc-0001",
    )
    assert held is not None

    outcome = await _ask(db_session, "在吗", event_id="e1")

    assert outcome == "busy"
    assert kernel["calls"] == []


async def test_a_turn_cannot_release_a_lock_it_no_longer_owns(
    db_session: AsyncSession,
    enabled_gate,
    feishu_person,
    sent,
    kernel,
):
    """An overrunning turn must not free the next turn's lock on its way out."""
    stale = await session.acquire_turn_lock(
        person_id=feishu_person.id,
        chat_id="oc-0001",
    )
    await session.release_turn_lock(
        person_id=feishu_person.id,
        chat_id="oc-0001",
        owner=stale,
    )
    current = await session.acquire_turn_lock(
        person_id=feishu_person.id,
        chat_id="oc-0001",
    )

    # The previous holder tries to release again — must not touch the new lock.
    await session.release_turn_lock(
        person_id=feishu_person.id,
        chat_id="oc-0001",
        owner=stale,
    )

    assert current is not None
    assert (
        await session.acquire_turn_lock(
            person_id=feishu_person.id,
            chat_id="oc-0001",
        )
        is None
    )


async def test_the_turn_lock_is_released_even_when_the_model_fails(
    db_session: AsyncSession,
    enabled_gate,
    feishu_person,
    sent,
    kernel,
):
    kernel["state"]["raises"] = True
    await _ask(db_session, "第一个问题", event_id="e1")

    kernel["state"]["raises"] = False
    outcome = await _ask(db_session, "第二个问题", event_id="e2")

    assert outcome == OUTCOME_ANSWERED


# ── durable transcript ──────────────────────────────────────────────


async def test_transcript_records_both_sides_with_evidence(
    db_session: AsyncSession,
    enabled_gate,
    feishu_person,
    sent,
    kernel,
):
    kernel["state"]["result"] = FakeCopilotResult(
        tool_calls=["get_my_support_tickets", "search_knowledge_base"],
    )

    await _ask(db_session, "我的工单到哪了", event_id="e1")

    rows = (
        (
            await db_session.execute(
                select(AssistantMessage).order_by(AssistantMessage.seq),
            )
        )
        .scalars()
        .all()
    )
    assert [r.role for r in rows] == [ROLE_USER, ROLE_ASSISTANT]
    assert [r.seq for r in rows] == [0, 1]
    assert rows[0].content == "我的工单到哪了"
    assert rows[1].tool_calls == ["get_my_support_tickets", "search_knowledge_base"]
    assert rows[1].outcome == OUTCOME_ANSWERED
    assert rows[1].provider == "bailian"
    assert rows[1].latency_ms is not None
    assert rows[1].trace_id == "trace-e1"


async def test_failed_turns_are_recorded_too(
    db_session: AsyncSession,
    enabled_gate,
    feishu_person,
    sent,
    kernel,
):
    """ "The assistant never answered me" must be checkable, not invisible."""
    kernel["state"]["raises"] = True

    await _ask(db_session, "我的电脑呢", event_id="e1")

    rows = (await db_session.execute(select(AssistantMessage))).scalars().all()
    assert len(rows) == 2
    assert any(r.outcome == OUTCOME_FAILED for r in rows)


async def test_a_second_turn_continues_the_sequence(
    db_session: AsyncSession,
    enabled_gate,
    feishu_person,
    sent,
    kernel,
):
    """Ordering must survive two turns landing in the same clock second."""
    await _ask(db_session, "第一个问题", event_id="e1")
    await _ask(db_session, "第二个问题", event_id="e2")

    rows = (
        (
            await db_session.execute(
                select(AssistantMessage).order_by(AssistantMessage.seq),
            )
        )
        .scalars()
        .all()
    )
    assert [r.seq for r in rows] == [0, 1, 2, 3]
    assert rows[0].content == "第一个问题"
    assert rows[2].content == "第二个问题"


async def test_one_thread_is_one_conversation_row(
    db_session: AsyncSession,
    enabled_gate,
    feishu_person,
    sent,
    kernel,
):
    await _ask(db_session, "第一个问题", event_id="e1")
    await _ask(db_session, "第二个问题", event_id="e2")

    rows = (await db_session.execute(select(AssistantConversation))).scalars().all()
    assert len(rows) == 1
    assert rows[0].turn_count == 2
    assert rows[0].person_id == feishu_person.id


async def test_expired_memory_starts_a_new_conversation_row(
    db_session: AsyncSession,
    enabled_gate,
    feishu_person,
    sent,
    kernel,
):
    await _ask(db_session, "第一个问题", event_id="e1")
    await session.clear_history(person_id=feishu_person.id, chat_id="oc-0001")
    await _ask(db_session, "很久以后的问题", event_id="e2")

    rows = (await db_session.execute(select(AssistantConversation))).scalars().all()
    assert len(rows) == 2


class TestTheTurnSurvivesItsOwnFailures:
    """Four defects that all ended with the employee getting nothing, or being
    quietly told something untrue about their own thread."""

    @pytest.mark.asyncio
    async def test_a_poisoned_session_still_produces_a_reply_and_a_record(
        self,
        db_session,
        feishu_person,
        kernel,
        enabled_gate,
    ) -> None:
        """Neither the kernel nor its tool dispatcher rolls back.

        So a DBAPI error inside any bridged tool leaves the session needing one,
        and the very next statement is the transcript write. Unrolled, that write
        raised ``PendingRollbackError`` straight out of ``run_turn``, past the
        delivery branch: no message reached the employee — not even the canned
        failure — and no transcript or audit row was written.
        """
        from app.assistant import conversation
        from app.assistant.models import AssistantMessage

        async def _poison(*_a, **_kw):
            # A failed FLUSH, measured to be what actually leaves the session in
            # "needs rollback": a failed SELECT does not (probed against this
            # driver — the first two cuts of this test used one and passed with
            # the fix reverted, proving nothing). An INSERT is also the realistic
            # shape, since a read-only tool cannot poison a session.
            db_session.add(
                AssistantMessage(
                    conversation_id=uuid.uuid4(),  # no such parent → FK 1452
                    seq=0,
                    role="user",
                    content="x",
                ),
            )
            await db_session.flush()

        kernel["state"]["raises"] = False
        import app.domains.support.copilot as copilot_module

        monkey = copilot_module.run_copilot_turn
        copilot_module.run_copilot_turn = _poison
        try:
            turn = await conversation.run_turn(
                db_session,
                person=feishu_person,
                chat_id="oc-poison",
                question="我的电脑保修到什么时候",
                lang="zh",
                trace_id="t-poison",
            )
        finally:
            copilot_module.run_copilot_turn = monkey

        assert turn.outcome == "failed"
        assert turn.reply.strip(), "the employee was told nothing at all"
        assert turn.message_id, "the failure left no transcript row"

    @pytest.mark.asyncio
    async def test_the_caller_can_still_write_its_audit_row(
        self,
        db_session,
        feishu_person,
        kernel,
        sent,
        enabled_gate,
    ) -> None:
        """The rollback expires the caller's ``person`` too.

        ``feishu_events`` reads ``person.id`` for its audit row after ``run_turn``
        returns; on an expired instance that read is lazy IO and raises
        ``MissingGreenlet``, so the ONE record of "the assistant did not answer
        me" is lost — inside the handler built to make that answerable. Driven
        through the real entry point, because the defect is in what the caller
        does next.
        """
        import app.domains.support.copilot as copilot_module
        from app.assistant.models import AssistantMessage
        from app.audit.models import AuditLog

        async def _poison(*_a, **_kw):
            db_session.add(
                AssistantMessage(
                    conversation_id=uuid.uuid4(),
                    seq=0,
                    role="user",
                    content="x",
                ),
            )
            await db_session.flush()

        # A real UUID: ``audit_logs.trace_id`` is a UUID column, so a tag-shaped
        # id makes the assertion query itself throw rather than the code.
        trace_id = str(uuid.uuid4())
        original = copilot_module.run_copilot_turn
        copilot_module.run_copilot_turn = _poison
        try:
            outcome = await handle_message_event(
                db_session,
                message_event(text="我的电脑保修到什么时候", event_id="evt-poison"),
                trace_id=trace_id,
            )
        finally:
            copilot_module.run_copilot_turn = original

        assert outcome == "failed"
        assert sent, "the employee was told nothing"
        rows = (
            (
                await db_session.execute(
                    select(AuditLog).where(
                        AuditLog.action == "assistant.message.received",
                        AuditLog.trace_id == trace_id,
                    ),
                )
            )
            .scalars()
            .all()
        )
        assert len(rows) == 1, "the failure left no audit row"

    @pytest.mark.asyncio
    async def test_an_empty_reply_is_recorded_as_a_failure(
        self,
        db_session,
        feishu_person,
        kernel,
        enabled_gate,
    ) -> None:
        """A provider that returns no text is a fault, not an answer.

        Recording it as ``answered`` puts it in the helpful-rate denominator and
        makes the metric read low for a reason that has nothing to do with the
        answers — the argument the ``turn_failed`` branch already makes.
        """
        from app.assistant import conversation
        from tests.test_assistant.conftest import FakeCopilotResult

        kernel["state"]["result"] = FakeCopilotResult(reply="   ")

        turn = await conversation.run_turn(
            db_session,
            person=feishu_person,
            chat_id="oc-empty",
            question="在吗",
            lang="zh",
            trace_id="t-empty",
        )

        assert turn.outcome == "failed"
        assert turn.reply.strip(), "and the employee still gets a next step"

    @pytest.mark.asyncio
    async def test_a_proposal_is_filed_against_this_turns_own_thread(
        self,
        db_session,
        feishu_person,
        kernel,
        cards,
        enabled_gate,
    ) -> None:
        """Not against the previous, already-closed one.

        The conversation row for THIS turn did not exist when the dispatcher was
        built, so the old lookup ("most recent row for this chat") attached the
        proposal to whatever thread came before — and joining the transcript to
        the action is most of what 「可回溯」 means.
        """
        from app.assistant import conversation
        from app.assistant.models import AssistantActionProposal, AssistantConversation

        # A thread that has already ended: same person, same chat, older.
        stale = AssistantConversation(
            person_id=feishu_person.id,
            channel="feishu",
            chat_id="oc-join",
            last_message_at=datetime(2020, 1, 1, tzinfo=UTC),
        )
        db_session.add(stale)
        await db_session.commit()

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

        turn = await conversation.run_turn(
            db_session,
            person=feishu_person,
            chat_id="oc-join",
            question="帮我申请 VPN",
            lang="zh",
            trace_id="t-join",
            open_id="ou_assistant_open_0001",
        )

        assert turn.card_only is True
        proposal = (await db_session.execute(select(AssistantActionProposal))).scalars().one()
        assert proposal.conversation_id is not None, "the proposal joined to nothing"
        assert proposal.conversation_id != stale.id, (
            "the proposal was filed against the previous, closed thread"
        )
        message = (
            (
                await db_session.execute(
                    select(AssistantMessage).where(
                        AssistantMessage.id == uuid.UUID(turn.message_id)
                    ),
                )
            )
            .scalars()
            .one()
        )
        assert proposal.conversation_id == message.conversation_id


class TestResetCannotBeUndoneByARunningTurn:
    """The reset is the one control an employee has over the model's memory.

    「新话题」 during a slow turn cleared the thread, answered 「好的，我们从头开始」
    — and then the still-running turn wrote the whole pre-reset history back. The
    employee was told the thread was reset and the model kept every word of it.
    """

    @pytest.mark.asyncio
    async def test_a_write_from_before_the_reset_is_dropped(
        self,
        db_session,
        feishu_person,
    ) -> None:
        from app.assistant import session as session_store

        person_id, chat_id = feishu_person.id, "oc-reset-race"
        await session_store.save_history(
            db_session,
            person_id=person_id,
            chat_id=chat_id,
            history=[{"role": "user", "content": "第一句"}],
        )
        # A turn starts and captures the epoch…
        epoch = await session_store.current_epoch(
            person_id=person_id,
            chat_id=chat_id,
        )
        # …the employee resets while it is still running…
        await session_store.clear_history(person_id=person_id, chat_id=chat_id)
        assert (
            await session_store.load_history(
                db_session,
                person_id=person_id,
                chat_id=chat_id,
            )
            == []
        )
        # …and the turn finishes and tries to write its thread back.
        await session_store.save_history(
            db_session,
            person_id=person_id,
            chat_id=chat_id,
            history=[
                {"role": "user", "content": "第一句"},
                {"role": "assistant", "content": "第一句的回答"},
            ],
            expect_epoch=epoch,
        )

        assert (
            await session_store.load_history(
                db_session,
                person_id=person_id,
                chat_id=chat_id,
            )
            == []
        ), "the reset was silently undone"

    @pytest.mark.asyncio
    async def test_a_normal_turn_still_writes(
        self,
        db_session,
        feishu_person,
    ) -> None:
        """The guard must not cost every turn its memory."""
        from app.assistant import session as session_store

        person_id, chat_id = feishu_person.id, "oc-reset-normal"
        epoch = await session_store.current_epoch(
            person_id=person_id,
            chat_id=chat_id,
        )
        await session_store.save_history(
            db_session,
            person_id=person_id,
            chat_id=chat_id,
            history=[{"role": "user", "content": "记住这句"}],
            expect_epoch=epoch,
        )

        assert await session_store.load_history(
            db_session,
            person_id=person_id,
            chat_id=chat_id,
        ) == [{"role": "user", "content": "记住这句"}]

    @pytest.mark.asyncio
    async def test_a_turn_that_starts_after_the_reset_writes_normally(
        self,
        db_session,
        feishu_person,
    ) -> None:
        from app.assistant import session as session_store

        person_id, chat_id = feishu_person.id, "oc-reset-after"
        await session_store.clear_history(person_id=person_id, chat_id=chat_id)
        epoch = await session_store.current_epoch(
            person_id=person_id,
            chat_id=chat_id,
        )
        await session_store.save_history(
            db_session,
            person_id=person_id,
            chat_id=chat_id,
            history=[{"role": "user", "content": "新话题的第一句"}],
            expect_epoch=epoch,
        )

        assert await session_store.load_history(
            db_session,
            person_id=person_id,
            chat_id=chat_id,
        ) == [{"role": "user", "content": "新话题的第一句"}]