group_fqcd_jr/app/service/agent_run_application_service.py

import hashlib
import json
from collections.abc import Sequence
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.actor import VISITOR_ACTOR_TYPE, is_visitor
from app.core.contracts import AgentRequest, DomainEvent, RequestContext
from app.core.conversation_privacy import sanitize_customer_service_message
from app.core.customer_service_rules import chitchat_streak
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.conversation_repository import ConversationRepository
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"


def build_outbox_metadata(
    request: AgentRequest, prior_user_messages: Sequence[str], *,
    clarification_round: int = 0, session_context: Sequence[str] = (),
) -> dict[str, object]:
    """构造 Worker 使用的内部元数据，**不信任外部传入的客服计数**。

    客服的三个字段一律由服务端覆写：

    - ``chitchat_streak``：由已落库的历史重算（客户端写 0 就能绕过连续闲聊收口）；
    - ``clarification_round``：来自会话行，是 E5a 澄清的轮次上限依据；
    - ``session_context``：当前会话内的已脱敏上下文。

    其余 Agent 一直是把 ``request.metadata`` 原样透出，这里保持与它们相同的行为。

    上限收紧的原因：``model_copy(update=...)`` **不做校验**，越界值会被静默写入
    （``clarification_round`` 契约上是 ``le=2``、``session_context`` 是 ``max_length=6``）。
    """
    metadata = request.metadata
    if request.agent_type == "customer_service":
        metadata = metadata.model_copy(update={
            "chitchat_streak": chitchat_streak(prior_user_messages, request.message),
            "clarification_round": min(max(clarification_round, 0), 2),
            "session_context": tuple(session_context[-6:]),
        })
    return metadata.model_dump(mode="json")


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 _recent_user_messages(
        self, request: AgentRequest, user_id: int, *, limit: int = 3
    ) -> tuple[str, ...]:
        """取本会话最近若干条**已脱敏**的用户消息，供连续闲聊计数使用。

        必须走 `ConversationRepository`：它按 `customer_id == user_id` 过滤。一期实现只按
        `session_id` 取历史、跨主体可读——那正是 `F-01` 要修的缺陷本体，所以这里不是
        "恢复旧实现"，而是在**新链路**上用带主体过滤的查询重写（与 Worker 侧的
        `runtime._conversation_history` 同源同口径）。
        """
        rows = await ConversationRepository(self.session).messages(
            request.session_id, user_id, limit
        )
        messages = tuple(
            str(row.content or "").strip() for row in reversed(rows)
            if str(row.role) == "user" and str(row.content or "").strip()
        )
        return messages[-limit:]

    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
        # 客服原文不进入会话与异步链路：**先脱敏**，再用脱敏后的形状参与幂等哈希——
        # 否则同一请求的两次提交会因「原文 vs 脱敏文本」算出两个哈希而互相冲突。
        stored_request = request
        if request.agent_type == "customer_service":
            stored_request = request.model_copy(update={
                "message": sanitize_customer_service_message(request.message)
            })
        user_id = int(context.user_id)
        request_hash = hashlib.sha256(
            json.dumps(stored_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)

            clarification_round = (
                session_row.clarification_round if session_row is not None else 0
            )
            prior_user_messages = (
                await self._recent_user_messages(request, user_id)
                if request.agent_type == "customer_service"
                else ()
            )
            outbox_metadata = build_outbox_metadata(
                stored_request, prior_user_messages,
                clarification_round=clarification_round,
            )

            trace_id = context.trace_id
            message = ConversationMessage(
                session_id=request.session_id, customer_id=user_id, portal="api",
                role="user", content=stored_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,
                    "actor_type": VISITOR_ACTOR_TYPE if is_visitor(context) else "authenticated",
                    "metadata": outbox_metadata,
                },
                occurred_at=now,
            ))
            return RunAccepted(run_id, trace_id)