group_fqcd_jr/app/worker/runtime.py

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_memory_embedding_service,
    get_milvus_knowledge_writer,
    get_milvus_profile_vector_client,
    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.knowledge_vector_worker import build_knowledge_handlers
from app.worker.memory_extraction_worker import MemoryExtractionWorker
from app.worker.outbox_worker import OutboxHandlerError, 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"

#: 组装层默认值的哨兵。知识写路径的三个依赖都要区分"没传（取生产装配）"与"显式
#: `None`（显式降级）"：`None` 若同时表示两者，测试里就无法在不改环境变量的前提下
#: 构造"Milvus 未配置"的场景，降级行为也就无法被固定。
_UNSET: Any = object()


@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,
        relationships: Any = None,
        knowledge_writer: Any = _UNSET,
        knowledge_embedder: Any = _UNSET,
        knowledge_endpoint_resolver: Any = _UNSET,
        profile_vector_client: Any = _UNSET,
        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 删除客户端：由**组装层**（app/worker/__main__.py）注入生产实现，
        # 这里不兜底。组件内部给默认实现会把"尚未装配"这一事实悄悄盖住——而"未注入即显式
        # 降级并留痕"是本模块刻意保留的语义（有单测守着），因此默认值保持 None。
        self.projection_cleaner = projection_cleaner
        # 客服短期会话 Redis 仅用于当前会话上下文，不参与长期画像召回。
        self.session_memory = (
            session_memory if session_memory is not None
            else build_customer_service_session_memory()
        )
        # 图关系服务：画像投影用它写入节点与关系（投顾的多跳推荐、风控的关系网络都读它）。
        # 默认取生产装配；图库不可用时该值为 None，投影如实降级而不是失败。
        if relationships is not None:
            self.relationships = relationships
        else:
            # 延迟导入：bootstrap 会间接导入本模块，模块级导入会形成循环依赖
            from app.service.agent.bootstrap import get_relationship_service

            self.relationships = get_relationship_service()
        # 知识向量同步：Milvus 写适配器 + 嵌入服务 + 端点解析器全部由组装层注入
        # （`app/service/agent/bootstrap.py`）。写适配器构造是**惰性**的（不连 Milvus），
        # 所以这里取默认值不会让 worker 起不来；真连不上时在写入时抛
        # `RecoverableAgentError`，交给 OutboxWorker 的退避重试与死信机制。
        self.knowledge_writer = (
            get_milvus_knowledge_writer() if knowledge_writer is _UNSET else knowledge_writer
        )
        self.knowledge_embedder = (
            get_memory_embedding_service() if knowledge_embedder is _UNSET
            else knowledge_embedder
        )
        self.knowledge_endpoint_resolver = (
            DatabaseModelEndpointResolver() if knowledge_endpoint_resolver is _UNSET
            else knowledge_endpoint_resolver
        )
        # 降级告警只打一次：dispatch 是轮询热路径，每轮一条 warning 会把日志淹掉。
        self._knowledge_degraded_logged = False
        # 画像投影（`memory_sync_outbox`）消费装配。
        #
        # 与 `knowledge_writer` 同一取向：客户端构造**惰性**（不连 Milvus），
        # `milvus_uri` 未配置时显式降级为不注册 handler（事件留 pending、可观测、可重放），
        # 绝不伪造同步成功。
        self.profile_vector_client = (
            get_milvus_profile_vector_client()
            if profile_vector_client is _UNSET
            else profile_vector_client
        )
        self.profile_endpoint_resolver = self.knowledge_endpoint_resolver
        self._profile_degraded_logged = False
        # 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",),
                "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:
        """长期记忆抽取只接收非客服、非访客的明确业务事实。"""
        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:
        """客服仅为已登录且 self 范围内的用户生成待确认画像候选。"""
        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 OutboxHandlerError("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 OutboxHandlerError("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 OutboxHandlerError("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 OutboxHandlerError("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_profile_rebuild(payload: dict[str, Any]) -> None:
                """画像重建 + 图投影，由记忆写入后发出的事件驱动。

                为什么绕一层事件而不在记忆抽取处直接调用：抽取时那条记忆还在**未提交**的
                事务里，另开 session 去重建画像看不到它（实测：快照加了、事实没进、图里
                也没多出关系）。事件只可能在本事务提交之后被消费，届时数据一定可见。
                """
                customer_id = payload.get("customer_id")
                if not customer_id:
                    raise OutboxHandlerError("profile.rebuild_requested payload is incomplete")
                # 延迟导入：bootstrap 会间接导入本模块，模块级导入会形成循环依赖
                from app.service.profile_assembly_service import ProfileAssemblyService
                from app.service.profile_graph_projection_service import (
                    ProfileGraphProjectionService,
                )

                outcome = await ProfileAssemblyService(session).rebuild(int(customer_id))
                if outcome.get("profile") is None:
                    # 客户尚未开户：画像行由开户流程创建（trade_account 等字段属注册侧所有），
                    # 这里不代建。事实已提升进 user_facts，开户后再重建即可。
                    logger.info("profile rebuild skipped (not opened) customer_id=%s", customer_id)
                    return
                projection = ProfileGraphProjectionService(session, self.relationships)
                result = await projection.project_customer(int(customer_id))
                if result.degraded:
                    logger.warning("graph projection degraded customer_id=%s reason=%s",
                                   customer_id, result.reason)

            async def dispatch_handover_queue_ready(payload: dict[str, Any]) -> None:
                """记录转人工队列已就绪；不向客户承诺已接单或处理时限。"""
                ticket_no = str(payload.get("ticket_no", "")).strip()
                if not ticket_no:
                    raise OutboxHandlerError(
                        "conversation.transfer_requested payload is incomplete"
                    )
                ticket = await session.scalar(
                    select(HandoverTicket).where(HandoverTicket.ticket_no == ticket_no)
                )
                if ticket is None:
                    raise OutboxHandlerError("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()

            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 OutboxHandlerError("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", "")),
                )

            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,
                # 画像重建：记忆写入后自动触发，使「记忆 → 画像 → 图」全链路无需手工介入
                "profile.rebuild_requested": dispatch_profile_rebuild,
                "conversation.transfer_requested": dispatch_handover_queue_ready,
            }
            # 知识向量同步/删除：Task 5 交付了 handler 与写适配器，但先前没有任何生产装配
            # 调用它们 —— 事件类型不在上面的白名单里，`OutboxWorker.publish_one` 的
            # `event_type.in_(tuple(self.handlers))` 就永远领不到这些行，现库 408 条
            # `knowledge.vector_sync_requested` 因此永久 pending、Milvus 零向量、
            # 检索永远返回空。这里把两个 handler 合并进同一个字典（**同一个 `session`**，
            # handler 不得 commit，事务仍归 OutboxWorker）。
            if self.knowledge_writer is not None:
                handlers.update(build_knowledge_handlers(
                    session,
                    writer=self.knowledge_writer,
                    embedder=self.knowledge_embedder,
                    endpoint_resolver=self.knowledge_endpoint_resolver,
                ))
            elif not self._knowledge_degraded_logged:
                # 显式降级 + 留痕：不注册 handler，知识事件保持 pending（库里可查、可重放），
                # 绝不伪造"已同步"。与 `projection_cleaner` 的降级口径一致。
                self._knowledge_degraded_logged = True
                logger.warning(
                    "knowledge vector handlers not registered: milvus writer unavailable "
                    "(check settings.milvus_uri); knowledge.vector_sync_requested / "
                    "knowledge.vector_delete_requested stay pending"
                )
            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()

    @staticmethod
    async def _conversation_history(
        session: Any, *, session_id: str, user_id: int, before_message_id: int | None,
        limit: int = 10,
    ) -> tuple[Any, ...]:
        """取该会话最近若干轮对话，按时间正序（旧 → 新）返回。

        以 MySQL 的会话消息为**唯一来源**，不引入 Redis 双写：消息在受理时已经落库，
        再同步一份到 Redis 只会带来不一致与 TTL 管理成本，换来的仅是一次索引查询的节省。
        方案 §2.2 设想的是 Redis 列表，这里取等价语义（同样"最近若干轮、超出即截断"）
        而不复制存储。

        `before_message_id` 排除本轮请求消息本身：它刚写入库，若也算进历史，
        模型会在上下文里看到自己的问题被重复一遍。

        截断按**条数**而非 token：这里没有与模型一致的分词器，按 token 截断只能靠估算、
        边界会随实现漂移；按条数是确定性的，宁可少给几轮，也不给一个不稳定的边界。
        """
        from app.core.contracts import ConversationTurn
        from app.repository.conversation_repository import ConversationRepository

        rows = await ConversationRepository(session).messages(
            session_id, user_id, limit + 1, before=before_message_id
        )
        # repository 按 id DESC 返回（最新在前），这里翻正为旧 → 新
        ordered = list(reversed(rows))[-limit:]
        turns: list[Any] = []
        for row in ordered:
            content = str(row.content or "").strip()
            if not content:
                continue
            turns.append(ConversationTurn(
                role="assistant" if str(row.role) == "assistant" else "user",
                content=content,
            ))
        return tuple(turns)

    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
        # 画像投影消费：与领域事件同一轮次内处理。失败只告警，不影响 run 的处理与
        # 轮询节奏——事件仍在库里，下一轮照常重试（退避由 worker 自己记在 next_retry_at）。
        try:
            projected = await self.consume_profile_projections() > 0
        except Exception:
            logger.warning("profile projection consumption failed", exc_info=True)
            projected = False
        dispatched = dispatched or projected
        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 consume_profile_projections(self, *, limit: int = 20) -> int:
        """消费 `memory_sync_outbox` 的画像投影事件，返回本次处理条数。

        两个目标存储的分工（**方案 A**：以架构师主干为主线，不引入第二套 Neo4j 投影）：

        - `milvus` → 写入长期记忆向量集合 `user_long_term_memory_v1`
          （此前**完全没有消费者**，事件永久滞留）；
        - `neo4j` → 复用主干 `ProfileGraphProjectionService`。主干已由
          `profile.rebuild_requested` 事件驱动同一条链，图投影是 `MERGE` 幂等的，
          因此这里再投一次不产生重复节点/关系，只用于把 outbox 行的投递状态收敛掉。

        为什么不让 handler 自己 commit：事务边界与 `dispatch_batch` 一致，
        由本方法按条提交；单条失败由 `MemorySyncOutboxWorker` 内部转成
        `failed`+退避或死信，不冒泡打断本轮其余事件。
        """
        if self.profile_vector_client is None:
            # 显式降级：不注册 handler 就交给 worker 判死信是**错的**（那是把配置缺失
            # 伪装成投递失败）。这里直接不消费，事件保持 pending，由启动日志提示。
            if not self._profile_degraded_logged:
                self._profile_degraded_logged = True
                logger.warning(
                    "profile projection disabled: milvus profile vector client unavailable; "
                    "memory_sync_outbox events stay pending"
                )
            return 0
        # 收窄到局部变量：闭包内访问 self 属性时 mypy 无法保留上面的 None 判定。
        vector_client = self.profile_vector_client

        from app.infrastructure.milvus_profile_projection import MilvusProfileProjection
        from app.worker.memory_sync_outbox_worker import MemorySyncOutboxWorker

        async def project_milvus(payload: dict[str, Any]) -> None:
            effective = await self._with_memory_sources(payload)
            projection = MilvusProfileProjection(vector_client, self._profile_embed)
            await projection.upsert(effective)

        async def project_neo4j(payload: dict[str, Any]) -> None:
            raw_customer_id = payload.get("customer_id")
            # 显式 isinstance 而不是 `in (None, "")`：后者不做类型收窄，mypy 无法确认
            # int() 的入参类型；同时也把"客户号必须是数字"这一契约写在类型检查里。
            if not isinstance(raw_customer_id, (int, str)) or raw_customer_id == "":
                raise RecoverableAgentError("profile projection payload has no customer_id")
            customer_id = int(raw_customer_id)
            # 延迟导入：与 dispatch_profile_rebuild 同一理由，避免模块级循环依赖。
            from app.service.profile_graph_projection_service import (
                ProfileGraphProjectionService,
            )

            async with SessionFactory() as session:
                outcome = await ProfileGraphProjectionService(
                    session, self.relationships
                ).project_customer(customer_id)
            if outcome.degraded:
                # 图库不可用：如实抛出，让 worker 走失败/退避，而不是记成已投递。
                raise RecoverableAgentError(f"graph projection degraded: {outcome.reason}")

        worker = MemorySyncOutboxWorker(
            {"milvus": project_milvus, "neo4j": project_neo4j}
        )
        handled = 0
        for _ in range(max(1, limit)):
            if not await worker.run_once():
                break
            handled += 1
        return handled

    async def _with_memory_sources(self, payload: dict[str, Any]) -> dict[str, Any]:
        """保证 payload 带 `memory_sources`；缺失时回退为查询当前有效记忆。

        为什么需要这个兜底：`memory_sources` 是本仓新增的投影入参，而**投顾线两处
        生产者**（`profile_governance_service` / `risk_questionnaire_service`）发的 payload
        是 `{customer_id, profile_uuid, version, profile}`，**没有**这个键。若不兜底，
        它们每次画像变更都会因 `memory_sources is invalid` 失败重试直至死信
        （本仓 `memory_sync_outbox` 已有这种 `ValueError` 行留痕）。

        为什么不是"缺失就报错"：缺失与"格式错"性质不同——缺失表示该生产者不知道要提供，
        属契约演进期的正常情况；格式错（不是列表、字段不合法）仍由适配器**失败关闭**，
        不会被这里掩盖。

        回退查的是 `memory_unit` 中 `status='active'` 的行，即"该客户当前有效的长期记忆"。
        这在语义上成立：长期记忆是**客户级**的，不是画像版本级的；且每条记忆自带
        `version`，适配器按 `memory_uuid + version` 做幂等，所以"用的是哪一版"仍然确定。

        每次兜底都记一条 warning，使"谁没提供 memory_sources"保持可见，而不是静默兼容。
        """
        sources = payload.get("memory_sources")
        # 只对"**键不存在或为 None**"兜底。若键存在但格式不对（例如字符串），
        # 原样放行交给适配器报错——那是真错误，兜底会把它悄悄修好、线上永远看不见。
        # （用 isinstance 判断会把这两种情况混为一谈，故用键存在性判断。）
        if sources is not None:
            return payload

        raw_customer_id = payload.get("customer_id")
        if not isinstance(raw_customer_id, (int, str)) or raw_customer_id == "":
            # 没有客户号就无法兜底；交给适配器按原 payload 失败关闭。
            return payload
        customer_id = int(raw_customer_id)

        from app.repository.profile_repository import ProfileRepository

        async with SessionFactory() as session:
            rows = await ProfileRepository(session).active_memories(customer_id)
        logger.warning(
            "profile projection payload has no memory_sources (customer_id=%s); "
            "fell back to %s active memories from memory_unit",
            customer_id,
            len(rows),
        )
        return {
            **payload,
            "memory_sources": [
                {
                    "memory_uuid": str(row["memory_uuid"]),
                    "memory_key": str(row["memory_key"]),
                    "content": str(row["content"]),
                    "memory_type": str(row["memory_type"]),
                    "confidence": float(row["confidence"]),
                    "version": int(row["version"]),
                    "valid_until": (
                        row["valid_until"].isoformat() if row["valid_until"] else None
                    ),
                }
                for row in rows
            ],
        }

    async def _profile_embed(self, text: str) -> list[float]:
        """向量化一条记忆正文；端点走与知识向量化同一套已批准端点解析。

        不复用 `bootstrap._embed_text`：那是模块私有函数，跨模块引用私有名会把
        两处的耦合藏起来。这里用同一组公开装配（端点解析器 + embedding 服务）。
        """
        endpoints = await self.profile_endpoint_resolver.resolve(
            agent_type="memory_recall", task_type="embedding"
        )
        if not endpoints:
            raise RecoverableAgentError("没有可用的 embedding 端点，无法投影长期记忆")
        execution = await self.knowledge_embedder.embed(endpoints, text)
        return list(execution.vector)

    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")
            # 短期会话记忆：加载本次之前的对话，模型靠它解析指代。
            history = await self._conversation_history(
                session, session_id=run.session_id, user_id=int(run.user_id),
                before_message_id=run.request_message_id,
            )
            request = AgentRequest(
                agent_type=run.agent_type, message=message.content, session_id=run.session_id,
                idempotency_key=idem.idempotency_key,
                metadata=AgentRequestMetadata.model_validate(
                    event.payload.get("metadata", {}) if event else {}
                ),
                history=history,
            )
            actor_type = (
                str(event.payload.get("actor_type", "authenticated"))
                if event else "authenticated"
            )
        # Re-check account and permissions at execution time, including delayed jobs.
        context = await self.restore_context(
            actor_type=actor_type, actor_id=str(run.user_id), trace_id=run.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,
                ),
            )
        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:
            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,
                ))