zhifu-platform/src/zhifu/modules/conversations/api.py

import hashlib
from typing import Annotated
from uuid import UUID, uuid4

from fastapi import APIRouter, Depends, Path, Query, Request
from fastapi.security import HTTPBearer
from langchain_core.messages import AIMessage, HumanMessage, message_to_dict
from pydantic import Field
from sqlalchemy import select

from zhifu.core.api_docs import ConversationView, EventPage, EventView, Page, replies
from zhifu.modules.after_sales.api import IdempotencyKey
from zhifu.modules.after_sales.schemas import Command
from zhifu.modules.after_sales.service import require
from zhifu.modules.agent_runtime.models import AgentRun
from zhifu.modules.agent_runtime.service import queue_run
from zhifu.modules.conversations.models import Conversation, ConversationEvent
from zhifu.modules.conversations.service import access, emit, event_view, view
from zhifu.modules.identity.auth import authenticate, authorize_store, now
from zhifu.modules.memory.service import enqueue as enqueue_memory
from zhifu.modules.notifications.service import notify

router = APIRouter(
    prefix="/openapi/v1/conversations",
    tags=["会话"],
    dependencies=[Depends(HTTPBearer(auto_error=False))],
)


class Create(Command):
    store_id: UUID = Field(description="会话所属店铺，需要该店铺 orders:read 权限")


class Control(Command):
    expected_control_version: int = Field(ge=1, description="最近查询到的会话 control_version")


class Message(Control):
    text: str = Field(
        min_length=1,
        max_length=4000,
        description="客户消息或接管客服的回复",
        examples=["请帮我查询订单状态"],
    )


@router.get("", summary="分页查询店铺会话", responses=replies(Page[ConversationView]))
async def list_conversations(
    request: Request,
    store_id: Annotated[UUID, Query(description="查询范围所属店铺")],
    cursor: Annotated[UUID | None, Query(description="上一页返回的 next_cursor")] = None,
    limit: Annotated[int, Query(ge=1, le=100, description="每页最多返回数量")] = 20,
):
    """CUSTOMER 使用 orders:read 且只查询本人会话；STAFF 使用 conversations:read。

    按 UUID 升序分页，使用 next_cursor 请求下一页。每项返回实际 channel 和带 UTC 时区的
    created_at；创建时间不随消息或控制权变化更新。
    """
    async with request.app.state.sessions.begin() as session:
        auth = await authenticate(request, session)
        require(auth.kind in {"CUSTOMER", "STAFF"}, "FORBIDDEN", "需要客户或客服身份", 403)
        permission = "orders:read" if auth.kind == "CUSTOMER" else "conversations:read"
        await authorize_store(session, auth, store_id.bytes, permission)
        query = select(Conversation).where(
            Conversation.tenant_id == auth.tenant_id, Conversation.store_id == store_id.bytes
        )
        if auth.kind == "CUSTOMER":
            query = query.where(
                Conversation.subject_id == auth.actor_id, Conversation.channel == "h5")
        if cursor:
            query = query.where(Conversation.id > cursor.bytes)
        rows = (await session.scalars(query.order_by(Conversation.id).limit(limit + 1))).all()
        return {
            "items": [view(row) for row in rows[:limit]],
            "next_cursor": str(UUID(bytes=rows[limit - 1].id)) if len(rows) > limit else None,
        }


@router.post(
    "",
    status_code=201,
    summary="客户创建会话",
    responses=replies(ConversationView, 201, errors=(401, 403, 404, 409, 422)),
)
async def create(body: Create, request: Request, key: IdempotencyKey):
    """需要 CUSTOMER 身份和店铺 orders:read 权限，初始控制权为 AGENT。

    Idempotency-Key 在租户和客户范围内唯一。重试返回原会话当前状态，换店铺重用该键返回 409。
    channel 为 h5，created_at 是带 UTC 时区的会话创建时间，重复请求保持原创建时间。
    会话保存聊天与控制状态，每次创建 Agent 运行时读取店铺当前发布配置。
    """
    async with request.app.state.sessions.begin() as session:
        auth = await authenticate(request, session)
        require(auth.kind == "CUSTOMER", "FORBIDDEN", "需要客户身份", 403)
        await authorize_store(session, auth, body.store_id.bytes, "orders:read")
        existing = await session.scalar(
            select(Conversation).where(
                Conversation.tenant_id == auth.tenant_id,
                Conversation.subject_id == auth.actor_id,
                Conversation.request_key == key,
            )
        )
        if existing:
            require(
                existing.store_id == body.store_id.bytes, "IDEMPOTENCY_CONFLICT", "请求参数不同"
            )
            return view(existing)
        conversation = Conversation(
            id=uuid4().bytes,
            tenant_id=auth.tenant_id,
            subject_id=auth.actor_id,
            store_id=body.store_id.bytes,
            request_key=key,
            created_at=now(),
        )
        session.add(conversation)
        await session.flush()
        return view(conversation)


