group_fqcd_jr/app/infrastructure/milvus_profile_projection.py

"""Milvus 长期记忆投影适配器。

只写入已经审核的 `memory_sources`，不接受画像快照整体冒充单条记忆。

来源：同事 `ZSY_develop` 分支（`app/infrastructure/milvus_profile_projection.py`），
本文件在其基础上做了**两处契约放宽**，均为"避免整批失败"，不改变写入语义：

1. **`customer_id` 接受 int 或数字字符串**。原实现要求 `isinstance(customer_id, int)`，
   而本仓**所有** outbox 生产者写的都是 `str(customer_id)`
   （`profile_generation_service`、投顾线 `profile_governance_service` 与
   `risk_questionnaire_service` 三处皆然）。不放宽则每个事件必然失败、重试 5 次后进死信。
2. **不可投影的 `memory_key` 跳过而非整批报错**。受控词表
   （`app/service/memory_taxonomy.py`）共 13 个键，其中 `constraint:*`（3 个）与
   `profile:*`（4 个）不以 `preference:`/`goal:` 开头。原实现对第一个不合规的键直接
   `raise`，一条 `constraint:` 记忆就会毒死该客户整批同步；现改为跳过并留痕，
   使"能投影的照常写入"。
"""

import logging
from collections.abc import Awaitable, Callable
from datetime import UTC, datetime
from typing import Any, Protocol
from uuid import UUID

from app.core.conversation_privacy import sanitize_customer_service_message
from app.core.errors import RecoverableAgentError

logger = logging.getLogger(__name__)

#: 长期记忆向量集合名的**唯一真相**（写、读、删三侧共用，见下方说明），不是"某处默认值"。
#:
#: 2026-09-14 故障复盘：本常量与召回/清理两侧读的 `settings.milvus_collection`
#: （`jr_memory`，**该集合从未被创建**）不一致，于是
#:   ① 语义召回每次 search 都抛异常 → 降级 `milvus_unavailable` → 召回恒空；
#:   ② 记忆删除走 `vector_collection_absent` → 报清理成功但一个向量都没删。
#: 两侧各自单测全绿，接缝没人守。现约定：**这里改，就三侧一起改**
#: （`app/service/agent/bootstrap.py`、`app/service/projection_cleanup_service.py`），
#: 并由 `tests/unit/infrastructure/test_memory_vector_collection_consistency.py` 守住。
PROFILE_COLLECTION = "user_long_term_memory_v1"
VECTOR_DIM = 1024

#: 可投影的键前缀。其余受控键（`constraint:*` / `profile:*`）属结构化字段，
#: 与向量召回不是同一用途，因此不进长期记忆向量集合。
PROJECTABLE_KEY_PREFIXES: tuple[str, ...] = ("preference:", "goal:")


class MilvusProfileClient(Protocol):
    async def query(self, **kwargs: Any) -> list[dict[str, Any]]: ...

    async def upsert(self, **kwargs: Any) -> Any: ...


EmbeddingProvider = Callable[[str], Awaitable[list[float]]]


