import asyncio from datetime import timedelta from uuid import UUID import pytest from sqlalchemy import delete, func, select from test_after_sales import scenario # noqa: F401 from test_agent_runtime import tool_run from test_conversations import setup from zhifu.modules.after_sales.models import Approval, OrderProjection, Ticket from zhifu.modules.agent_runtime.business_wait import reconcile from zhifu.modules.agent_runtime.config import GRAPH_VERSION from zhifu.modules.agent_runtime.models import AgentRun, RunInputEvent from zhifu.modules.agent_runtime.runtime import build_tool_graph 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 from zhifu.modules.integrations.models import ExternalOperation pytestmark = pytest.mark.integration async def draft_wait(context): path = await setup(context) row = await tool_run(context, "create_exchange_proposal", context["proposal"]) sessions = context["app"].state.sessions async with sessions.begin() as session: run = await session.get(AgentRun, row.id) run.graph_version = GRAPH_VERSION conversation = await session.get(Conversation, UUID(path.rsplit("/", 1)[1]).bytes) conversation.active_run_id = run.id run.conversation_id, run.control_version = conversation.id, conversation.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" run_path = f"/openapi/v1/agent/runs/{UUID(bytes=run_id)}" view = (await context["call"]("GET", run_path)).json() assert view["status"] == "WAITING_USER" assert view["waiting"]["expected_input_type"] == "proposal_confirmation" ticket_id = view["waiting"]["proposal_id"] ticket = (await context["call"]("GET", f"/openapi/v1/tickets/{ticket_id}")).json() return run_id, run_path, path, ticket, view async def confirm(context, ticket, key="business-confirm"): response = await context["call"]( "POST", f"/openapi/v1/after-sales/proposals/{ticket['proposal_id']}/confirm", payload={ "proposal_version": ticket["proposal_version"], "expected_ticket_version": ticket["version"], }, key=key) assert response.status_code == 200, response.text return response.json() async def decide(context, ticket, decision="APPROVED"): response = await context["call"]( "POST", f"/openapi/v1/approvals/{ticket['approval_id']}/decisions", "reviewer", { "proposal_version": ticket["proposal_version"], "expected_version": ticket["version"], "decision": decision, "reason": "已核对当前版本", }, key="business-decision") assert response.status_code == 200, response.text return response.json() async def test_confirmation_approval_restore_same_run_once(scenario): # noqa: F811 run_id, path, conversation_path, ticket, view = await draft_wait(scenario) sessions = scenario["app"].state.sessions text = await scenario["call"]("POST", path + "/inputs", payload={ "interrupt_id": view["waiting"]["interrupt_id"], "expected_run_version": view["run_version"], "answer": "我确认且主管批准", }) assert text.status_code == 409 and text.json()["code"] == "BUSINESS_EVENT_REQUIRED" confirmed = await asyncio.gather(*[confirm(scenario, ticket) for _ in range(3)]) assert sum(await asyncio.gather(*[reconcile(sessions) for _ in range(3)])) == 1 waiting = (await scenario["call"]("GET", path)).json() assert waiting["status"] == "WAITING_APPROVAL" assert waiting["waiting"]["approval_id"] == confirmed[0]["approval_id"] assert await claim(sessions, 2) is None await decide(scenario, confirmed[0]) assert sum(await asyncio.gather(*[reconcile(sessions) for _ in range(3)])) == 1 resumed = (await scenario["call"]("GET", path)).json() assert resumed["run_id"] == str(UUID(bytes=run_id)) and resumed["status"] == "QUEUED" assert resumed["waiting"] is None and resumed["model_calls"] == 0 events = (await scenario["call"]("GET", conversation_path + "/messages")).json()["items"] assert [event["type"] for event in events].count("run.resumed") == 1 async with sessions() as session: for model, expected in ( (RunInputEvent, 2), (Approval, 1), (Ticket, 1), (ExternalOperation, 0) ): assert await session.scalar(select(func.count()).select_from(model).where( model.tenant_id == scenario["tenant"].bytes)) == expected assert (await claim(sessions, 2))[0] == run_id @pytest.mark.parametrize("decision", ["APPROVED", "REJECTED"]) async def test_decision_before_event_poll_uses_current_proof(scenario, decision): # noqa: F811 _, path, _, ticket, _ = await draft_wait(scenario) await decide(scenario, await confirm(scenario, ticket), decision) assert await reconcile(scenario["app"].state.sessions) == 1 assert (await scenario["call"]("GET", path)).json()["status"] == "QUEUED" assert await reconcile(scenario["app"].state.sessions) == 0 async def test_revision_replaces_wait_and_requires_new_confirmation(scenario): # noqa: F811 _, path, _, ticket, view = await draft_wait(scenario) confirmed = await confirm(scenario, ticket) sessions = scenario["app"].state.sessions await reconcile(sessions) revised = await scenario["call"]( "POST", f"/openapi/v1/after-sales/proposals/{ticket['proposal_id']}/revisions", payload={ "expected_version": confirmed["version"], "quantity": 1, "reason": "修订故障描述", }) assert revised.status_code == 200, revised.text assert await reconcile(sessions) == 1 waiting = (await scenario["call"]("GET", path)).json() assert waiting["status"] == "WAITING_USER" assert waiting["waiting"]["proposal_version"] == 2 assert waiting["waiting"]["interrupt_id"] != view["waiting"]["interrupt_id"] old = await scenario["call"]( "POST", f"/openapi/v1/approvals/{confirmed['approval_id']}/decisions", "reviewer", { "proposal_version": 1, "expected_version": confirmed["version"], "decision": "APPROVED", "reason": "旧版本", }) assert old.status_code == 409 await confirm(scenario, revised.json(), key="revised-confirm") assert await reconcile(sessions) == 1 assert (await scenario["call"]("GET", path)).json()["status"] == "WAITING_APPROVAL" async def test_rule_revocation_requires_revision_and_wait_expires(scenario): # noqa: F811 run_id, path, _, ticket, _ = await draft_wait(scenario) await confirm(scenario, ticket) sessions = scenario["app"].state.sessions response = await scenario["call"]( "POST", f"/admin/v1/stores/{scenario['store']}/exchange-rules/revoke", "agent", {"expected_version": 1, "reason": "当前规则撤销"}) assert response.status_code == 200, response.text assert await reconcile(sessions) == 1 result = (await scenario["call"]("GET", f"/openapi/v1/tickets/{ticket['ticket_id']}")).json() assert result["status"] == "NEEDS_REVISION" and result["approval_status"] == "INVALIDATED" assert (await scenario["call"]("GET", path)).json()["status"] == "WAITING_USER" async with sessions.begin() as session: (await session.get(AgentRun, run_id)).deadline_at = now() - timedelta(seconds=1) assert await claim(sessions, 2) is None assert (await scenario["call"]("GET", path)).json()["status"] == "FAILED" async def test_takeover_and_approver_revocation_prevent_resume(scenario): # noqa: F811 _, path, conversation, ticket, _ = await draft_wait(scenario) await confirm(scenario, ticket) sessions = scenario["app"].state.sessions await reconcile(sessions) response = await scenario["call"]("POST", conversation + "/takeover", "agent", {"expected_control_version": 1}) assert response.status_code == 200, response.text assert await reconcile(sessions) == 0 assert (await scenario["call"]("GET", path)).json()["status"] == "CANCELLED" # 新的独立运行验证审批完成后、恢复前的当前权限检查。 row = await tool_run(scenario, "create_exchange_proposal", scenario["proposal"]) async with sessions.begin() as session: (await session.get(AgentRun, row.id)).graph_version = GRAPH_VERSION run_id, token = await claim(sessions, 2) await build_tool_graph(sessions).ainvoke({"run_id": run_id, "token": token}) async with sessions() as session: run = await session.get(AgentRun, run_id) ticket_id = run.waiting["proposal_id"] ticket = (await scenario["call"]("GET", f"/openapi/v1/tickets/{ticket_id}")).json() await decide(scenario, await confirm(scenario, ticket)) async with sessions.begin() as session: await session.execute(delete(StoreGrant).where( StoreGrant.principal_id == scenario["accounts"]["reviewer"].bytes, StoreGrant.permission == "approvals:decide")) await reconcile(sessions) async with sessions() as session: run = await session.get(AgentRun, run_id) assert run.status == "FAILED" and run.error_code == "APPROVER_REVOKED" async def test_stale_snapshot_waits_and_changed_order_requires_revision(scenario): # noqa: F811 _, path, _, ticket, _ = await draft_wait(scenario) sessions = scenario["app"].state.sessions async with sessions.begin() as session: order = await session.get(OrderProjection, scenario["order_id"].bytes) order.observed_at = now() - timedelta(minutes=6) assert await reconcile(sessions) == 0 unchanged = (await scenario["call"]("GET", f"/openapi/v1/tickets/{ticket['ticket_id']}")).json() assert unchanged["status"] == "DRAFT" response = await scenario["call"]("PUT", f"/internal/v1/orders/{scenario['order_id']}", "connector", {**scenario["source"], "source_version": 2}) assert response.status_code == 200, response.text assert await reconcile(sessions) == 1 changed = (await scenario["call"]("GET", f"/openapi/v1/tickets/{ticket['ticket_id']}")).json() assert changed["status"] == "NEEDS_REVISION" assert (await scenario["call"]("GET", path)).json()["status"] == "WAITING_USER"