hunjing/backend/app/conversations/routes.py

"""对话域路由：创建会话、发送消息（幂等原子）与 Run SSE。"""

import json
from collections.abc import AsyncIterator
from typing import Annotated

from fastapi import APIRouter, Depends, Header, HTTPException, Request, status
from fastapi.responses import JSONResponse, StreamingResponse
from sqlalchemy import select

from app.auth.dependencies import get_current_user, require_state_change
from app.conversations import service
from app.conversations.schemas import MessageCreatedOut, MessageIn, MessageOut, RunOut
from app.db.session import DbSession
from app.models import PersonalDomain, Run, User

router = APIRouter(prefix="/v1/conversations", tags=["conversations"])


async def _domain_of(db: DbSession, user: User) -> PersonalDomain:
    result = await db.execute(
        select(PersonalDomain).where(PersonalDomain.user_id == user.id).limit(1)
    )
    domain = result.scalar_one_or_none()
    if domain is None:
        raise HTTPException(status.HTTP_409_CONFLICT, "用户缺少个人域")
    return domain


@router.post("", status_code=201)
async def create_conversation(
    db: DbSession,
    user: Annotated[User, Depends(require_state_change)],
    request: Request,
    body: dict[str, str],
) -> dict[str, str]:
    domain = await _domain_of(db, user)
    conversation = await service.create_conversation(
        db, domain.id, body.get("title", "未命名对话"), getattr(request.state, "trace_id", None)
    )
    return {"id": conversation.id, "title": conversation.title}


@router.post(
    "/{conversation_id}/messages",
    status_code=201,
    response_model=None,
)
async def post_message(
    conversation_id: str,
    db: DbSession,
    user: Annotated[User, Depends(require_state_change)],
    request: Request,
    body: MessageIn,
    idempotency_key: Annotated[str, Header(alias="Idempotency-Key")],
) -> MessageCreatedOut | JSONResponse:
    if not idempotency_key or len(idempotency_key) > 128:
        raise HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY, "Idempotency-Key 不合法")
    domain = await _domain_of(db, user)
    message, run, replayed = await service.replay_or_execute_message(
        db,
        domain.id,
        conversation_id,
        idempotency_key,
        body,
        getattr(request.state, "trace_id", None),
    )
    payload = MessageCreatedOut(
        message=MessageOut(
            id=message.id,
            conversation_id=message.conversation_id,
            branch_id=message.branch_id,
            role=message.role,
            content=message.content,
            parser_version=message.parser_version,
            created_at=message.created_at.isoformat(),
        ),
        run=(
            RunOut(
                id=run.id,
                conversation_id=run.conversation_id,
                status=run.status,
                created_at=run.created_at.isoformat(),
                completed_at=run.completed_at.isoformat() if run.completed_at else None,
            )
            if run is not None
            else None
        ),
    )
    if replayed:
        # 重放成功按契约返回 200 与原资源（避开装饰器固定 201）
        return JSONResponse(status_code=200, content=json.loads(payload.model_dump_json()))
    return payload


run_router = APIRouter(prefix="/v1/runs", tags=["runs"])


@run_router.get("/{run_id}/events")
async def run_events(
    run_id: str,
    db: DbSession,
    user: Annotated[User, Depends(get_current_user)],
) -> StreamingResponse:
    """Run SSE：返回当前快照与终止状态（垂直切片空状态闭环）。"""
    result = await db.execute(select(Run).where(Run.id == run_id))
    run = result.scalar_one_or_none()
    if run is None:
        raise HTTPException(status.HTTP_404_NOT_FOUND, "Run 不存在")
    domain = await _domain_of(db, user)
    if run.personal_domain_id != domain.id:
        raise HTTPException(status.HTTP_404_NOT_FOUND, "Run 不存在")

    terminal = run.status in {"completed", "failed", "cancelled"}

    async def event_stream() -> AsyncIterator[str]:
        def envelope(event_type: str, data: dict[str, object], sequence: int) -> str:
            payload = json.dumps(
                {"type": event_type, "data": data, "sequence": sequence}, ensure_ascii=False
            )
            return f"data: {payload}\n\n"

        yield envelope("run.status.changed", {"run_id": run.id, "status": run.status}, 0)
        if terminal:
            yield envelope(
                "message.completed",
                {
                    "run_id": run.id,
                    "message_ids": [],
                    "final_status": run.status,
                },
                1,
            )

    return StreamingResponse(
        event_stream(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-store"},
    )