@router.get("/{conversation_id}", summary="查询会话控制状态", responses=replies(ConversationView))
async def get_conversation(conversation_id: UUID, request: Request):
    """客户可查询本人会话；客服需要 conversations:read。发送消息前读取 control_version。

    同时返回实际 channel 与带 UTC 时区的 created_at，供会话来源和创建时间展示。
    """
    async with request.app.state.sessions.begin() as session:
        auth = await authenticate(request, session)
        return view(await access(session, auth, conversation_id.bytes, read=True))


@router.get(
    "/{conversation_id}/messages", summary="按序号读取会话持久事件", responses=replies(EventPage)
)
async def history(
    conversation_id: UUID,
    request: Request,
    after: Annotated[int, Query(ge=0, description="仅返回严格大于此序号的事件")] = 0,
    limit: Annotated[int, Query(ge=1, le=100, description="本次最多返回的事件数量")] = 50,
):
    """身份要求同会话查询。items 包含消息、控制权变化及运行进度等全部持久事件。

    顶层 sequence 是会话最新序号。继续分页时使用 items 最后一项的 sequence，
    防止在事件较多时跳过尚未读取的消息。历史事件保留生成时的 control_version。
    created_at 是带 UTC 时区的事件原始创建时间；显示时间使用此字段，分页与去重继续使用 sequence。
    """
    async with request.app.state.sessions.begin() as session:
        auth = await authenticate(request, session)
        conversation = await access(session, auth, conversation_id.bytes, read=True)
        events = (
            await session.scalars(
                select(ConversationEvent)
                .where(
                    ConversationEvent.tenant_id == auth.tenant_id,
                    ConversationEvent.conversation_id == conversation.id,
                    ConversationEvent.sequence > after,
                )
                .order_by(ConversationEvent.sequence)
                .limit(limit)
            )
        ).all()
        return {"items": [event_view(event) for event in events], "sequence": conversation.sequence}


async def prior(session, conversation, auth, key, digest):
    event = await session.scalar(
        select(ConversationEvent).where(
            ConversationEvent.tenant_id == auth.tenant_id,
            ConversationEvent.conversation_id == conversation.id,
            ConversationEvent.actor_id == auth.actor_id,
            ConversationEvent.command_key == key,
        )
    )
    if event:
        require(event.request_hash == digest, "IDEMPOTENCY_CONFLICT", "请求参数不同")
    return event


@router.post(
    "/{conversation_id}/messages",
    status_code=201,
    summary="发送客户消息或客服回复",
    responses=replies(EventView, 201, errors=(401, 403, 404, 409, 422, 429, 503)),
)
async def message(conversation_id: UUID, body: Message, request: Request, key: IdempotencyKey):
    """客户只能发送本人会话消息；客服需要 conversations:reply 且已接管该会话。

    客户在 AGENT 模式发言会创建异步运行，payload.run_id 返回运行标识；HUMAN 模式仅保存消息。
    新运行读取店铺当前已发布版本，原会话的历史消息继续作为上下文。
    已接受的运行固定保存配置；幂等重试、补充资料、审批等待和重启恢复保留原运行配置。
    同一会话只允许一个活动运行，否则返回 RUN_IN_PROGRESS。
    Agent 运行遵循租户并发上限与交互额度预留；额度不足时排队，账户任务已满返回 429。
    每次模型请求前检查当日 token 预算；不足时运行失败，run.failed 事件包含
    code=TOKEN_BUDGET_EXCEEDED、MONEY_BUDGET_EXCEEDED 或 BILLING_PRICE_MISSING，
    客户可在预算或价格配置完成后以新幂等键重新发送。
    幂等键作用域为租户、会话及操作者；发送和控制权操作共用此作用域。
    响应 created_at 为带 UTC 时区的事件原始创建时间，重复请求返回相同值。
    """
    async with request.app.state.sessions.begin() as session:
        auth = await authenticate(request, session)
        conversation = await access(session, auth, conversation_id.bytes, "conversations:reply")
        return await accept_message(session, auth, conversation, body, key)


