zhifu-platform/tests/test_agent_inputs.py

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)