import asyncio from datetime import timedelta from uuid import UUID, uuid4 import pytest from sqlalchemy import func, select from test_after_sales import scenario # noqa: F401 from test_agent_runtime import tool_run from test_conversations import setup from zhifu.core.errors import BusinessError from zhifu.modules.agent_runtime.config import GRAPH_RELEASES, GRAPH_VERSION from zhifu.modules.agent_runtime.models import AgentRun, RunInputEvent from zhifu.modules.agent_runtime.runtime import build_tool_graph, locked_run from zhifu.modules.agent_runtime.tools import model_tools from zhifu.modules.agent_runtime.worker import claim from zhifu.modules.conversations.models import Conversation from zhifu.modules.identity.access_models import StoreGrant from zhifu.modules.identity.auth import now pytestmark = pytest.mark.integration async def waiting_run(context, *, conversation=False): # 直接提交工具命令验证真实图节点与事务，不产生或替代供应商模型响应。 path = await setup(context) if conversation else None row = await tool_run(context, "request_clarification", { "question": "请选择订单并补充换货原因。", "missing_fields": ["order_id", "reason"], "candidate_order_ids": [str(context["order_id"])], }) sessions = context["app"].state.sessions async with sessions.begin() as session: run = await session.get(AgentRun, row.id) run.graph_version = GRAPH_VERSION if path: item = await session.get(Conversation, UUID(path.rsplit("/", 1)[1]).bytes) item.active_run_id = run.id run.conversation_id, run.control_version = item.id, item.control_version run_id, token = await claim(sessions, 2) result = await build_tool_graph(sessions).ainvoke({"run_id": run_id, "token": token}) assert result["next_node"] == "wait" response = await context["call"]("GET", f"/openapi/v1/agent/runs/{UUID(bytes=run_id)}") assert response.status_code == 200, response.text view = response.json() assert view["status"] == "WAITING_USER" and view["waiting"]["candidate_orders"] return view, token, path def answer(view): return {"interrupt_id": view["waiting"]["interrupt_id"], "expected_run_version": view["run_version"], "answer": "选择当前订单，按键失灵"} async def test_resume_same_run_atomic_idempotency_and_event(scenario): # noqa: F811 view, old_token, conversation = await waiting_run(scenario, conversation=True) call = scenario["call"] path = f"/openapi/v1/agent/runs/{view['run_id']}" assert await claim(scenario["app"].state.sessions, 2) is None responses = await asyncio.gather(*[ call("POST", path + "/inputs", payload=answer(view), key="same-input") for _ in range(3)]) assert all(response.status_code == 202 for response in responses), [r.text for r in responses] assert len({response.json()["input_event_id"] for response in responses}) == 1 result = responses[0].json() assert result["run"]["run_id"] == view["run_id"] assert result["run"]["run_version"] == view["run_version"] + 1 assert result["run"]["waiting"] is None and result["run"]["status"] == "QUEUED" conflict = await call("POST", path + "/inputs", payload={**answer(view), "answer": "其他内容"}, key="same-input") assert conflict.json()["code"] == "IDEMPOTENCY_CONFLICT" repeat = await call("POST", path + "/inputs", payload=answer(view)) assert repeat.json()["code"] == "RUN_NOT_WAITING" history = (await call("GET", conversation + "/messages")).json()["items"] assert [event["type"] for event in history].count("message.user") == 1 assert {"run.waiting", "run.resumed"} <= {event["type"] for event in history} sessions = scenario["app"].state.sessions run_id, token = await claim(sessions, 2) assert str(UUID(bytes=run_id)) == view["run_id"] and token > old_token async with sessions.begin() as session: with pytest.raises(BusinessError, match="执行权已失效"): await locked_run(session, run_id, old_token) async with sessions() as session: assert await session.scalar(select(func.count()).select_from(RunInputEvent).where( RunInputEvent.run_id == run_id)) == 1 async def test_resume_identity_version_and_cancel_boundaries(scenario): # noqa: F811 view, _, conversation = await waiting_run(scenario, conversation=True) call = scenario["call"] path = f"/openapi/v1/agent/runs/{view['run_id']}" for actor in ("other", "foreign"): assert (await call("POST", path + "/inputs", actor, answer(view))).status_code == 404 assert (await call("POST", path + "/inputs", "agent", answer(view))).status_code == 403 extra = await call("POST", path + "/inputs", payload={**answer(view), "approved": True}) assert extra.status_code == 422 old = await call("POST", path + "/inputs", payload={**answer(view), "expected_run_version": 1}) assert old.json()["code"] == "RUN_VERSION_CONFLICT" wrong = await call("POST", path + "/inputs", payload={**answer(view), "interrupt_id": str(uuid4())}) assert wrong.json()["code"] == "INTERRUPT_CONFLICT" assert (await call("POST", conversation + "/takeover", "agent", {"expected_control_version": 1})).status_code == 200 assert (await call("GET", path)).json()["status"] == "CANCELLED" assert (await call("POST", path + "/inputs", payload=answer(view))).status_code == 409 async def test_wait_expiry_and_service_receipt_authorization(scenario): # noqa: F811 view, _, _ = await waiting_run(scenario) call = scenario["call"] path = f"/openapi/v1/agent/runs/{view['run_id']}" sessions = scenario["app"].state.sessions response = await call("POST", path + "/inputs", payload=answer(view)) assert response.status_code == 202, response.text body = {key: value for key, value in answer(view).items() if key != "answer"} body["input_event_id"] = response.json()["input_event_id"] assert (await call("POST", path + "/resume", payload=body)).status_code == 403 assert (await call("POST", path + "/resume", "connector", body)).status_code == 404 async with sessions.begin() as session: session.add(StoreGrant(tenant_id=scenario["tenant"].bytes, store_id=scenario["store"].bytes, principal_id=scenario["accounts"]["connector"].bytes, permission="agent:resume")) assert (await call("POST", path + "/resume", "connector", body)).status_code == 202 wrong = await call("POST", path + "/resume", "connector", {**body, "expected_run_version": 1}) assert wrong.json()["code"] == "INTERRUPT_CONFLICT" await call("POST", path + "/cancel") cancelled = await call("POST", path + "/resume", "connector", body) assert cancelled.json()["code"] == "RUN_NOT_RESUMABLE" expired, _, _ = await waiting_run(scenario) async with sessions.begin() as session: row = await session.get(AgentRun, UUID(expired["run_id"]).bytes) row.deadline_at = now() - timedelta(seconds=1) expired_path = f"/openapi/v1/agent/runs/{expired['run_id']}" expired_response = await call("POST", expired_path + "/inputs", payload=answer(expired)) assert expired_response.json()["code"] == "RUN_WAIT_EXPIRED" assert await claim(sessions, 2) is None assert (await call("GET", expired_path)).json()["status"] == "FAILED" async def test_registered_old_graph_remains_executable(scenario): # noqa: F811 assert {"after-sales-v1", GRAPH_VERSION} <= GRAPH_RELEASES.keys() old_tools = {tool["function"]["name"] for tool in model_tools("after-sales-v1")} current_tools = {tool["function"]["name"] for tool in model_tools(GRAPH_VERSION)} assert "request_clarification" not in old_tools assert "request_clarification" in current_tools row = await tool_run(scenario, "search_orders", {}) sessions = scenario["app"].state.sessions run_id, token = await claim(sessions, 2) async with sessions() as session: assert (await session.get(AgentRun, row.id)).graph_version == "after-sales-v1" result = await build_tool_graph(sessions).ainvoke({"run_id": run_id, "token": token}) assert result["next_node"] == "model" async with sessions.begin() as session: (await session.get(AgentRun, row.id)).graph_version = "unregistered-version" async with sessions.begin() as session: with pytest.raises(BusinessError, match="运行版本不可用"): await locked_run(session, run_id, token)