128 lines
5.8 KiB
Python
128 lines
5.8 KiB
Python
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.advisor_rollout_service import AdvisorRolloutService
|
|
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:
|
|
if request.agent_type == "advisor":
|
|
await AdvisorRolloutService(self.session).ensure_allowed(context)
|
|
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)
|