class MilvusProfileProjection:
    """按记忆 UUID 幂等写入长期记忆向量。"""

    def __init__(
        self,
        client: MilvusProfileClient,
        embed: EmbeddingProvider,
        *,
        collection: str = PROFILE_COLLECTION,
    ) -> None:
        self._client = client
        self._embed = embed
        self._collection = collection

    async def upsert(self, payload: dict[str, Any]) -> None:
        customer_id, profile_version, sources = self._normalize(payload)
        if not sources:
            # ⚠️ 零可投影记忆时**直接返回**，不要先去 `load_collection`。
            #
            # `load_collection` 在集合不存在或未加载时会抛错，客户端把它包装成
            # `RecoverableAgentError`（`milvus_profile_vector_client.py:40-47`
            # 「画像向量集合不可用」）。而 `upsert` 此前**无条件**在 embed 循环之前调它，
            # 于是"本来就没有可写内容"的事件也被记成投递失败、重试 5 次进死信。
            #
            # 实测正是这样：10001 / 10002 开户风险测评投的两条画像事件，
            # `memory_unit` 里没有它们的记忆（`_with_memory_sources` 兜底查出 0 条），
            # 在 load_collection 那一步就死了 —— 而 `user_long_term_memory_v1`
            # 当时根本不存在。**"没东西可写"不该等同于"写失败"**。
            logger.info(
                "profile projection skipped: no projectable memory customer_id=%s",
                customer_id,
            )
            return
        load_collection = getattr(self._client, "load_collection", None)
        if load_collection is not None:
            await load_collection(collection_name=self._collection)
        rows: list[dict[str, Any]] = []
        for source in sources:
            vector = await self._embed(source["content"])
            if len(vector) != VECTOR_DIM:
                raise RecoverableAgentError("画像向量维度不一致")
            existing = await self._client.query(
                collection_name=self._collection,
                filter=f'memory_uuid == "{source["memory_uuid"]}"',
                output_fields=["memory_uuid", "version", "customer_id"],
            )
            # 幂等：库里已有更新版本时不回退覆盖。
            if existing and int(existing[0].get("version", 0)) > source["version"]:
                continue
            rows.append({
                "memory_uuid": source["memory_uuid"],
                "customer_id": customer_id,
                "content": source["content"],
                "embedding": vector,
                "memory_type": source["memory_type"],
                "memory_key": source["memory_key"],
                "confidence": source["confidence"],
                "version": source["version"],
                "status": "active",
                "valid_until_ts": source["valid_until_ts"],
                "updated_at_ts": source["updated_at_ts"],
            })
        if rows:
            await self._client.upsert(collection_name=self._collection, data=rows)
        # 全部被跳过时也要留一条记录：否则"零写入"与"没跑"在日志上无法区分。
        logger.info(
            "milvus profile projection: customer_id=%s profile_version=%s written=%s",
            customer_id,
            profile_version,
            len(rows),
        )

    @staticmethod
    def _coerce_customer_id(raw: Any) -> int:
        """接受 int 或数字字符串。

        生产端统一写 `str(customer_id)`（本仓三处生产者皆然），若严格要求 int，
        所有事件都会失败。仅接受**纯数字**字符串，非数字一律拒绝，
        避免把 uuid 之类当成客户号写进向量库。
        """
        if isinstance(raw, bool):
            raise ValueError("customer_id is invalid")
        if isinstance(raw, int):
            value = raw
        elif isinstance(raw, str) and raw.strip().isdigit():
            value = int(raw.strip())
        else:
            raise ValueError("customer_id is invalid")
        if value <= 0:
            raise ValueError("customer_id is invalid")
        return value

    @staticmethod
    def _coerce_profile_version(payload: dict[str, Any]) -> int:
        """兼容两种键名：`profile_version`（本适配器契约）与 `version`（本仓生产端）。"""
        raw = payload.get("profile_version", payload.get("version"))
        if isinstance(raw, bool) or not isinstance(raw, int) or raw <= 0:
            raise ValueError("profile_version is invalid")
        return raw

    @classmethod
    def _normalize(
        cls,
        payload: dict[str, Any],
    ) -> tuple[int, int, list[dict[str, Any]]]:
        customer_id = cls._coerce_customer_id(payload.get("customer_id"))
        profile_version = cls._coerce_profile_version(payload)
        sources = payload.get("memory_sources")
        if not isinstance(sources, list):
            raise ValueError("memory_sources is invalid")
        normalized: list[dict[str, Any]] = []
        skipped: list[str] = []
        now = int(datetime.now(UTC).timestamp())
        for source in sources:
            if not isinstance(source, dict):
                raise ValueError("memory source is invalid")
            required = [
                source.get(name)
                for name in ("memory_uuid", "memory_key", "content", "memory_type")
            ]
            if not all(isinstance(value, str) and value.strip() for value in required):
                raise ValueError("memory source fields are invalid")
            try:
                memory_uuid = str(UUID(str(source["memory_uuid"])))
            except ValueError as exc:
                raise ValueError("memory_uuid is invalid") from exc
            memory_key = str(source["memory_key"]).strip()
            if not memory_key.startswith(PROJECTABLE_KEY_PREFIXES):
                # 跳过而非整批失败：`constraint:*` / `profile:*` 是结构化事实，
                # 不进向量召回。一条不可投影的键不得毒死同一客户其余记忆。
                skipped.append(memory_key)
                continue
            confidence = source.get("confidence")
            version = source.get("version")
            if not isinstance(confidence, (int, float)) or not 0 <= confidence <= 1:
                raise ValueError("memory confidence is invalid")
            if not isinstance(version, int) or version <= 0:
                raise ValueError("memory version is invalid")
            valid_until = source.get("valid_until")
            valid_until_ts = None
            if isinstance(valid_until, str) and valid_until:
                try:
                    valid_until_ts = int(datetime.fromisoformat(valid_until).timestamp())
                except ValueError as exc:
                    raise ValueError("memory valid_until is invalid") from exc
            normalized.append({
                "memory_uuid": memory_uuid,
                "memory_key": memory_key,
                "content": sanitize_customer_service_message(str(source["content"])).strip(),
                "memory_type": str(source["memory_type"]).strip(),
                "confidence": float(confidence),
                "version": version,
                "valid_until_ts": valid_until_ts,
                "updated_at_ts": now,
            })
        if skipped:
            logger.info(
                "milvus profile projection: skipped %s non-projectable memory keys %s",
                len(skipped),
                sorted(set(skipped)),
            )
        return customer_id, profile_version, normalized