import asyncio
import json
from pathlib import Path
from uuid import UUID, uuid4

from sqlalchemy import func, select
from test_after_sales import scenario  # noqa: F401

from zhifu.modules.after_sales.models import Approval, OrderProjection, Ticket
from zhifu.modules.agent_runtime.budget_callback import CallAllowance
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
from zhifu.modules.identity.auth import now


def reserve_allowance():
    # 独占预留文件限制本专项的实际外呼总数，失败后不会自动再次执行。
    with Path(".work/clarification-allowance.json").open("x", encoding="utf-8") as stream:
        json.dump({"reserved_requests": 3, "automatic_retries": 0}, stream)


async def test_real_model_waits_for_order_and_reason(scenario):  # noqa: F811
    sessions, state = scenario["app"].state.sessions, scenario["app"].state.state
    async with sessions() as session:
        assert await session.scalar(select(func.count()).select_from(AgentRun).where(
            AgentRun.status.in_(["QUEUED", "RUNNING"]))) == 0
    for role in ("agent", "memory"):
        async for _ in state.scan_iter(match=f"monitor:worker:{role}:*"):
            raise RuntimeError("模型补问专项要求停止其他生成工作进程")
    async with sessions.begin() as session:
        session.add(OrderProjection(
            id=uuid4().bytes, tenant_id=scenario["tenant"].bytes, store_id=scenario["store"].bytes,
            subject_id=scenario["accounts"]["customer"].bytes, source_system="acceptance",
            source_version=1, item_id="keyboard-2", product_type="keyboard", available_quantity=1,
            signed_at=now(), observed_at=now(),
        ))
    await asyncio.to_thread(reserve_allowance)
    allowance, run_id = CallAllowance(3), None
    call = scenario["call"]
    try:
        async with asyncio.timeout(100):
            response = await call("POST", "/openapi/v1/agent/runs", payload={
                "store_id": str(scenario["store"]),
                "message": "我想换货，请查询我的订单让我选择，我还没有说明故障原因。",
            })
            assert response.status_code == 202, response.text
            run_id = UUID(response.json()["run_id"])
            assert await execute_one(sessions, state,
                AgentSettings(max_output_tokens=512, model_timeout=30),
                scenario["app"].state.knowledge.search, allowance=allowance)
            path = f"/openapi/v1/agent/runs/{run_id}"
            response = await call("GET", path)
            view = response.json()
            assert view["status"] == "WAITING_USER", response.text
            assert len(view["waiting"]["candidate_orders"]) == 2
            assert {"order_id", "reason"} <= set(view["waiting"]["missing_fields"])
            async with sessions() as session:
                for model in (Approval, Ticket):
                    assert await session.scalar(select(func.count()).select_from(model).where(
                        model.tenant_id == scenario["tenant"].bytes)) == 0
            response = await call("POST", path + "/inputs", payload={
                "interrupt_id": view["waiting"]["interrupt_id"],
                "expected_run_version": view["run_version"],
                "answer": f"选择订单{scenario['order_id']}，按键失灵，申请1件。",
            })
            assert response.status_code == 202, response.text
            assert response.json()["run"]["run_id"] == str(run_id)
            assert response.json()["run"]["status"] == "QUEUED"
            assert (await call("POST", path + "/cancel")).status_code == 200
    finally:
        if run_id is not None:
            async with sessions() as session:
                run = await session.get(AgentRun, run_id.bytes)
                attempts = (await session.scalars(select(ModelAttempt).where(
                    ModelAttempt.run_id == run_id.bytes))).all()
                report = {
                    "run_id": str(run_id), "status": run.status, "request_limit": 3,
                    "reserved_attempts": allowance.started, "max_output_tokens": 512,
                    "model_timeout_seconds": 30, "total_timeout_seconds": 100,
                    "automatic_retries": 0,
                    "attempts": [{"status": row.status, "usage": row.usage} for row in attempts],
                    "messages": run.messages,
                }
            await asyncio.to_thread(Path(".work/clarification-model.json").write_text,
                json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")