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)