import asyncio
from uuid import UUID

import pytest
from sqlalchemy import select
from test_after_sales import scenario  # noqa: F401
from test_conversations import setup

from zhifu.core.errors import BusinessError
from zhifu.modules.agent_runtime.config import AgentSettings
from zhifu.modules.agent_runtime.models import AgentRun, ModelAttempt
from zhifu.modules.agent_runtime.worker import execute_one


async def test_real_model_uses_previous_conversation(scenario):  # noqa: F811
    path = await setup(scenario)
    app = scenario["app"]
    for text in ("我的商品是键盘，先记住这个信息。", "我上一条消息说的商品是什么？"):
        response = await scenario["call"](
            "POST", path + "/messages", payload={"text": text, "expected_control_version": 1}
        )
        assert response.status_code == 201, response.text
        assert await execute_one(app.state.sessions, app.state.state, AgentSettings())
    response = await scenario["call"]("GET", path + "/messages")
    answers = [event for event in response.json()["items"] if event["type"] == "message.assistant"]
    assert len(answers) == 2 and "键盘" in answers[-1]["payload"]["text"]
    response = await scenario["call"]("GET", path)
    assert response.json()["active_run_id"] is None


async def test_takeover_during_real_model_request_rejects_late_answer(scenario):  # noqa: F811
    path = await setup(scenario)
    app = scenario["app"]
    response = await scenario["call"](
        "POST",
        path + "/messages",
        payload={"text": "请说明申请售后时需要提供哪些信息。", "expected_control_version": 1},
    )
    run_id = UUID(response.json()["payload"]["run_id"])
    task = asyncio.create_task(execute_one(app.state.sessions, app.state.state, AgentSettings()))
    try:
        async with asyncio.timeout(10):
            while True:
                async with app.state.sessions() as session:
                    attempt = await session.scalar(
                        select(ModelAttempt).where(ModelAttempt.run_id == run_id.bytes)
                    )
                if attempt:
                    break
                await asyncio.sleep(0.02)
        response = await scenario["call"](
            "POST", path + "/takeover", "agent", {"expected_control_version": 1}
        )
        assert response.status_code == 200
        with pytest.raises(BusinessError, match="控制权已变更"):
            await task
    finally:
        if not task.done():
            task.cancel()
            await asyncio.gather(task, return_exceptions=True)
    async with app.state.sessions() as session:
        run = await session.get(AgentRun, run_id.bytes)
        assert run.status == "CANCELLED"
    response = await scenario["call"]("GET", path + "/messages")
    assert all(event["type"] != "message.assistant" for event in response.json()["items"])