async def accept_message(session, auth, conversation, body, key):
    """H5与可信渠道共用消息服务，调用方持有身份及会话锁。"""
    digest = hashlib.sha256(body.model_dump_json().encode()).digest()
    previous = await prior(session, conversation, auth, key, digest)
    if previous:
        return event_view(previous)
    require(conversation.control_version == body.expected_control_version,
            "CONTROL_CONFLICT", "会话控制版本已变化")
    if auth.kind == "STAFF":
        require(conversation.control_owner == "HUMAN"
                and conversation.assigned_staff_id == auth.actor_id,
                "FORBIDDEN", "需要当前接管客服身份", 403)
    run_id = None
    if auth.kind == "CUSTOMER" and conversation.control_owner == "AGENT":
        require(conversation.active_run_id is None, "RUN_IN_PROGRESS", "当前回复仍在处理中")
        # 最近12条消息与24000字符上限在所有入口采用相同规则。
        events = (await session.scalars(select(ConversationEvent).where(
            ConversationEvent.tenant_id == auth.tenant_id,
            ConversationEvent.conversation_id == conversation.id,
            ConversationEvent.event_type.in_(
                ["message.user", "message.assistant", "message.staff"]),
        ).order_by(ConversationEvent.sequence.desc()).limit(12))).all()
        messages = []
        length = len(body.text)
        for event in events:
            text = event.payload["text"]
            if length + len(text) > 24000:
                break
            length += len(text)
            kind = HumanMessage if event.event_type == "message.user" else AIMessage
            messages.append(message_to_dict(kind(content=text)))
        messages.reverse()
        messages.append(message_to_dict(HumanMessage(content=body.text)))
        run = await queue_run(session, auth, conversation.store_id, "conv:" + uuid4().hex,
                              digest, messages)
        run.conversation_id = conversation.id
        run.control_version = conversation.control_version
        conversation.active_run_id = run.id
        run_id = str(UUID(bytes=run.id))
    event = emit(session, conversation,
                 "message.user" if auth.kind == "CUSTOMER" else "message.staff",
                 {"text": body.text, "run_id": run_id}, auth.actor_id, key, digest)
    if auth.kind == "CUSTOMER":
        await enqueue_memory(session, auth, conversation, event)
        if conversation.assigned_staff_id is not None:
            await notify(session, tenant_id=auth.tenant_id, store_id=conversation.store_id,
                event_key=f"message:{conversation.id.hex()}:{event.sequence}",
                permission="conversations:read", kind="客户消息", title="客户有新消息",
                summary="当前接待的会话收到客户消息，请查看最新内容。", page="conversation",
                action="查看客户消息", target_id=str(UUID(bytes=conversation.id)),
                recipient_id=conversation.assigned_staff_id)
    return event_view(event)


@router.post(
    "/{conversation_id}/{action}",
    summary="客服接管或释放会话",
    responses=replies(EventView, errors=(401, 403, 404, 409, 422)),
)
async def control(
    conversation_id: UUID,
    action: Annotated[str, Path(description="takeover 接管；release 交还 Agent，其他值返回 404")],
    body: Control,
    request: Request,
    key: IdempotencyKey,
):
    """需要 STAFF 身份及 conversations:manage 权限。release 仅允许当前接管客服调用。

    成功增加 control_version，取消旧活动运行，并产生 control.changed 事件。
    释放后需要客户再次发送消息才会创建新运行。迟到模型答复不能越过新控制版本提交。
    响应 created_at 为本次控制事件的创建时间；payload.created_at 为会话创建时间，均带 UTC 时区。
    """
    require(action in {"takeover", "release"}, "RESOURCE_NOT_FOUND", "操作不存在", 404)
    async with request.app.state.sessions.begin() as session:
        auth = await authenticate(request, session)
        require(auth.kind == "STAFF", "FORBIDDEN", "需要客服身份", 403)
        conversation = await access(session, auth, conversation_id.bytes, "conversations:manage")
        digest = hashlib.sha256((action + body.model_dump_json()).encode()).digest()
        previous = await prior(session, conversation, auth, key, digest)
        if previous:
            return event_view(previous)
        require(
            conversation.control_version == body.expected_control_version,
            "CONTROL_CONFLICT",
            "会话控制版本已变化",
        )
        if action == "release":
            require(
                conversation.control_owner == "HUMAN"
                and conversation.assigned_staff_id == auth.actor_id,
                "FORBIDDEN",
                "需要当前接管客服身份",
                403,
            )
        else:
            require(conversation.control_owner == "AGENT", "CONTROL_CONFLICT", "已有客服接管")
        if conversation.active_run_id:
            run = await session.get(AgentRun, conversation.active_run_id, with_for_update=True)
            if run.status in {"QUEUED", "RUNNING", "WAITING_USER", "WAITING_APPROVAL"}:
                run.status = "CANCELLED"
                run.version += 1
                run.waiting = None
                run.lease_until = None
            conversation.active_run_id = None
        conversation.control_version += 1
        conversation.control_owner = "HUMAN" if action == "takeover" else "AGENT"
        conversation.assigned_staff_id = auth.actor_id if action == "takeover" else None
        return event_view(
            emit(
                session,
                conversation,
                "control.changed",
                view(conversation),
                auth.actor_id,
                key,
                digest,
            )
        )