group_fqcd_jr/app/service/agent_run_application_service.py

import hashlib
import json
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
from uuid import uuid4

from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.contracts import AgentRequest, DomainEvent, RequestContext
from app.core.errors import (
    ForbiddenAgentError,
    IdempotencyConflictError,
    SessionNotAccessibleError,
)
from app.model.audit import InteractionAudit
from app.model.conversation import ConversationMessage
from app.model.platform import AgentRun, RequestIdempotency
from app.model.session import ConversationSession
from app.repository.outbox_repository import OutboxRepository
from app.service.agent.bootstrap import get_agent_factory
from app.service.agent.factory import AgentFactory


@dataclass(frozen=True)
class RunAccepted:
    run_id: str
    trace_id: str
    status: str = "queued"


class AgentRunApplicationService:
    def __init__(self, session: AsyncSession, factory: AgentFactory | None = None) -> None:
        self.session = session
        self.factory = factory if factory is not None else get_agent_factory()

    async def accept(self, request: AgentRequest, context: RequestContext) -> RunAccepted:
        try:
            self.factory.authorize(request.agent_type, context)
        except ForbiddenAgentError:
            async with self.session.begin():
                self.session.add(InteractionAudit(
                    actor_type="user", actor_id=int(context.user_id), portal=context.portal,
                    action_type="agent.access_denied", session_id=request.session_id,
                    detail={"agent_type": request.agent_type, "trace_id": context.trace_id},
                    created_at=datetime.now(UTC).replace(tzinfo=None),
                ))
            raise
        user_id = int(context.user_id)
        request_hash = hashlib.sha256(
            json.dumps(request.model_dump(mode="json"), sort_keys=True).encode("utf-8")
        ).hexdigest()
        now = datetime.now(UTC).replace(tzinfo=None)
        async with self.session.begin():
            session_row = await self.session.scalar(
                select(ConversationSession).where(
                    ConversationSession.session_id == request.session_id,
                    ConversationSession.user_id == user_id,
                ).with_for_update()
            )
            if session_row is not None:
                if session_row.status != "active" or session_row.agent_type != request.agent_type:
                    raise ForbiddenAgentError("会话不可用于当前 Agent")
                session_row.message_count += 1
                session_row.last_active_at = now
            owner = await self.session.scalar(
                select(ConversationMessage.customer_id)
                .where(ConversationMessage.session_id == request.session_id)
                .where(ConversationMessage.customer_id.is_not(None))
                .limit(1)
            )
            if owner is not None and owner != user_id:
                raise SessionNotAccessibleError("会话不属于当前用户")
            existing = await self.session.scalar(
                select(RequestIdempotency).where(
                    RequestIdempotency.user_id == user_id,
                    RequestIdempotency.agent_type == request.agent_type,
                    RequestIdempotency.idempotency_key == request.idempotency_key,
                )
            )
            if existing is not None:
                if existing.request_hash != request_hash:
                    raise IdempotencyConflictError("同一幂等键对应不同请求")
                run = await self.session.scalar(
                    select(AgentRun).where(AgentRun.idempotency_id == existing.id)
                )
                if run is None:
                    raise RuntimeError("idempotency record has no run")
                return RunAccepted(run.run_id, run.trace_id, run.status)

            trace_id = context.trace_id
            message = ConversationMessage(
                session_id=request.session_id, customer_id=user_id, portal="api",
                role="user", content=request.message, trace_id=trace_id, created_at=now,
            )
            self.session.add(message)
            await self.session.flush()
            idem = RequestIdempotency(
                user_id=user_id, session_id=request.session_id, agent_type=request.agent_type,
                idempotency_key=request.idempotency_key, request_hash=request_hash,
                trace_id=trace_id, expire_at=now + timedelta(hours=24),
                created_at=now, updated_at=now,
            )
            self.session.add(idem)
            try:
                await self.session.flush()
            except IntegrityError as exc:
                raise IdempotencyConflictError("幂等键正在被并发请求占用") from exc
            run_id = str(uuid4())
            run = AgentRun(
                run_id=run_id, idempotency_id=idem.id, session_id=request.session_id,
                user_id=user_id, agent_type=request.agent_type, trace_id=trace_id,
                request_message_id=message.id, created_at=now, updated_at=now,
            )
            self.session.add(run)
            await self.session.flush()
            await OutboxRepository(self.session).append(DomainEvent(
                event_id=str(uuid4()), event_type="agent.run_requested", aggregate_type="agent_run",
                aggregate_id=run_id, trace_id=trace_id,
                payload={"run_id": run_id, "metadata": request.metadata.model_dump(mode="json")},
                occurred_at=now,
            ))
            return RunAccepted(run_id, trace_id)