import asyncio import contextlib import logging from collections.abc import Awaitable, Callable from dataclasses import dataclass from datetime import UTC, datetime, timedelta from typing import Any, Protocol, cast from uuid import uuid4 from sqlalchemy import select, update from app.core.config import Settings, get_settings from app.core.contracts import AgentRequest, AgentRequestMetadata, AgentResult, RequestContext from app.core.errors import AgentError, RecoverableAgentError, RunLeaseLostError from app.infrastructure.db import SessionFactory from app.model.audit import InteractionAudit from app.model.conversation import ConversationMessage from app.model.platform import AgentRun, DomainEventOutbox, HandoverTicket, RequestIdempotency from app.repository.agent_run_repository import AgentRunRepository from app.service.agent.bootstrap import ( get_agent_factory, get_memory_cache_adapter, get_model_service, ) from app.service.agent.executor import AgentExecutor from app.service.agent.factory import AgentFactory from app.service.agent_persistence_service import AgentPersistenceService from app.service.customer_service_session_memory_service import ( CustomerServiceSessionMemory, CustomerServiceSessionTurn, build_customer_service_session_memory, ) from app.service.identity_service import IdentityService from app.service.memory_extraction_service import ( ExtractionEndpointResolver, MemoryExtractionService, ) from app.service.memory_lifecycle_service import MemoryLifecycleService, Mode from app.service.memory_recall_service import MemoryRecallService from app.service.memory_service import CacheDeleteAdapter, MemoryService from app.service.memory_taxonomy import BUSINESS_EVENT_TYPES from app.service.model_gateway import DatabaseModelEndpointResolver, ModelGenerationService from app.worker.customer_profile_candidate_worker import CustomerProfileCandidateWorker from app.worker.episode_worker import ( EpisodeConsumptionResult, EpisodeExtractionConsumer, EpisodeWorker, ) from app.worker.memory_extraction_worker import MemoryExtractionWorker from app.worker.outbox_worker import OutboxWorker logger = logging.getLogger(__name__) # episode 聚合按轮次节流：每 N 轮顺带处理一次已静默的会话片段， # 避免每轮轮询都做一次客户维度的 distinct 查询。 EPISODE_INTERVAL_ROUNDS = 30 # 一轮内最多消费的事件条数。`OutboxWorker.publish_one` 每次只领一条（单条事务让幂等与 # skip_locked 锁语义保持简单），所以清空速度原先等于轮询速度：262 条积压要 262 轮。 OUTBOX_DISPATCH_LIMIT = 10 # 单轮消费的片段条数上限：批处理必须可中断，不能一次吃完整库。 EPISODE_CONSUME_LIMIT = 20 PROJECTION_AUDIT_ACTION = "memory.projection_cleanup" @dataclass(frozen=True) class ProjectionCleanupOutcome: """投影清理结果；`cleaned=False` 时 `detail` 必须说明真实缺口。""" cleaned: bool detail: str = "" class ProjectionCleaner(Protocol): """Milvus/Neo4j 投影清理边界：由组装层注入，未注入即显式降级（绝不伪造成功）。""" async def cleanup(self, *, memory_uuid: str, operation: str) -> ProjectionCleanupOutcome: ... class WorkerRuntime: def __init__( self, factory: AgentFactory | None = None, settings: Settings | None = None, resolve_identity: Callable[[RequestContext], Awaitable[RequestContext]] | None = None, model_service: ModelGenerationService | None = None, endpoint_resolver: ExtractionEndpointResolver | None = None, memory_cache: CacheDeleteAdapter | None = None, projection_cleaner: ProjectionCleaner | None = None, session_memory: CustomerServiceSessionMemory | None = None, ) -> None: self.factory = factory if factory is not None else get_agent_factory() self.settings = settings or get_settings() self.resolve_identity = resolve_identity or IdentityService().resolve # 记忆抽取必须走与业务 Agent 相同的模型路由入口。默认取生产装配 # （app/service/agent/bootstrap.py），允许构造参数注入替身模型与端点解析器， # 使验收探针不依赖库中真实模型端点配置。 self.model_service = model_service if model_service is not None else get_model_service() self.memory_extraction = MemoryExtractionService( self.model_service, endpoint_resolver if endpoint_resolver is not None else DatabaseModelEndpointResolver(), ) # 召回热缓存适配器：任何记忆写入（事件消费、片段消费）都必须让该客户的 # 召回热缓存失效，否则新记忆在 TTL 内召回不到。默认取生产装配，可注入替身。 self.memory_cache: CacheDeleteAdapter | None = ( memory_cache if memory_cache is not None else get_memory_cache_adapter() ) # Milvus/Neo4j 删除客户端：当前组装层没有提供（bootstrap 只装配召回用的读适配器）， # 因此默认 None = 投影清理显式降级并留痕，绝不写成"删除成功"。 self.projection_cleaner = projection_cleaner # 该组件是客服当前会话 Redis List，不是长期记忆召回缓存或客户画像。 self.session_memory = ( session_memory if session_memory is not None else build_customer_service_session_memory() ) # episode 聚合是低频批处理，按轮次节流而不是每轮都查。 self._episode_rounds = 0 async def restore_context( self, *, actor_type: str, actor_id: str, trace_id: str ) -> RequestContext: """按已验证的内部事件身份恢复执行上下文。""" identity = RequestContext(user_id=actor_id, trace_id=trace_id) if actor_type == "visitor": return identity.model_copy(update={ "roles": ("visitor",), # 与访客 JWT 对齐，只恢复公开 Agent 和公开知识的最小权限。 "permissions": ("agent:run", "knowledge:query"), "data_scope": "public", }) return await self.resolve_identity(identity) @staticmethod def should_request_memory_extraction( *, agent_type: str, context: RequestContext, message: str, result: AgentResult, business_events: tuple[str, ...] | list[str], ) -> bool: """只允许非客服 Agent 的已登录用户业务事实进入客户记忆抽取队列。""" # 客服只使用未来受控的会话短期上下文，绝不沉淀为长期画像或客户偏好。 if agent_type == "customer_service" or "visitor" in context.roles: return False return MemoryService.should_extract_memory( conversation_content=message, role="user", tool_result=any(call.status == "succeeded" for call in result.result.tool_calls), event_type=business_events[0] if business_events else None, signals=MemoryService.detect_memory_signals(message), ) @staticmethod def should_request_profile_candidate( *, agent_type: str, context: RequestContext, message: str, ) -> bool: """客服只为已登录客户生成候选，不为访客生成正式画像数据。""" if agent_type != "customer_service" or "visitor" in context.roles: return False if not {"customer", "authenticated_user"}.intersection(context.roles): return False if context.data_scope != "self": return False # 只有明确的长期偏好、约束或目标才进入候选队列，普通问答不触发模型抽取。 return bool(MemoryService.detect_memory_signals(message)) async def dispatch_one(self, *, run_id: str | None = None) -> bool: # Outbox acknowledges a durable SQL queue entry, not an in-memory task. async with SessionFactory() as session: async def dispatch(payload: dict[str, Any]) -> None: run = await AgentRunRepository(session).get(str(payload["run_id"])) if run is None: raise ValueError("run not found") async def dispatch_memory_extraction(payload: dict[str, Any]) -> None: if "message_id" not in payload or "customer_id" not in payload: raise ValueError("memory extraction payload is incomplete") # 幂等键只认事件 id，由 worker 自己按 payload 回查，避免调用方漏传。 # 注入召回缓存适配器：写入生效后立即失效该客户的热缓存。 await MemoryExtractionWorker( session, extractor=self.memory_extraction, cache=self.memory_cache ).handle(payload) async def dispatch_profile_candidate(payload: dict[str, Any]) -> None: if "message_id" not in payload or "customer_id" not in payload: raise ValueError("profile candidate payload is incomplete") await CustomerProfileCandidateWorker( session, extractor=self.memory_extraction, cache=self.memory_cache ).handle(payload) async def dispatch_run_completed(payload: dict[str, Any]) -> None: # 结果消息与审计已由 complete_run 同事务落库，此事件只承担 # "运行已完成"的对外通知职责。当前没有独立外部消费者， # 这里显式消费以免事件永久滞留；接入推送链路时在此处扩展。 if not str(payload.get("run_id", "")): raise ValueError("agent.run_completed payload is incomplete") async def dispatch_cache_invalidate(payload: dict[str, Any]) -> None: await self._invalidate_config_cache(payload) async def dispatch_projection_cleanup(payload: dict[str, Any]) -> None: # memory.invalidated / memory.deleted 由 MemoryLifecycleService 按 # memory_uuid 写入，这里做幂等的投影清理（Milvus 向量、Neo4j 关系）。 await self._cleanup_projection(payload, session=session) async def dispatch_memory_deletion(payload: dict[str, Any]) -> None: # 客户级联失效/删除的公共入口：业务侧（用户注销、合规删除令） # 只写 memory.deletion_requested 事件，不直接操作记忆表。 customer_id = payload.get("customer_id") if not customer_id: raise ValueError("memory.deletion_requested payload is incomplete") mode = str(payload.get("mode", "invalidate")) if mode not in {"invalidate", "delete"}: raise ValueError("memory.deletion_requested mode is invalid") await MemoryLifecycleService(session).run( int(customer_id), mode=cast("Mode", mode), reason=str(payload.get("reason", "customer_lifecycle")), trace_id=str(payload.get("trace_id", "")), ) async def dispatch_handover_queue_ready(payload: dict[str, Any]) -> None: # 事件成功消费只表示管理员队列可查看，不能修改 pending 状态，更不能向用户 # 表述人工已经接单、处理或承诺时效。 ticket_no = str(payload.get("ticket_no", "")).strip() if not ticket_no: raise ValueError("conversation.transfer_requested payload is incomplete") ticket = await session.scalar( select(HandoverTicket).where(HandoverTicket.ticket_no == ticket_no) ) if ticket is None: raise ValueError("handover ticket not found") session.add(InteractionAudit( actor_type="system", actor_id=None, target_customer_id=ticket.customer_id, session_id=ticket.session_id, portal="worker", action_type="handover.queue_ready", detail={ "ticket_no": ticket.ticket_no, "source_agent": ticket.source_agent, "reason_code": ticket.reason_code, "ticket_status": ticket.status, }, created_at=datetime.now(UTC).replace(tzinfo=None), )) await session.flush() handlers: dict[str, Callable[[dict[str, Any]], Awaitable[None]]] = { "agent.run_requested": dispatch, "memory.extraction_requested": dispatch_memory_extraction, "customer_profile.candidate_requested": dispatch_profile_candidate, "agent.run_completed": dispatch_run_completed, "config.cache_invalidate_requested": dispatch_cache_invalidate, "memory.deletion_requested": dispatch_memory_deletion, # 投影清理：这两类事件此前没有消费者，永久 pending。 "memory.invalidated": dispatch_projection_cleanup, "memory.deleted": dispatch_projection_cleanup, "conversation.transfer_requested": dispatch_handover_queue_ready, } return await OutboxWorker(session, handlers).publish_one(aggregate_id=run_id) async def dispatch_batch(self, *, limit: int = OUTBOX_DISPATCH_LIMIT) -> int: """一轮内尽量多消费事件，返回实际消费条数。 单条领取的语义与幂等边界都不变，只是不再让"每轮一条"限制清空速度： 队列空时提前退出，因此稳态下与原来的一次调用开销相同，积压时才提速。 """ consumed = 0 for _ in range(max(1, limit)): if not await self.dispatch_one(): break consumed += 1 return consumed async def _invalidate_config_cache(self, payload: dict[str, Any]) -> None: """删除发布配置与记忆召回热缓存键；Redis 不可用时只记录告警，不阻塞事件消费。 记忆召回热缓存的键必须由 `MemoryRecallService.cache_keys` 枚举：此前这里手写的 `mem:mid:hot:{customer_id}` 并不是召回缓存的真实前缀（真实前缀是 `mem:recall`）， 失效动作一直打在并不存在的键上。事件 payload 里没有查询摘要与 limit，客户级 枚举是唯一可靠的失效方式（召回热缓存本身也只服务常见参数组合）。 """ release_id = payload.get("release_id") raw_customer_id = payload.get("customer_id") keys: list[str] = [] if raw_customer_id: keys.extend(MemoryRecallService.cache_keys(int(raw_customer_id))) if release_id is not None: keys.append(f"config:release:{release_id}") if not keys: return client = await self._redis_client() if client is None: return try: await client.delete(*keys) except Exception: logger.warning("config cache invalidation degraded keys=%s", ",".join(keys)) finally: await client.aclose() async def _cleanup_projection(self, payload: dict[str, Any], *, session: Any) -> None: """幂等清理一条记忆的派生投影（Milvus 向量、Neo4j 关系）。 为什么选择"显式降级 + 留痕"而不是假装成功：当前组装层没有提供任何删除客户端 （`bootstrap` 只装配了召回用的 Milvus 读适配器，Neo4j 连读适配器都没有）， 伪造"已删除"会让合规删除令在投影侧静默失效。因此无客户端时只做两件事： 记录告警，并写一条 `interaction_audit`（`status=skipped_no_client`），事件本身 照常标记为已消费——权威库（MySQL）状态已经正确，投影是可重建的派生数据， 让事件永久 pending 只会阻塞队列里其它事件。 注入 `projection_cleaner` 后同一入口执行真实删除，审计只记录适配器返回的真实 结论（`cleaned` 为假即写 `skipped`），不做任何"假定成功"的兜底。 """ memory_uuid = str(payload.get("memory_uuid", "")).strip() if not memory_uuid: raise ValueError("memory projection cleanup payload is incomplete") operation = str(payload.get("operation", "invalidate")) raw_customer_id = payload.get("customer_id") customer_id = int(raw_customer_id) if raw_customer_id else None cleaner = self.projection_cleaner if cleaner is None: logger.warning( "projection cleanup degraded: milvus/neo4j delete client not configured " "memory_uuid=%s operation=%s", memory_uuid, operation, ) await self._audit_projection( session, memory_uuid, customer_id, operation, status="skipped_no_client", reason="projection delete client not configured", ) return outcome = await cleaner.cleanup(memory_uuid=memory_uuid, operation=operation) if outcome.cleaned: logger.info("projection cleanup done memory_uuid=%s operation=%s", memory_uuid, operation) else: logger.warning("projection cleanup degraded memory_uuid=%s reason=%s", memory_uuid, outcome.detail) await self._audit_projection( session, memory_uuid, customer_id, operation, status="cleaned" if outcome.cleaned else "skipped", reason=outcome.detail, ) async def _audit_projection( self, session: Any, memory_uuid: str, customer_id: int | None, operation: str, *, status: str, reason: str, ) -> None: """投影清理留痕：审计行与事件同事务提交，降级状态在库里可查。""" session.add(InteractionAudit( actor_type="system", actor_id=None, target_customer_id=customer_id, session_id=None, portal=None, action_type=PROJECTION_AUDIT_ACTION, detail={ "memory_uuid": memory_uuid, "operation": operation, "status": status, "reason": reason, }, created_at=datetime.now(UTC).replace(tzinfo=None), )) await session.flush() async def _redis_client(self) -> Any: try: from redis.asyncio import Redis except ImportError: logger.warning("redis client unavailable; cache invalidation skipped") return None client: Any = Redis.from_url( self.settings.redis_url, socket_connect_timeout=self.settings.redis_connect_timeout_seconds, socket_timeout=self.settings.redis_connect_timeout_seconds, ) return client async def run_once(self) -> bool: dispatched = await self.dispatch_batch() > 0 self._episode_rounds += 1 if self._episode_rounds % EPISODE_INTERVAL_ROUNDS == 0: # 会话片段聚合：内部幂等（content_hash 唯一键），失败只告警， # 不得影响 run 的处理与轮询节奏。 try: await self.aggregate_episodes() except Exception: logger.warning("episode aggregation failed", exc_info=True) # 聚合之后立刻消费：片段只有被消费才会变成记忆，否则"待提取"永久滞留。 try: await self.consume_episodes() except Exception: logger.warning("episode extraction failed", exc_info=True) now = datetime.now(UTC).replace(tzinfo=None) async with SessionFactory() as session: run_id = await session.scalar(select(AgentRun.run_id).where( AgentRun.status.in_(("queued", "running", "cancel_requested")), (AgentRun.locked_until.is_(None) | (AgentRun.locked_until < now)), ).order_by(AgentRun.created_at).limit(1)) if run_id is None: return dispatched return await self.execute(run_id) or dispatched async def aggregate_episodes(self, *, customer_limit: int = 50) -> int: """把已静默的会话片段聚合为 episode，返回新写入的片段数。 可重复调用：同一片段的 `content_hash` 命中唯一键时直接跳过， 不需要额外的幂等表。 """ # 必须由调用方开启并提交事务：EpisodeWorker 只做 add/flush（savepoint 幂等）， # 自己不提交。漏掉提交会同时出现"返回已插入"与"表里没有行"。 async with SessionFactory() as session, session.begin(): # customer_id 列可空：SQL 的 is_not(None) 不足以让类型检查器收窄， # 这里显式再过滤一次。 raw_customers = list(await session.scalars( select(ConversationMessage.customer_id) .where(ConversationMessage.customer_id.is_not(None)) .distinct() .limit(max(1, customer_limit)) )) customers = [int(item) for item in raw_customers if item is not None] worker = EpisodeWorker(session) inserted = 0 for customer_id in customers: result = await worker.aggregate(customer_id) inserted += len(result.inserted) return inserted async def consume_episodes( self, *, limit: int = EPISODE_CONSUME_LIMIT ) -> EpisodeConsumptionResult: """消费一批待提取片段（低频批处理），把片段提升为长期记忆。 必须由调用方开启并提交事务：消费者只做 flush 与状态更新，自己不提交， 否则会出现"返回已处理但表里状态没变"。单个片段失败不冒泡（在消费者内部 转换为 `失败` + 重试计数），因此本方法只在数据库层面失败时抛错。 """ async with SessionFactory() as session, session.begin(): return await EpisodeExtractionConsumer( session, extractor=self.memory_extraction, cache=self.memory_cache, limit=limit, ).consume_pending() async def execute(self, run_id: str) -> bool: # A new fencing token for each claim also fences restarts of the same process. worker_id = str(uuid4()) async with SessionFactory() as session, session.begin(): run = await session.scalar(select(AgentRun).where( AgentRun.run_id == run_id).with_for_update()) if run is None: return False if run.status == "cancel_requested": # 文档 §6.4：运行真正落到 cancelled 时，原请求在 request_idempotency 中 # 以 failed + RUN_CANCELLED 结束（HTTP 层取消受理时可能已写过一次， # 这里幂等覆盖，保证 worker 抢先落终态的场景也不漏）。 run.status = "cancelled" run.completed_at = datetime.now(UTC).replace(tzinfo=None) run.locked_until = None await session.execute(update(RequestIdempotency).where( RequestIdempotency.id == run.idempotency_id ).values(status="failed", error_code="RUN_CANCELLED", updated_at=run.completed_at)) return True claimed = await AgentRunRepository(session).claim( run_id, worker_id, self.settings.worker_lease_seconds) if not claimed: return False task = asyncio.create_task(self._execute_claimed(run_id, worker_id)) heartbeat = asyncio.create_task(self._heartbeat(run_id, worker_id, task)) try: await task except asyncio.CancelledError: await self._failure(run_id, worker_id, "RUN_INTERRUPTED", retryable=True) # Runtime cancellation is shutdown; a lost lease only cancels the child. current = asyncio.current_task() if current is not None and current.cancelling(): raise except RunLeaseLostError: await self._failure(run_id, worker_id, "RUN_LEASE_LOST", retryable=True) except Exception as exc: code = exc.code if isinstance(exc, AgentError) else "AGENT_INTERNAL_ERROR" await self._failure(run_id, worker_id, code, retryable=isinstance(exc, RecoverableAgentError)) # 记录堆栈：只记录异常类型会让线上排障无从下手（run 的 error_code 只有 # AGENT_INTERNAL_ERROR，看不到真实原因）。 logger.warning("run failed run_id=%s error_type=%s", run_id, type(exc).__name__, exc_info=True) finally: heartbeat.cancel() with contextlib.suppress(asyncio.CancelledError): await heartbeat return True async def _heartbeat( self, run_id: str, worker_id: str, task: asyncio.Task[None] ) -> None: try: while True: await asyncio.sleep(self.settings.worker_lease_seconds / 3) async with SessionFactory() as session, session.begin(): renewed = await AgentRunRepository(session).renew( run_id, worker_id, self.settings.worker_lease_seconds) if not renewed: task.cancel() return except Exception: task.cancel() logger.warning("lease renewal failed run_id=%s", run_id) async def _execute_claimed(self, run_id: str, worker_id: str) -> None: async with SessionFactory() as session: run = await AgentRunRepository(session).get(run_id) if run is None: raise ValueError("run not found") message = await session.get(ConversationMessage, run.request_message_id) idem = await session.get(RequestIdempotency, run.idempotency_id) event = await session.scalar(select(DomainEventOutbox).where( DomainEventOutbox.aggregate_id == run_id, DomainEventOutbox.event_type == "agent.run_requested").limit(1)) if message is None or idem is None: raise ValueError("run input missing") metadata = event.payload.get("metadata", {}) if event else {} request = AgentRequest( agent_type=run.agent_type, message=message.content, session_id=run.session_id, idempotency_key=idem.idempotency_key, metadata=AgentRequestMetadata.model_validate(metadata), ) actor_type = ( str(event.payload.get("actor_type", "authenticated")) if event else "authenticated" ) actor_id = str(run.user_id) trace_id = run.trace_id context = await self.restore_context( actor_type=actor_type, actor_id=actor_id, trace_id=trace_id ) result: AgentResult | None = None async for event_data in AgentExecutor(self.factory).execute( request.agent_type, request, context, run_id ): if event_data.event_type == "done": result = AgentResult.model_validate(event_data.payload["result"]) if result is None: raise ValueError("Agent produced no terminal result") # 业务事件查询必须用独立 Session：在 complete_run 的 Session 上先跑 SELECT 会触发 # SQLAlchemy 的 autobegin，使 complete_run 内部的 session.begin() 抛 # "A transaction is already begun on this Session"，运行直接失败。 async with SessionFactory() as events_session: business_events = list(await events_session.scalars( select(DomainEventOutbox.event_type).where( DomainEventOutbox.aggregate_id == run_id, DomainEventOutbox.event_type.in_(tuple(BUSINESS_EVENT_TYPES)), ) )) async with SessionFactory() as session: await AgentPersistenceService(session).complete_run( run_id, result, worker_id=worker_id, memory_extraction_requested=self.should_request_memory_extraction( agent_type=run.agent_type, context=context, message=request.message, result=result, business_events=business_events, ), profile_candidate_requested=self.should_request_profile_candidate( agent_type=run.agent_type, context=context, message=request.message, ), ) # 只有数据库成功保存完整用户/助手轮次后才写短期 Redis；Redis 故障不应让已完成 # 的客服回答回滚或重试。短期组件会再次脱敏，形成持久化链路的第二道保护。 await self._append_customer_service_session_memory( agent_type=run.agent_type, actor_id=str(run.user_id), session_id=run.session_id, request_message=request.message, response_message=result.result.text, ) async def _append_customer_service_session_memory( self, *, agent_type: str, actor_id: str, session_id: str, request_message: str, response_message: str, ) -> None: """将一轮成功客服会话写入 Redis；此派生写入永不影响已提交的主事务。""" if agent_type != "customer_service": return try: await self.session_memory.append( actor_id=actor_id, session_id=session_id, turns=( CustomerServiceSessionTurn(role="user", content=request_message), CustomerServiceSessionTurn(role="assistant", content=response_message), ), ) except Exception: # 注入的第三方 Redis 适配器也可能违反“不抛异常”约定，Worker 仍必须隔离故障。 logger.warning("客服短期会话写入降级，不影响已完成的客服运行", exc_info=True) async def _failure( self, run_id: str, worker_id: str, error_code: str, *, retryable: bool ) -> None: now = datetime.now(UTC).replace(tzinfo=None) async with SessionFactory() as session, session.begin(): run = await session.scalar(select(AgentRun).where( AgentRun.run_id == run_id).with_for_update()) if run is None or run.worker_id != worker_id: return if run.status not in {"running", "cancel_requested"}: return if run.status == "cancel_requested": run.status = "cancelled" elif retryable and run.attempt_count < self.settings.worker_retry_limit: run.status = "queued" else: run.status = "failed" run.error_code, run.updated_at = error_code, now run.locked_until = ( now + timedelta(seconds=min(60, 2**run.attempt_count)) if run.status == "queued" else None ) if run.status != "queued": run.completed_at = now # 取消导致的终态属于“原请求已终止”，按文档 §6.4 在 request_idempotency # 中写 failed + RUN_CANCELLED；`agent_run.error_code` 仍保留真实触发原因 #（如 RUN_INTERRUPTED），两者层级不同。 await session.execute(update(RequestIdempotency).where( RequestIdempotency.id == run.idempotency_id ).values( status="failed", error_code="RUN_CANCELLED" if run.status == "cancelled" else error_code, updated_at=now, )) session.add(InteractionAudit( actor_type="agent", actor_id=run.user_id, session_id=run.session_id, portal="api", action_type=f"agent.run_{run.status}", detail={"run_id": run_id, "error_code": error_code}, created_at=now, ))