zhifu-platform/tests/test_business_wait.py

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"