group_fqcd_jr/tests/integration/test_run_cancellation_mysql.py

"""取消语义的 MySQL 集成测试（文档 §6.4）。

覆盖三条真实链路：
1. 取消 `queued` 运行 → `202`，`agent_run.status=cancel_requested`，
   且原请求在 `request_idempotency` 落 `failed + RUN_CANCELLED`；
2. **重复取消**（换幂等键、真实 HTTP）→ 返回同一状态，**不报错**；
3. 已成功 / 已失败的运行取消 → `409 RUN_NOT_CANCELLABLE`。

依赖真实 MySQL；所有写入在 finally 中按 session_id 清理。
"""

from uuid import uuid4

import httpx
import pytest
from sqlalchemy import delete, select

from app.api.dependencies.auth import build_request_context
from app.core.contracts import AgentRequest, RequestContext
from app.infrastructure.db import SessionFactory
from app.main import create_app
from app.model.audit import InteractionAudit
from app.model.conversation import ConversationMessage
from app.model.platform import AgentRun, DomainEventOutbox, OutboxDelivery, RequestIdempotency
from app.service.agent_run_application_service import AgentRunApplicationService
from app.worker.runtime import WorkerRuntime

pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("acceptance_registry")]


async def state_of(run_id: str) -> tuple[str, str | None, str]:
    async with SessionFactory() as session:
        run = await session.scalar(select(AgentRun).where(AgentRun.run_id == run_id))
        assert run is not None
        idem = await session.get(RequestIdempotency, run.idempotency_id)
        assert idem is not None
        return run.status, idem.error_code, idem.status


async def cleanup(session_id: str, run_id: str) -> None:
    async with SessionFactory() as session, session.begin():
        event_ids = select(DomainEventOutbox.event_id).where(
            DomainEventOutbox.aggregate_id == run_id)
        await session.execute(delete(OutboxDelivery).where(OutboxDelivery.event_id.in_(event_ids)))
        await session.execute(delete(DomainEventOutbox).where(
            DomainEventOutbox.aggregate_id == run_id))
        await session.execute(delete(InteractionAudit).where(
            InteractionAudit.session_id == session_id))
        await session.execute(delete(AgentRun).where(AgentRun.session_id == session_id))
        await session.execute(delete(RequestIdempotency).where(
            RequestIdempotency.session_id == session_id))
        await session.execute(delete(ConversationMessage).where(
            ConversationMessage.session_id == session_id))


def context() -> RequestContext:
    return RequestContext(user_id="1", trace_id=str(uuid4()), roles=("customer",),
                          permissions=("agent:run", "agent:cancel"))


def client_for(ctx: RequestContext) -> httpx.AsyncClient:
    application = create_app()
    application.dependency_overrides[build_request_context] = lambda: ctx
    return httpx.AsyncClient(transport=httpx.ASGITransport(app=application),
                             base_url="http://test")


async def submit(ctx: RequestContext, session_id: str) -> str:
    async with SessionFactory() as session:
        accepted = await AgentRunApplicationService(session).accept(
            AgentRequest(agent_type="customer_service", message="cancel integration",
                         session_id=session_id, idempotency_key=str(uuid4())),
            ctx,
        )
    return accepted.run_id


async def test_cancel_is_idempotent_and_terminates_original_request() -> None:
    ctx = context()
    session_id = f"cancel-{uuid4()}"
    run_id = ""
    try:
        run_id = await submit(ctx, session_id)
        async with client_for(ctx) as client:
            first = await client.post(f"/api/v1/agent-runs/{run_id}/cancellations",
                                      json={"reason": "user_cancelled"},
                                      headers={"Idempotency-Key": str(uuid4())})
            # 换一个幂等键重复取消：必须幂等返回同一状态，而不是 409。
            second = await client.post(f"/api/v1/agent-runs/{run_id}/cancellations",
                                       json={"reason": "user_cancelled"},
                                       headers={"Idempotency-Key": str(uuid4())})

        assert first.status_code == 202
        assert second.status_code == 202
        assert first.json()["data"] == second.json()["data"]
        assert first.json()["data"]["status"] == "cancel_requested"
        assert first.json()["data"]["run_id"] == run_id

        status, error_code, idem_status = await state_of(run_id)
        assert status == "cancel_requested"
        # 文档 §6.4：原请求以 failed + RUN_CANCELLED 结束，不扩展状态枚举。
        assert (idem_status, error_code) == ("failed", "RUN_CANCELLED")
    finally:
        if run_id:
            await cleanup(session_id, run_id)


async def test_cancel_after_worker_completion_is_not_cancellable(
    acceptance_registry,
) -> None:
    ctx = context()
    session_id = f"cancel-done-{uuid4()}"
    run_id = ""
    runtime = WorkerRuntime(
        acceptance_registry, resolve_identity=lambda value: _identity(value, ctx)
    )
    try:
        run_id = await submit(ctx, session_id)
        await runtime.execute(run_id)
        status, _error_code, _idem = await state_of(run_id)
        assert status == "succeeded"

        async with client_for(ctx) as client:
            response = await client.post(f"/api/v1/agent-runs/{run_id}/cancellations",
                                         json={"reason": "user_cancelled"},
                                         headers={"Idempotency-Key": str(uuid4())})

        assert response.status_code == 409
        assert response.json()["error"]["code"] == "RUN_NOT_CANCELLABLE"
        assert response.json()["error"]["retryable"] is False
    finally:
        if run_id:
            await cleanup(session_id, run_id)


async def test_worker_advances_cancel_requested_to_cancelled_with_idempotency_marker() -> None:
    """worker 真正落 `cancelled` 时，幂等记录仍必须是 `failed + RUN_CANCELLED`。"""
    ctx = context()
    session_id = f"cancel-worker-{uuid4()}"
    run_id = ""
    try:
        run_id = await submit(ctx, session_id)
        async with client_for(ctx) as client:
            response = await client.post(f"/api/v1/agent-runs/{run_id}/cancellations",
                                         json={"reason": "user_cancelled"},
                                         headers={"Idempotency-Key": str(uuid4())})
        assert response.status_code == 202

        runtime = WorkerRuntime(resolve_identity=lambda value: _identity(value, ctx))
        await runtime.execute(run_id)

        status, error_code, idem_status = await state_of(run_id)
        assert status == "cancelled"
        assert (idem_status, error_code) == ("failed", "RUN_CANCELLED")
    finally:
        if run_id:
            await cleanup(session_id, run_id)


async def _identity(value: RequestContext, fallback: RequestContext) -> RequestContext:
    del value
    return fallback