fix(memory-projection): 订正 outbox 取值口径并接通画像投影链路

背景：memory_sync_outbox 这条链此前**完全没有消费者**，且生产端照 docs/00 §6.4.6
写成大写 MILVUS/NEO4J + 中文「待处理」，而消费端按 target_store 的**值**分派 handler、
且只领 status in {pending, failed} —— 两个条件都不满足，事件任何消费者都领不到、
永久滞留且不报错（唯一键 (event_uuid, target_store) 对大小写无约束，MySQL 也不报错）。
根因是代码与测试都硬编码字面量，所以测试跟着一起错、谁也没拦住。

订正
- profile_generation_service：取值改为全仓一致的小写（milvus/neo4j/upsert/pending）
- 测试改为引用常量并断言消费端契约，不再硬编码（硬编码是本次跑偏的直接原因）
- 新增契约回归测试：断言大写值分派不到 handler、会进死信，谁改回大写立刻红
- 新增 tools/normalize_memory_sync_outbox.py：订正历史脏行（默认 dry-run、幂等）

接通投影链路（此前零消费者）
- 新增 Milvus 集合 user_long_term_memory_v1 及建集合工具（幂等、不覆盖已有集合）
- 新增 MilvusProfileProjection / MilvusProfileVectorClient，并修掉移植带来的两处必炸点：
  customer_id 由「必须 int」放宽为接受数字字符串（本仓所有生产者都写 str，
  不放宽则每个事件必然失败）；不可投影的 memory_key 由「整批 raise」改为跳过留痕
  （否则一条 constraint: 记忆毒死该客户整批，而受控词表 13 个键里有 7 个不满足前缀）
- 新增 MemorySyncOutboxWorker（领取/指数退避/死信骨架保留原样）并接入 WorkerRuntime
- milvus → 向量投影；neo4j → 复用主干 ProfileGraphProjectionService（方案 A，
  不引入第二套投影，避免同一事实在图中两种说法、违反主干既有的只投影已确认事实的不变式）
- 生产端从 memory_unit(status=active) 组装 memory_sources，随事件带上确定快照
- 前置移植 conversation_privacy：写外部存储前脱敏手机号/证件号/银行卡等

验证
- 新增 17 个单测；全量 2 failed, 1307 passed, 2 skipped
  （2 个失败为既有环境项：断言请求体中文原文而 httpx 序列化成 \uXXXX，非本次引入）
- mypy app → 0 错（227 文件）；audit_schema → 89 张业务表无缺失/意外，未改动表结构
- 真机：真实 embedding(1024 维) + 真实 Milvus 写入并回读通过
- 整合链路（测试记忆 → 生产端组装 → outbox → 消费端投递 → Milvus 回读）通过，
  且 MySQL 已回滚、Milvus 无残留

文档
- 新增 docs/32-记忆投影链路实现说明.md：真实口径、根因、契约与验证证据（供接手）
- AGENTS.md：新增该易错点；新增 Windows 中文输出乱码的正确命令（-X utf8）；
  校正测试基线与 mypy 文件数

未做：未改 docs/00 基线、未动数据库迁移、未改投顾线代码、未启动常驻 Worker。
遗留：投顾线两处生产者的 payload 缺 memory_sources，会被消费至死信，待架构师确认是否投影。

AGENTS.md

@@ -72,6 +72,24 @@
 
 - 解释器：本机用 **`.\.venv\Scripts\python.exe`**；架构师环境用 `D:\conda\envs\jr_py313\python.exe`。
   两者等价，**各用本机可用的那个**（`.venv` 被 `.gitignore` 忽略、不进仓库，不存在"需要统一"的问题）。
+- ⚠️ **Windows 控制台跑测试请加 `-X utf8`，否则中文输出是乱码**：
+
+  ```powershell
+  .\.venv\Scripts\python.exe -X utf8 -m pytest tests -q
+  ```
+
+  原因：本机控制台是代码页 **936（GBK）**，Python 的 `sys.stdout.encoding` 随之为 `gbk`，
+  而终端按 UTF-8 解码 ⇒ 中文（断言消息、`docs/` 中文文件名、测试内 `print`）全部乱码。
+  **乱码只影响"显示"，不影响测试结果**（`passed`/`failed` 计数是英文，照常可信）。
+
+  为什么必须用 `-X utf8` 而不是别的办法：
+  - `chcp 65001` **单独无效**（Python 仍以 GBK 输出，终端按 UTF-8 解码，反而更乱）；
+  - 在 `tests/conftest.py` 里 `sys.stdout.reconfigure(encoding="utf-8")` **只能修测试内部的 `print`**，
+    **修不了 pytest 自己的输出**（失败摘要、短测试摘要）—— pytest 的终端写入器在启动时就绑定了编码，
+    早于 conftest 被加载。**实测两组对照：摘要照样乱码。**
+  - `-X utf8` 在解释器启动时生效，早于 pytest 的一切，且只影响当次进程、无副作用。
+  - 等价替代：环境变量 `PYTHONUTF8=1` 或 `PYTHONIOENCODING=utf-8`（PyCharm 用户可写进 Run Configuration）。
+  同一原因，**跑 `tools/*.py` 等脚本时也要加 `-X utf8`**（例如 `python -X utf8 tools\check_authoritative_docs.py`）。
 - 数据库现为 **69 张表**（含 `alembic_version`）＝ **68 张业务表** = **场内 51 + 场外/推广 17**。
   后 17 张（`offsite_*` / `promotion_*`）**不进 `docs/00` 基线**（规则 8：场外基金运营流程独立），
   逐表登记见 `docs/28-场外与推广域数据表登记.md`。核验命令：`python tools/audit_schema.py`。
@@ -86,13 +104,22 @@
   （`app/core/knowledge_schema.py`）——**不要在任何地方硬编码字段名**，那会把另一套环境打挂。
 - ⚠️ **Docker Desktop 不会常驻**：它没运行时 Milvus 不可用（`docker` CLI 报连不上守护进程）。
   跑真机验证前先确认 Docker Desktop 在运行。
-- 测试基线：`3 failed, 1219 passed, 2 skipped`（2026-09-11 合并主干后实测）。
-  三个失败**都不是代码缺陷**，接手时不要"修"它们：
-  ① `tests/unit/repository/test_fund_readonly_contract.py`（**底座既有缺陷，不要修也不要报**）；
-  ② ③ `tests/unit/service/test_offsite_document_recognition_adapter.py` 的 2 个用例 —— **环境相关**：
+- ⚠️ **`memory_sync_outbox` 的取值必须是小写英文**（`milvus`/`neo4j`、`upsert`、
+  `pending`/`failed`/`processed`/`dead`）。`docs/00` §6.4.6 那一栏曾写作大写
+  `MILVUS`/`NEO4J`、`UPSERT` + 中文 `待处理`，**与全仓实现从未对齐，照它写会静默失效**：
+  消费端按 `handlers.get(target_store)` 分派、且只领 `status in {"pending","failed"}`，
+  大写 + 中文两个条件都不满足 ⇒ **事件任何消费者都领不到、永久滞留且不报错**
+  （唯一键 `(event_uuid, target_store)` 对大小写无约束，MySQL 也不报错）。
+  取值口径以**主干既有读取方**为准（`projection_reconciliation_service.py`、
+  `graph_projection_worker.py`），不是文档。详见 `docs/32-记忆投影链路实现说明.md`。
+- 测试基线：`2 failed, 1307 passed, 2 skipped`（2026-09-12 记忆投影链路落地后实测；
+  此前为 `3 failed, 1219 passed`——第 3 个失败
+  `tests/unit/repository/test_fund_readonly_contract.py` 已由投顾线合入主干时修复，**不要再当既有缺陷引用**）。
+  剩下 2 个失败**都不是代码缺陷**，接手时不要"修"它们：
+  `tests/unit/service/test_offsite_document_recognition_adapter.py` 的 2 个用例 —— **环境相关**：
   它们断言请求体里是中文原文，而 httpx 会把中文序列化成 `\uXXXX`，字节序列自然不匹配。
   功能无影响；若要修，正确做法是断言 `json.loads(body)` 后的字段值（字节级断言不该用来测 JSON）。
-- **mypy：`mypy app` → `Success: no issues found in 180 source files`（0 错）。**
+- **mypy：`mypy app` → `Success: no issues found in 227 source files`（0 错）。**
   ⚠️ 曾在本机报 184 个错，**已查明是环境版本旧**，与代码质量无关 —— 复现矩阵：
 
   | SQLAlchemy | mypy | 报错数 |

app/core/conversation_privacy.py

@@ -0,0 +1,32 @@
+"""客服会话落库前的敏感凭据最小化处理。
+
+来源：同事 `ZSY_develop` 分支（`app/core/conversation_privacy.py`），整文件移植，未改语义。
+纯正则、无外部依赖，供记忆投影（`MilvusProfileProjection`）在写入向量前做最后一道脱敏。
+
+为什么要单独抽一层：记忆内容最终会进入 Milvus 与 Neo4j，一旦写入就脱离了会话事务的
+管控范围。把脱敏放在**写入适配器内部**（而不是依赖调用方记得做），是为了让"未经脱敏的
+文本不得落外部存储"成为代码保证，而不是流程约定。
+"""
+
+import re
+
+# 替换顺序从带业务语义的凭据开始，避免通用数字规则先破坏上下文。
+_SENSITIVE_PATTERNS: tuple[tuple[re.Pattern[str], str], ...] = (
+    (
+        re.compile(r"(?i)((?:登录|交易)?密码)\s*(?:[:：=]|是)\s*[^\s，。；,;]{1,64}"),
+        r"\1[已隐藏]",
+    ),
+    (re.compile(r"(?i)((?:登录|交易)?密码)\s*\d{4,32}"), r"\1[已隐藏]"),
+    (re.compile(r"(?i)(验证码|短信码|校验码)\s*(?:[:：=]|是)?\s*\d{4,8}"), r"\1[已隐藏]"),
+    (re.compile(r"(?<!\d)\d{17}[\dXx](?!\d)"), "[证件号已隐藏]"),
+    (re.compile(r"(?<!\d)(?:\d[ -]?){15,18}\d(?!\d)"), "[银行卡号已隐藏]"),
+    (re.compile(r"(?<!\d)1[3-9]\d{9}(?!\d)"), "[手机号已隐藏]"),
+)
+
+
+def sanitize_customer_service_message(message: str) -> str:
+    """保留风险关键词，移除不应进入会话、Outbox 或后续 Redis 的凭据值。"""
+    sanitized = message
+    for pattern, replacement in _SENSITIVE_PATTERNS:
+        sanitized = pattern.sub(replacement, sanitized)
+    return sanitized

app/infrastructure/milvus_profile_projection.py

@@ -0,0 +1,191 @@
+"""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__)
+
+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)
+        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

app/infrastructure/milvus_profile_vector_client.py

@@ -0,0 +1,65 @@
+"""长期记忆画像向量的 Milvus 客户端。
+
+与 `MilvusKnowledgeWriter` 分开：那个适配器的 `upsert` 签名绑定知识集合的
+`knowledge_id` 主键与字段表，而画像投影的主键是 `memory_uuid`、且需要先 `query`
+按版本判重。两者共用一套连接口径（惰性连接 + 失败一律 `RecoverableAgentError`），
+但不硬把两种 schema 塞进同一个类。
+
+与召回侧的客户端也分开：写路径不与检索进程共用连接（读写物理隔离，向量库故障
+不能从写路径传染到问答主链路），与 `get_milvus_knowledge_writer` 的取向一致。
+"""
+
+from typing import Any
+
+from app.core.errors import RecoverableAgentError
+
+
+class MilvusProfileVectorClient:
+    """满足 `MilvusProfileProjection` 所需的 `query` / `upsert` / `load_collection`。"""
+
+    def __init__(self, uri: str, token: str = "") -> None:
+        self._uri = uri
+        self._token = token
+        self._client: Any = None
+
+    async def _ensure(self) -> Any:
+        if self._client is None:
+            try:
+                from pymilvus import AsyncMilvusClient  # type: ignore[import-untyped]
+            except ImportError as exc:  # pragma: no cover - 依赖已声明，缺装是环境问题
+                raise RecoverableAgentError("pymilvus 未安装，无法写入画像向量") from exc
+            try:
+                self._client = AsyncMilvusClient(uri=self._uri, token=self._token or None)
+            except Exception as exc:
+                raise RecoverableAgentError("Milvus 画像写客户端初始化失败") from exc
+        return self._client
+
+    async def load_collection(self, *, collection_name: str) -> None:
+        """把集合载入内存后再查/写。
+
+        集合不存在时抛 `RecoverableAgentError`：由 outbox 退避重试并最终判死信，
+        而不是静默跳过——"集合没建"是装配问题，必须可见。
+        """
+        client = await self._ensure()
+        try:
+            await client.load_collection(collection_name=collection_name)
+        except Exception as exc:
+            raise RecoverableAgentError("画像向量集合不可用") from exc
+
+    async def query(self, **kwargs: Any) -> list[dict[str, Any]]:
+        """按 filter 查询；返回空列表表示无匹配（不是错误）。"""
+        client = await self._ensure()
+        try:
+            raw = await client.query(**kwargs)
+        except Exception as exc:
+            raise RecoverableAgentError("画像向量查询失败") from exc
+        return list(raw or [])
+
+    async def upsert(self, **kwargs: Any) -> Any:
+        client = await self._ensure()
+        try:
+            return await client.upsert(**kwargs)
+        except RecoverableAgentError:
+            raise
+        except Exception as exc:
+            raise RecoverableAgentError("画像向量写入失败") from exc

app/repository/profile_repository.py

@@ -50,6 +50,17 @@ _SQL_NEXT_VERSION = text("""
     FROM profile_snapshots WHERE customer_id = :customer_id
 """)
 
+#: 该客户**有效**的长期记忆（投影到 Milvus 长期记忆向量集合的数据源）。
+#:
+#: 只取 `status='active'`：失效/被取代的记忆不应再进入向量召回，否则投顾会召回
+#: 已过期偏好。字段与 `MilvusProfileProjection` 的 `memory_sources` 契约一一对应。
+_SQL_ACTIVE_MEMORIES = text("""
+    SELECT memory_uuid, memory_key, content, memory_type, confidence, version, valid_until
+    FROM memory_unit
+    WHERE customer_id = :customer_id AND status = 'active'
+    ORDER BY id
+""")
+
 #: 旧画像置非当前（`current_customer_id` 有唯一键，必须先置 0 才能插新的当前版本）。
 _SQL_CLEAR_CURRENT = text("""
     UPDATE profile_snapshots
@@ -88,6 +99,13 @@ class ProfileRepository:
         row = result.mappings().first()
         return dict(row) if row is not None else None
 
+    async def active_memories(self, customer_id: int) -> list[dict[str, Any]]:
+        """该客户所有 `status='active'` 的长期记忆，按 id 稳定排序。"""
+        result = await self._session.execute(
+            _SQL_ACTIVE_MEMORIES, {"customer_id": customer_id}
+        )
+        return [dict(row) for row in result.mappings().all()]
+
     async def next_version(self, customer_id: int) -> int:
         return int(await self._session.scalar(_SQL_NEXT_VERSION, {"customer_id": customer_id}))
 

app/service/agent/bootstrap.py

@@ -19,6 +19,7 @@ from app.infrastructure.fund_quote_cache import FundQuoteCache
 from app.infrastructure.graph import build_graph_driver
 from app.infrastructure.memory_cache import MemoryCacheAdapter
 from app.infrastructure.milvus_knowledge_writer import MilvusKnowledgeWriter
+from app.infrastructure.milvus_profile_vector_client import MilvusProfileVectorClient
 from app.infrastructure.vector_memory import VectorMemoryAdapter
 from app.service.agent.factory import AgentFactory
 from app.service.agent.governance import PlatformGovernance
@@ -159,6 +160,26 @@ def get_milvus_knowledge_writer() -> MilvusKnowledgeWriter | None:
     return MilvusKnowledgeWriter(uri, settings.milvus_token or "")
 
 
+@lru_cache(maxsize=1)
+def get_milvus_profile_vector_client() -> MilvusProfileVectorClient | None:
+    """长期记忆画像向量**写**适配器（装配入口）；`milvus_uri` 缺失时返回 None。
+
+    与 `get_milvus_knowledge_writer` 同一取向：读写隔离、构造惰性、缺配置显式降级。
+    返回 None 的语义是**显式降级**——`WorkerRuntime` 会因此不注册画像投影 handler，
+    `memory_sync_outbox` 的事件在库里保持 pending（可观测、可重放），并留一条 warning，
+    绝不静默，也绝不伪造同步成功。
+    """
+    settings = get_settings()
+    uri = (settings.milvus_uri or "").strip()
+    if not uri:
+        logging.getLogger(__name__).warning(
+            "milvus_uri not configured; profile vector projection disabled and "
+            "memory_sync_outbox events will stay pending"
+        )
+        return None
+    return MilvusProfileVectorClient(uri, settings.milvus_token or "")
+
+
 async def _embed_text(text: str) -> list[float]:
     """把文本向量化；端点来自发布配置（task_type=embedding），无端点时失败关闭。"""
     endpoints = await DatabaseModelEndpointResolver().resolve(

app/service/profile_generation_service.py

@@ -44,14 +44,24 @@ from app.core.errors import ValidationAgentError
 from app.core.profile_projection import PROFILE_FIELD_POLICY
 from app.repository.profile_repository import ProfileRepository
 
-#: `memory_sync_outbox` 的两个目标存储（`docs/00` §6.4.6 的 `target_store` 取值）。
-TARGET_MILVUS = "MILVUS"
-TARGET_NEO4J = "NEO4J"
+#: `memory_sync_outbox` 的两个目标存储。
+#:
+#: ⚠️ **一律小写**：消费端按 `target_store` 的值分派 handler
+#: （`MemorySyncOutboxWorker.handlers.get(event.target_store)`），且
+#: `graph_projection_worker` / `projection_reconciliation_service` 的领取与重放
+#: 都以 `status in {"pending","failed"}` + `"processed"` 为准。全仓（包含投顾线两处生产者）
+#: 统一使用小写 `milvus`/`neo4j`、`upsert`、`pending`。
+#:
+#: 历史说明：`docs/00` §6.4.6 该栏曾写作大写 `MILVUS`/`NEO4J`、`UPSERT` 与中文 `待处理`，
+#: 与上述实现从未对齐；本模块原先照文档写，是**全仓唯一的异类**，导致自己写的事件
+#: 任何消费者都领不到。现统一为小写，并已把历史 2 行就地改齐（主键/唯一键不变）。
+TARGET_MILVUS = "milvus"
+TARGET_NEO4J = "neo4j"
 SYNC_TARGETS: tuple[str, ...] = (TARGET_MILVUS, TARGET_NEO4J)
 
-#: 同步操作与状态取值（与 `memory_sync_outbox` DDL 的语义一致）。
-SYNC_OPERATION_UPSERT = "UPSERT"
-SYNC_STATUS_PENDING = "待处理"
+#: 同步操作与状态取值。
+SYNC_OPERATION_UPSERT = "upsert"
+SYNC_STATUS_PENDING = "pending"
 AGGREGATE_TYPE_PROFILE = "profile"
 
 
@@ -168,8 +178,12 @@ class ProfileGenerationService:
             "aggregate_uuid": profile_uuid,
             "customer_id": str(customer_id),
             "version": version,
+            # 别名：投影适配器契约用 `profile_version`。两个键都写，避免消费端
+            # 因生产者用词不同而取不到值（本仓三种 payload 形状的历史遗留）。
+            "profile_version": version,
             "snapshot_hash": snapshot_hash,
             "snapshot": snapshot,
+            "memory_sources": await self._memory_sources(customer_id),
         }
         for target in SYNC_TARGETS:
             self._repo.add_sync_event(
@@ -192,6 +206,34 @@ class ProfileGenerationService:
             sync_events=len(SYNC_TARGETS),
         )
 
+    async def _memory_sources(self, customer_id: int) -> list[dict[str, Any]]:
+        """组装投影适配器要的 `memory_sources`（数据源：`memory_unit` 中 `status='active'`）。
+
+        为什么由生产端组装而不是消费端回查：消费端到时那条记忆可能已改版本，
+        让它自己去查会把"投递的是哪一版"变成不确定。事件里带上当时的确定快照，
+        投递语义才与 `aggregate_version` 一致（幂等判据也才有意义）。
+
+        为什么是 `memory_unit` 而不是 `user_facts`：两者用途不同——`user_facts` 是
+        **已确认的结构化事实**，喂画像与图投影；`memory_unit` 是**长期记忆条目**，
+        正是长期记忆向量集合要存的东西。这不与"图投影只读 user_facts"的不变式冲突，
+        因为 Milvus 存的就是记忆本身的向量，不是画像事实的另一种说法。
+        """
+        rows = await self._repo.active_memories(customer_id)
+        return [
+            {
+                "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
+        ]
+
     @staticmethod
     def _generation_basis(assessment_row: Mapping[str, Any] | None) -> dict[str, object]:
         """记录这次画像**依据了什么**（`docs/00`：使用的测评版本、交易窗口和记忆版本列表）。

app/worker/memory_sync_outbox_worker.py

@@ -0,0 +1,100 @@
+"""画像投影 Outbox 消费器。
+
+外部存储通过 handler 注入；本模块只负责领取、重试、死信和 MySQL 状态更新。
+
+来源：同事 `ZSY_develop` 分支（`app/worker/memory_sync_outbox_worker.py`）。
+本文件保留其**领取/重试/退避/死信骨架原样未改**——这部分（`skip_locked` 并发领取、
+指数退避、5 次转死信）是正确的，也是这条链上唯一把"投递可靠性"做对的地方。
+
+分派口径：`runtime.py` 是本消费器的装配方，它会把 handler 键注册为**全仓统一的小写**
+`milvus` / `neo4j`（与生产端写入的 `target_store` 取值一致）。
+"""
+
+from collections.abc import Awaitable, Callable
+from datetime import UTC, datetime, timedelta
+from typing import Any
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.infrastructure.db import SessionFactory
+from app.model.memory import MemorySyncOutbox
+
+ProjectionHandler = Callable[[dict[str, Any]], Awaitable[Any]]
+MAX_RETRY_COUNT = 5
+
+
+class MemorySyncOutboxWorker:
+    """按目标存储独立消费画像投影事件。"""
+
+    def __init__(
+        self,
+        handlers: dict[str, ProjectionHandler],
+        *,
+        session_factory: Callable[[], AsyncSession] = SessionFactory,
+    ) -> None:
+        self.handlers = handlers
+        self.session_factory = session_factory
+
+    async def run_once(
+        self, *, target_store: str | None = None, event_uuid: str | None = None
+    ) -> bool:
+        """领取并处理一条到期事件；没有可处理事件时返回 False。"""
+        if not self.handlers:
+            return False
+        async with self.session_factory() as session:
+            now = datetime.now(UTC).replace(tzinfo=None)
+            conditions: list[Any] = [
+                MemorySyncOutbox.status.in_({"pending", "failed"}),
+                MemorySyncOutbox.next_retry_at.is_(None)
+                | (MemorySyncOutbox.next_retry_at <= now),
+            ]
+            if target_store is not None:
+                conditions.append(MemorySyncOutbox.target_store == target_store)
+            if event_uuid is not None:
+                conditions.append(MemorySyncOutbox.event_uuid == event_uuid)
+            event = await session.scalar(
+                select(MemorySyncOutbox)
+                .where(*conditions)
+                .order_by(MemorySyncOutbox.id)
+                .limit(1)
+                .with_for_update(skip_locked=True)
+            )
+            if event is None:
+                await session.rollback()
+                return False
+            handler = self.handlers.get(event.target_store)
+            if handler is None:
+                # 没有对应 handler 时判死信（不留 pending 空转），并写明原因。
+                self._fail(event, "target_handler_not_configured", now, dead=True)
+                await session.commit()
+                return True
+            try:
+                await handler(event.payload)
+            except Exception as exc:
+                self._fail(event, type(exc).__name__, now)
+            else:
+                event.status = "processed"
+                event.processed_at = datetime.now(UTC).replace(tzinfo=None)
+                event.last_error = None
+                event.next_retry_at = None
+            await session.commit()
+            return True
+
+    @staticmethod
+    def _fail(
+        event: MemorySyncOutbox,
+        reason: str,
+        now: datetime,
+        *,
+        dead: bool = False,
+    ) -> None:
+        """写入可重试失败或死信状态，不吞掉失败事实。"""
+        event.retry_count = int(event.retry_count) + 1
+        event.last_error = reason[:500]
+        event.status = "dead" if dead or event.retry_count >= MAX_RETRY_COUNT else "failed"
+        event.next_retry_at = (
+            None
+            if event.status == "dead"
+            else now + timedelta(seconds=min(300, 2 ** int(event.retry_count)))
+        )

app/worker/runtime.py

@@ -22,6 +22,7 @@ from app.service.agent.bootstrap import (
     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
@@ -91,6 +92,7 @@ class WorkerRuntime:
         knowledge_writer: Any = _UNSET,
         knowledge_embedder: Any = _UNSET,
         knowledge_endpoint_resolver: Any = _UNSET,
+        profile_vector_client: Any = _UNSET,
     ) -> None:
         self.factory = factory if factory is not None else get_agent_factory()
         self.settings = settings or get_settings()
@@ -138,6 +140,18 @@ class WorkerRuntime:
         )
         # 降级告警只打一次：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
 
@@ -408,6 +422,14 @@ class WorkerRuntime:
 
     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 唯一键），失败只告警，
@@ -431,6 +453,85 @@ class WorkerRuntime:
             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:
+            projection = MilvusProfileProjection(vector_client, self._profile_embed)
+            await projection.upsert(payload)
+
+        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 _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，返回新写入的片段数。
 

docs/32-记忆投影链路实现说明.md

@@ -0,0 +1,229 @@
+# 记忆投影链路（Outbox → Milvus 长期记忆 / Neo4j）实现说明
+
+**适用分支**：`NL_develop`（用户端线）｜**日期**：2026-09-12｜**状态**：已实现并通过真机验证
+
+---
+
+## 1. 这条链路是干什么的
+
+记忆（对话里被抽取出来的长期事实）要能被后续召回，必须从 MySQL 同步到两处外部存储：
+
+```
+对话消息
+  └─→ memory.extraction_requested（domain_event_outbox）
+        └─→ MemoryExtractionWorker → MemoryService.upsert() → memory_unit（MySQL，唯一真相）
+              └─→ profile.rebuild_requested（domain_event_outbox）
+                    ├─→ ProfileAssemblyService.rebuild()        → profile_snapshots
+                    ├─→ ProfileGraphProjectionService          → Neo4j（图）
+                    └─→ ProfileGenerationService.generate()    → memory_sync_outbox
+                          ├─ milvus → MilvusProfileProjection   → Milvus 长期记忆向量
+                          └─ neo4j  → ProfileGraphProjectionService（幂等复投）
+```
+
+`memory_sync_outbox` 是"画像版本 → 外部存储"的投递队列，唯一键
+`uk_memory_sync_event (event_uuid, target_store)`：**同一 `event_uuid` 对两个目标库各写一条**。
+
+---
+
+## 2. ⚠️ 最容易踩的坑：枚举取值必须全仓统一（小写 + 英文）
+
+这条链曾**完全失效但不报错**，根因就是取值口径不统一。
+
+### 唯一正确口径
+
+| 字段 | 取值 | 大小写 |
+|---|---|---|
+| `target_store` | `milvus` / `neo4j` | **小写** |
+| `operation` | `upsert` / `archive` / `delete` | **小写** |
+| `status` | `pending` / `failed` / `processed` / `dead` | **小写英文** |
+
+（`aggregate_type` 用 `memory` / `profile` / `relationship` / `deletion`。）
+
+### 为什么照 `docs/00` §6.4.6 写会坏
+
+`docs/00` §6.4.6 那一栏曾写作大写 `MILVUS`/`NEO4J`、`UPSERT` 与中文 `待处理`，
+与**全仓实现从未对齐**。按那份文档写会造成：
+
+1. `MemorySyncOutboxWorker` 按 `handlers.get(event.target_store)` 分派 handler
+   —— 大写值找不到 handler；
+2. 领取条件是 `status.in_({"pending","failed"})` —— 中文 `待处理` 不满足。
+
+⇒ **两个条件都不满足，事件任何消费者都领不到，永久滞留且不报错。**
+唯一键 `(event_uuid, target_store)` 对大小写没有约束，MySQL 也不会报错，所以是**静默失效**。
+
+判断依据应以**主干既有读取方**为准，不是文档：
+`projection_reconciliation_service.py:20`（`{"pending","failed"}`）、
+`graph_projection_worker.py`（`pending`/`processed`/`dead`）。
+
+### 现状
+
+- 生产端取值集中在 `app/service/profile_generation_service.py` 的常量
+  （`TARGET_MILVUS`/`TARGET_NEO4J`/`SYNC_OPERATION_UPSERT`/`SYNC_STATUS_PENDING`）；
+- **测试不再硬编码字面量**，并断言 `SYNC_STATUS_PENDING == "pending"`、
+  `set(SYNC_TARGETS) == {"milvus","neo4j"}`
+  （硬编码正是当初跑偏的直接原因）；
+- `tests/unit/worker/test_memory_sync_outbox_worker.py` 有一条契约回归测试，
+  断言大写 `MILVUS` 分派不到 handler、会进死信 —— 谁改回大写，测试立刻红；
+- 历史 2 行已就地把取值改齐（只改值，主键/唯一键/payload 未动），
+  正式订正工具 `tools/normalize_memory_sync_outbox.py`
+  （**默认 dry-run，加 `--apply` 才写库**，幂等可复跑；换环境若也有同批旧值可直接用）。
+
+---
+
+## 3. Milvus 长期记忆向量集合
+
+| 项 | 值 |
+|---|---|
+| 集合名 | `user_long_term_memory_v1` |
+| 主键 | `memory_uuid`（VARCHAR 64，UUID 字符串，**按 UUID 幂等 upsert**） |
+| 向量 | `embedding`，`FLOAT_VECTOR` dim **1024**，索引 `AUTOINDEX` + `COSINE` |
+| 其他字段 | `customer_id`(INT64) / `version`(INT64,可空) / `valid_until_ts`(INT64,可空) / `updated_at_ts`(INT64,可空) / `confidence`(DOUBLE) / `content`(VARCHAR 2048) / `memory_type`(VARCHAR 32) / `memory_key`(VARCHAR 64) / `status`(VARCHAR 16) |
+
+建集合：`python tools/setup_milvus_profile_collection.py`
+——**幂等**，集合已存在时只做结构比对报告、不覆盖不删重建（共享 Milvus 实例里还有别的项目的集合）。
+
+### `memory_sources` 契约（适配器的输入）
+
+```json
+{
+  "customer_id": 9102,
+  "profile_version": 2,
+  "memory_sources": [{
+    "memory_uuid": "uuid 字符串",
+    "memory_key": "preference:risk_level",
+    "content": "正文",
+    "memory_type": "preference",
+    "confidence": 0.9,
+    "version": 1,
+    "valid_until": null
+  }]
+}
+```
+
+数据源：`memory_unit` 中 `status='active'` 的行
+（`ProfileRepository.active_memories()`）。**由生产端组装**而不是消费端回查：
+消费端到时记忆可能已改版本，事件里带确定快照才能与 `aggregate_version` 语义一致。
+
+> 为什么用 `memory_unit` 而不是 `user_facts`：两者用途不同 —— `user_facts` 是
+> **已确认的结构化事实**（喂画像与图投影），`memory_unit` 是**长期记忆条目**
+> （正是长期记忆向量集合要存的东西）。这不与"图投影只读 `user_facts`"的不变式冲突。
+
+---
+
+## 4. 移植自 `ZSY_develop` 的改动（含两处契约放宽）
+
+来源：同事 `ZSY_develop` 分支。整文件移植、骨架未改的部分：
+`app/core/conversation_privacy.py`、`app/worker/memory_sync_outbox_worker.py`
+（领取/`skip_locked`/指数退避/5 次转死信的可靠性骨架原样保留）、
+`app/infrastructure/milvus_profile_projection.py`。
+
+对适配器做了**两处契约放宽**，都是"避免整批失败"，不改变写入语义：
+
+1. **`customer_id` 接受 int 或数字字符串**
+   原实现要求 `isinstance(customer_id, int)`，而本仓**所有**生产者写的都是
+   `str(customer_id)`。不放宽则**每个事件必然失败**、重试 5 次后进死信。
+   放宽仅限**纯数字**串，非数字（如 uuid）仍拒绝。
+2. **不可投影的 `memory_key` 跳过而非整批报错**
+   受控词表 `app/service/memory_taxonomy.py` 共 13 个键，其中 `constraint:*`（3 个）与
+   `profile:*`（4 个）不以 `preference:`/`goal:` 开头。原实现对第一个不合规的键直接
+   `raise` —— 一条 `constraint:` 记忆就会毒死该客户整批同步。现改为**跳过并留痕**。
+
+**收窄而非扩容的理由**：`preference:*`/`goal:*` 是偏好与目标，适合语义召回；
+`constraint:*`/`profile:*` 是结构化约束与属性，应由结构化通道查询（`user_facts` → 图），
+放进向量集合只会造成召回噪声。若日后要给它们做语义召回，应扩容而非靠现在的跳过。
+
+---
+
+## 5. Neo4j 分支：方案 A（不引入第二套投影）
+
+同事分支另有一套 `neo4j_profile_projection.py`（按客户各建私有
+`Preference`/`Goal` 节点、数据源 `memory_unit`）。**未采用**，理由：
+
+主干 `ProfileGraphProjectionService` 已由 `profile.rebuild_requested` 驱动同一条链，
+其数据源是 `user_facts`（**已确认事实**）、`MERGE` **共享 tag 节点**，且文件头写明不变式：
+
+> "只投影'已确认'的事实……保证图里的偏好标签与画像口径一致；
+> 否则同一件事在画像和图里会有两种说法。"
+
+两套并存 = 同一事实在图中两种表示，正好违反这条不变式。
+因此 `memory_sync_outbox` 的 `neo4j` 分支**复用主干服务**：
+图投影是 `MERGE` 幂等的，再投一次不产生重复节点/关系，只用于收敛 outbox 的投递状态。
+
+---
+
+## 6. 消费端装配
+
+`WorkerRuntime.consume_profile_projections()`（`app/worker/runtime.py`），在
+`run_once()` 每轮执行，按条提交、单条失败不冒泡。
+
+- `milvus` handler → `MilvusProfileProjection`（注入 `MilvusProfileVectorClient` + 向量化）；
+- `neo4j` handler → `ProfileGraphProjectionService`；其 `degraded` 会**如实抛错**
+  走失败/退避，不记成已投递；
+- `milvus_uri` 未配置时**显式降级**：不消费、事件留 `pending`（可观测、可重放），
+  启动路径留一条 warning，绝不伪造同步成功（与 `knowledge_writer` 同一取向）；
+- 目标存储没有对应 handler 时判**死信**并记 `target_handler_not_configured`
+  —— 这正是当初大写值事件的下场（有回归测试守着）。
+
+向量化复用与知识向量同一套已批准端点解析（`agent_type="memory_recall"`,
+`task_type="embedding"`）。
+
+---
+
+## 7. 真机验证证据（2026-09-12）
+
+| 验证项 | 结果 |
+|---|---|
+| 集合创建幂等 | 首次 `created`，复跑 `exists` |
+| 真实 embedding 维度 | **1024**（与集合定义一致） |
+| 真实写入 + 回读 | 2 条可投影键写入成功并可回读（内容/版本正确） |
+| 不可投影键 | `constraint:liquidity` **未写入**（跳过生效，未毒死整批） |
+| 测试数据清理 | 已按 `customer_id=999999` 删除，集合残留 **0** 条 |
+| 单元测试 | 新增 17 个（适配器 10 + worker 5 + 生产端 2），全过 |
+| 全量回归 | `2 failed, 1307 passed, 2 skipped` —— 与基线一致，**无新增失败** |
+| mypy | `Success: no issues found in 227 source files` |
+
+> 全量的 2 个失败是既有环境相关项（`test_offsite_document_recognition_adapter.py`
+> 断言请求体里的中文原文，而 httpx 序列化成 `\uXXXX`），与本次改动无关。
+
+> 真机注意：Milvus 写入后**短时间内可能查不到**（索引尚未可见），
+> 验证脚本按重试处理；同理删除后立即查询可能仍返回旧行，需重查确认。
+
+---
+
+## 8. 尚未完成 / 依赖他人
+
+1. **常驻 Worker 未运行**：核对时数据库里 `memory_sync_outbox` 2 行、
+   `domain_event_outbox` 积压
+   （`agent.run_requested` 428、`profile.rebuild_requested` 216、
+   `agent.run_completed` 61、`memory.extraction_requested` 58）。
+   **`memory_unit` 与 `user_facts` 目前都是 0 行** —— 代码链路是通的，
+   但没有 Worker 在跑，所以记忆永远不会被抽取出来。起
+   `python -m app.worker --once`（或常驻）即会开始消费。
+2. **投顾线两处生产者的取值**：`profile_governance_service.py` 与
+   `risk_questionnaire_service.py` 已写小写 `milvus`/`neo4j` + `pending`（与本文口径一致），
+   但它们的 `payload` 是 `{customer_id, profile_uuid, version, profile}`，
+   **没有 `memory_sources`** → 投影时会因 `memory_sources is invalid` 失败重试至死信。
+   需投顾线补 `memory_sources` 或明确这两个来源是否也要投影长期记忆。
+   （属架构师线的改动，本线未动。）
+3. `docs/00` §6.4.6 的取值栏与实现不一致：按评审要求**未改基线文档**，
+   实际口径以本文第 2 节为准。
+
+---
+
+## 9. 相关文件
+
+**新增**
+- `app/core/conversation_privacy.py`（移植）
+- `app/infrastructure/milvus_profile_projection.py`（移植 + 2 处放宽）
+- `app/infrastructure/milvus_profile_vector_client.py`
+- `app/worker/memory_sync_outbox_worker.py`（移植）
+- `tools/setup_milvus_profile_collection.py`
+- `tests/unit/infrastructure/test_milvus_profile_projection.py`
+- `tests/unit/worker/test_memory_sync_outbox_worker.py`
+
+**修改**
+- `app/service/profile_generation_service.py`（取值改小写、payload 加 `memory_sources` 与 `profile_version`）
+- `app/repository/profile_repository.py`（`active_memories()`）
+- `app/service/agent/bootstrap.py`（`get_milvus_profile_vector_client()`）
+- `app/worker/runtime.py`（`consume_profile_projections()` + `run_once` 接线）
+- `tests/unit/service/test_profile_generation_service.py`（+2 用例、断言改引用常量）

tests/unit/infrastructure/test_milvus_profile_projection.py

@@ -0,0 +1,201 @@
+"""`MilvusProfileProjection` 的定向测试。
+
+前 4 个用例移植自同事 `ZSY_develop` 的
+`tests/unit/infrastructure/test_milvus_profile_projection.py`；
+后 4 个覆盖本仓对其做的**两处契约放宽**（`customer_id` 兼容字符串、
+不可投影键跳过而非整批失败）与脱敏，这些是移植时必须钉住的差异点。
+"""
+
+from uuid import uuid4
+
+import pytest
+
+from app.core.errors import RecoverableAgentError
+from app.infrastructure.milvus_profile_projection import MilvusProfileProjection
+
+
+class FakeMilvus:
+    def __init__(self, existing: list[dict[str, object]] | None = None) -> None:
+        self.existing = existing or []
+        self.queries: list[dict[str, object]] = []
+        self.upserts: list[dict[str, object]] = []
+
+    async def query(self, **kwargs: object) -> list[dict[str, object]]:
+        self.queries.append(kwargs)
+        return self.existing
+
+    async def upsert(self, **kwargs: object) -> None:
+        self.upserts.append(kwargs)
+
+
+def payload() -> dict[str, object]:
+    return {
+        "customer_id": 7,
+        "profile_version": 1,
+        "memory_sources": [{
+            "memory_uuid": str(uuid4()),
+            "memory_key": "preference:risk_level",
+            "content": "稳健型",
+            "memory_type": "preference",
+            "confidence": 0.9,
+            "version": 2,
+            "valid_until": None,
+        }],
+    }
+
+
+def _source(data: dict[str, object]) -> dict[str, object]:
+    sources = data["memory_sources"]
+    assert isinstance(sources, list)
+    source = sources[0]
+    assert isinstance(source, dict)
+    return source
+
+
+@pytest.mark.asyncio
+async def test_upsert_writes_schema_fields_and_vector() -> None:
+    client = FakeMilvus()
+    projection = MilvusProfileProjection(client, _embed)
+
+    await projection.upsert(payload())
+
+    assert len(client.upserts) == 1
+    row = client.upserts[0]["data"][0]
+    assert row["customer_id"] == 7
+    assert row["status"] == "active"
+    assert len(row["embedding"]) == 1024
+
+
+@pytest.mark.asyncio
+async def test_lower_memory_version_is_not_overwritten() -> None:
+    data = payload()
+    memory_uuid = _source(data)["memory_uuid"]
+    client = FakeMilvus(existing=[{
+        "memory_uuid": memory_uuid, "customer_id": 7, "version": 3,
+    }])
+
+    await MilvusProfileProjection(client, _embed).upsert(data)
+
+    assert client.upserts == []
+
+
+@pytest.mark.asyncio
+async def test_embedding_dimension_is_enforced() -> None:
+    with pytest.raises(RecoverableAgentError, match="维度"):
+        await MilvusProfileProjection(client=FakeMilvus(), embed=_embed_short).upsert(
+            payload()
+        )
+
+
+@pytest.mark.asyncio
+async def test_non_uuid_memory_id_is_rejected() -> None:
+    data = payload()
+    _source(data)["memory_uuid"] = "unsafe\" or true"
+
+    with pytest.raises(ValueError, match="memory_uuid"):
+        await MilvusProfileProjection(FakeMilvus(), _embed).upsert(data)
+
+
+# --- 本仓放宽的契约（移植差异点） -------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_string_customer_id_is_accepted() -> None:
+    """本仓三处生产者写的都是 `str(customer_id)`；不接受字符串则事件必然全部失败。"""
+    data = payload()
+    data["customer_id"] = "9102"
+
+    client = FakeMilvus()
+    await MilvusProfileProjection(client, _embed).upsert(data)
+
+    assert client.upserts[0]["data"][0]["customer_id"] == 9102
+
+
+@pytest.mark.asyncio
+async def test_non_numeric_customer_id_is_rejected() -> None:
+    """放宽不等于不校验：uuid 之类的非数字串必须拒绝，不能当成客户号写进向量库。"""
+    data = payload()
+    data["customer_id"] = "957c0552-7fa2-4f2a-924d-d2d2e133b245"
+
+    with pytest.raises(ValueError, match="customer_id"):
+        await MilvusProfileProjection(FakeMilvus(), _embed).upsert(data)
+
+
+@pytest.mark.asyncio
+async def test_version_key_fallback_is_supported() -> None:
+    """本仓生产端 payload 用 `version`；适配器契约用 `profile_version`。两个都要认。"""
+    data = payload()
+    del data["profile_version"]
+    data["version"] = 5
+
+    client = FakeMilvus()
+    await MilvusProfileProjection(client, _embed).upsert(data)
+
+    assert len(client.upserts) == 1
+
+
+@pytest.mark.asyncio
+async def test_non_projectable_memory_key_is_skipped_not_fatal() -> None:
+    """`constraint:*` / `profile:*` 不在可投影前缀内。
+
+    关键：一条不可投影的键**不得**毒死同一客户其余可投影记忆。
+    """
+    data = payload()
+    good = _source(data)
+    data["memory_sources"] = [
+        {
+            "memory_uuid": str(uuid4()),
+            "memory_key": "constraint:liquidity",
+            "content": "半年内需要流动性",
+            "memory_type": "constraint",
+            "confidence": 0.8,
+            "version": 1,
+            "valid_until": None,
+        },
+        good,
+    ]
+
+    client = FakeMilvus()
+    await MilvusProfileProjection(client, _embed).upsert(data)
+
+    assert len(client.upserts) == 1
+    rows = client.upserts[0]["data"]
+    assert [row["memory_key"] for row in rows] == ["preference:risk_level"]
+
+
+@pytest.mark.asyncio
+async def test_all_keys_non_projectable_writes_nothing() -> None:
+    """全部不可投影时不写 Milvus，但也不报错（不是失败，是无需投影）。"""
+    data = payload()
+    _source(data)["memory_key"] = "profile:occupation"
+
+    client = FakeMilvus()
+    await MilvusProfileProjection(client, _embed).upsert(data)
+
+    assert client.upserts == []
+
+
+@pytest.mark.asyncio
+async def test_sensitive_credentials_are_sanitized_before_write() -> None:
+    """落外部存储前必须脱敏：手机号不得原样写进向量库。"""
+    data = payload()
+    _source(data)["content"] = "我的手机号是 15936583816，稳健型"
+
+    client = FakeMilvus()
+    await MilvusProfileProjection(client, _embed).upsert(data)
+
+    content = client.upserts[0]["data"][0]["content"]
+    assert "15936583816" not in content
+    assert "[手机号已隐藏]" in content
+
+
+def _vector(size: int = 1024) -> list[float]:
+    return [0.0] * size
+
+
+async def _embed(_: str) -> list[float]:
+    return _vector()
+
+
+async def _embed_short(_: str) -> list[float]:
+    return _vector(3)

tests/unit/service/test_profile_generation_service.py

@@ -21,6 +21,8 @@ from app.core.errors import ValidationAgentError
 from app.core.profile_projection import project_profile
 from app.service.profile_generation_service import (
     REQUIRED_SNAPSHOT_FIELDS,
+    SYNC_OPERATION_UPSERT,
+    SYNC_STATUS_PENDING,
     SYNC_TARGETS,
     ProfileGenerationService,
     build_snapshot,
@@ -45,11 +47,13 @@ class FakeRepository:
         assessment: dict[str, Any] | None = None,
         current: dict[str, Any] | None = None,
         next_version: int = 2,
+        memories: list[dict[str, Any]] | None = None,
     ) -> None:
         self._profile = profile
         self._assessment = assessment
         self._current = current
         self._next_version = next_version
+        self._memories = memories or []
         self.executed: list[str] = []
         self.inserted_snapshot: dict[str, Any] | None = None
         self.sync_events: list[dict[str, Any]] = []
@@ -66,6 +70,10 @@ class FakeRepository:
         self.executed.append("current_snapshot")
         return self._current
 
+    async def active_memories(self, _cid: int) -> list[dict[str, Any]]:
+        self.executed.append("active_memories")
+        return self._memories
+
     async def next_version(self, _cid: int) -> int:
         self.executed.append("next_version")
         return self._next_version
@@ -179,7 +187,13 @@ async def test_first_generation_writes_two_events_and_clears_old_current(
     # `aggregate_type='profile'` 由仓储层固定写入（不在 kwargs 里），此处断言服务传入的实体标识
     assert len({e["aggregate_uuid"] for e in repo.sync_events}) == 1
     assert all(e["aggregate_uuid"] == result.profile_uuid for e in repo.sync_events)
-    assert all(e["operation"] == "UPSERT" for e in repo.sync_events)
+    # 断言取值本身，且**与消费端领取条件对齐**：outbox worker 只领 `pending`/`failed`，
+    # 写成别的取值事件就永远没人消费。这里不再硬编码字面量（硬编码正是当初跑偏的原因）。
+    assert all(e["operation"] == SYNC_OPERATION_UPSERT for e in repo.sync_events)
+    assert all(e["status"] == SYNC_STATUS_PENDING for e in repo.sync_events)
+    assert SYNC_OPERATION_UPSERT == "upsert"
+    assert SYNC_STATUS_PENDING == "pending"
+    assert set(SYNC_TARGETS) == {"milvus", "neo4j"}
     # 顺序：清旧当前标记必须在插新版本之前（否则撞唯一键）
     assert repo.executed.index("clear_current") < repo.executed.index("insert_snapshot")
     # 新版本标为当前
@@ -187,6 +201,58 @@ async def test_first_generation_writes_two_events_and_clears_old_current(
     assert repo.inserted_snapshot["version"] == 2
 
 
+@pytest.mark.asyncio
+async def test_payload_carries_memory_sources_for_projection(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """payload 必须带 `memory_sources` 与 `profile_version`。
+
+    这是 Milvus 长期记忆投影的输入契约：适配器要 `memory_sources` 才知道往向量库
+    写什么，要 `profile_version`（或 `version`）才认得出这一批属于哪个画像版本。
+    缺了它，事件能被领取、却什么也投影不出来——属于"静默空转"，必须由测试挡住。
+    """
+    memories = [{
+        "memory_uuid": "11111111-2222-3333-4444-555555555555",
+        "memory_key": "preference:risk_level",
+        "content": "稳健型",
+        "memory_type": "preference",
+        "confidence": 0.9,
+        "version": 1,
+        "valid_until": None,
+    }]
+    repo = FakeRepository(
+        profile=profile_row(), assessment=assessment_row(), current=None, memories=memories
+    )
+
+    result = await service(monkeypatch, repo).generate(9102, now=NOW)
+
+    for sync_event in repo.sync_events:
+        payload = sync_event["payload"]
+        assert payload["profile_version"] == result.version
+        assert payload["version"] == result.version
+        sources = payload["memory_sources"]
+        assert len(sources) == 1
+        assert sources[0]["memory_uuid"] == memories[0]["memory_uuid"]
+        assert sources[0]["memory_key"] == "preference:risk_level"
+
+
+@pytest.mark.asyncio
+async def test_payload_memory_sources_is_empty_without_active_memories(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """没有有效记忆时给**空列表**（而不是省略该键）。
+
+    省略键会让适配器的 `memory_sources is invalid` 报错、事件反复重试直至死信；
+    空列表是"确实没有要投影的记忆"，语义不同。这里把这个区别钉住。
+    """
+    repo = FakeRepository(profile=profile_row(), assessment=assessment_row(), current=None)
+
+    await service(monkeypatch, repo).generate(9102, now=NOW)
+
+    for sync_event in repo.sync_events:
+        assert sync_event["payload"]["memory_sources"] == []
+
+
 @pytest.mark.asyncio
 async def test_each_version_gets_a_fresh_profile_uuid(monkeypatch: pytest.MonkeyPatch) -> None:
     """`profile_uuid` 有唯一键，**每个版本必须用新 uuid**（实测撞过 Duplicate entry）。"""

tests/unit/worker/test_memory_sync_outbox_worker.py

@@ -0,0 +1,137 @@
+"""`MemorySyncOutboxWorker` 的定向测试。
+
+4 个用例移植自同事 `ZSY_develop` 的
+`tests/unit/worker/test_memory_sync_outbox_worker.py`；最后一个用例是本仓新增的
+**契约回归测试**——它是这次整条链故障的根因所在，必须有人守着。
+"""
+
+from datetime import datetime
+
+import pytest
+
+from app.model.memory import MemorySyncOutbox
+from app.worker.memory_sync_outbox_worker import MemorySyncOutboxWorker
+
+
+class FakeSession:
+    def __init__(self, event: MemorySyncOutbox | None) -> None:
+        self.event = event
+        self.commits = 0
+        self.rollbacks = 0
+
+    async def __aenter__(self) -> "FakeSession":
+        return self
+
+    async def __aexit__(self, *args: object) -> None:
+        return None
+
+    async def scalar(self, statement: object) -> MemorySyncOutbox | None:
+        del statement
+        return self.event
+
+    async def commit(self) -> None:
+        self.commits += 1
+
+    async def rollback(self) -> None:
+        self.rollbacks += 1
+
+
+def event(*, target: str = "neo4j", retry_count: int = 0) -> MemorySyncOutbox:
+    return MemorySyncOutbox(
+        id=1, event_uuid="event-1", aggregate_type="profile",
+        aggregate_uuid="profile-1", aggregate_version=1, target_store=target,
+        operation="upsert", payload={"customer_id": 7}, status="pending",
+        retry_count=retry_count, next_retry_at=None, last_error=None,
+        created_at=datetime(2026, 1, 1), processed_at=None,
+    )
+
+
+@pytest.mark.asyncio
+async def test_success_marks_event_processed() -> None:
+    item = event()
+    session = FakeSession(item)
+    seen: list[dict[str, object]] = []
+
+    async def handler(payload: dict[str, object]) -> None:
+        seen.append(payload)
+
+    worker = MemorySyncOutboxWorker({"neo4j": handler}, session_factory=lambda: session)
+    assert await worker.run_once() is True
+    assert seen == [{"customer_id": 7}]
+    assert item.status == "processed"
+    assert item.processed_at is not None
+    assert session.commits == 1
+
+
+@pytest.mark.asyncio
+async def test_handler_failure_uses_backoff_and_keeps_event() -> None:
+    item = event()
+    session = FakeSession(item)
+
+    async def handler(payload: dict[str, object]) -> None:
+        del payload
+        raise TimeoutError
+
+    worker = MemorySyncOutboxWorker({"neo4j": handler}, session_factory=lambda: session)
+    assert await worker.run_once() is True
+    assert item.status == "failed"
+    assert item.retry_count == 1
+    assert item.next_retry_at is not None
+    assert item.last_error == "TimeoutError"
+
+
+@pytest.mark.asyncio
+async def test_fifth_failure_enters_dead_state() -> None:
+    item = event(retry_count=4)
+    session = FakeSession(item)
+
+    async def handler(payload: dict[str, object]) -> None:
+        del payload
+        raise RuntimeError
+
+    worker = MemorySyncOutboxWorker({"neo4j": handler}, session_factory=lambda: session)
+    await worker.run_once()
+    assert item.status == "dead"
+    assert item.retry_count == 5
+    assert item.next_retry_at is None
+
+
+@pytest.mark.asyncio
+async def test_missing_handler_enters_dead_state_without_external_call() -> None:
+    item = event(target="milvus")
+    session = FakeSession(item)
+    worker = MemorySyncOutboxWorker({"neo4j": lambda _: None}, session_factory=lambda: session)
+
+    assert await worker.run_once() is True
+    assert item.status == "dead"
+    assert item.last_error == "target_handler_not_configured"
+
+
+# --- 本仓新增：契约回归 ------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_no_handler_is_registered_for_uppercase_target_store() -> None:
+    """契约回归：`target_store` 只能是小写 `milvus`/`neo4j`。
+
+    背景（这条链真实故障的根因）：`profile_generation_service` 曾照 `docs/00` §6.4.6
+    写作大写 `MILVUS`/`NEO4J` + 中文状态 `待处理`，而本 worker 按 `target_store` 的**值**
+    分派 handler、且只领 `{"pending","failed"}`。结果该事件两个条件都不满足，
+    **任何消费者都领不到，永久滞留且不报错**（唯一键 `(event_uuid, target_store)`
+    对大小写没有约束，所以静默）。
+
+    本用例把"大写分派不到"这一事实钉住：将来谁把取值改回大写，这里会立刻红。
+    """
+    item = event(target="MILVUS", retry_count=0)
+    session = FakeSession(item)
+    called: list[object] = []
+
+    async def handler(payload: dict[str, object]) -> None:
+        called.append(payload)
+
+    worker = MemorySyncOutboxWorker({"milvus": handler}, session_factory=lambda: session)
+    assert await worker.run_once() is True
+
+    assert called == []  # 大写键分派不到 milvus handler
+    assert item.status == "dead"
+    assert item.last_error == "target_handler_not_configured"

tools/normalize_memory_sync_outbox.py

@@ -0,0 +1,91 @@
+"""数据订正：把 `memory_sync_outbox` 的历史取值改齐为全仓小写口径。
+
+**默认 dry-run**（只报告将影响几行），加 `--apply` 才真正提交。幂等，可重复运行。
+
+### 背景
+
+`docs/00` §6.4.6 的取值栏曾写作大写 `MILVUS`/`NEO4J`、`UPSERT` 与中文 `待处理`，
+与全仓实现从未对齐。按那份文档写入的事件会**静默失效**：
+
+- 消费端 `MemorySyncOutboxWorker` 按 `handlers.get(event.target_store)` 分派 handler；
+- 领取条件是 `status.in_({"pending","failed"})`。
+
+大写 + 中文两个条件都不满足 ⇒ 事件任何消费者都领不到、永久滞留且不报错
+（唯一键 `(event_uuid, target_store)` 对大小写没有约束，MySQL 也不会报错）。
+
+因此凡是在本仓写入过 `memory_sync_outbox` 的环境，都可能有这类脏行。
+
+### 安全口径
+
+- 只改 `target_store` / `operation` / `status` 的**值**；
+- **不触碰**主键、唯一键 `uk_memory_sync_event (event_uuid, target_store)`、
+  `payload`、时间戳、`retry_count`；
+- 就地改值不会新增行（两组取值大小写不同，本会各自成行，这里把它们并到正确的一组）；
+- 中文字符不受 `LOWER()` 影响（非 ASCII 字节不变），故 `LOWER()` 是安全的。
+
+用法：
+    .\\.venv\\Scripts\\python.exe tools\\normalize_memory_sync_outbox.py           # 先看
+    .\\.venv\\Scripts\\python.exe tools\\normalize_memory_sync_outbox.py --apply   # 再改
+"""
+
+import asyncio
+import sys
+
+from sqlalchemy import text
+
+from app.infrastructure.db import engine
+
+_STATUS_CASE = """
+    CASE status
+        WHEN '待处理' THEN 'pending'
+        WHEN '处理中' THEN 'processing'
+        WHEN '已完成' THEN 'processed'
+        WHEN '失败' THEN 'failed'
+        ELSE LOWER(status)
+    END
+"""
+
+_UPDATE = text(f"""
+    UPDATE memory_sync_outbox
+       SET target_store = LOWER(target_store),
+           operation = LOWER(operation),
+           status = {_STATUS_CASE}
+     WHERE target_store <> LOWER(target_store)
+        OR operation <> LOWER(operation)
+        OR status <> {_STATUS_CASE}
+""")
+
+_GROUP = text(
+    "SELECT target_store, operation, status, COUNT(*) n "
+    "FROM memory_sync_outbox GROUP BY target_store, operation, status"
+)
+
+_DETAIL = text(
+    "SELECT id, event_uuid, target_store, operation, status, retry_count "
+    "FROM memory_sync_outbox ORDER BY id"
+)
+
+
+async def main(apply: bool) -> int:
+    async with engine.connect() as conn:
+        before = (await conn.execute(_GROUP)).mappings().all()
+        print("改前:", [dict(row) for row in before] or "（空表）")
+
+        affected = (await conn.execute(_UPDATE)).rowcount
+        print("将订正行数:", affected)
+
+        if not apply:
+            print("dry-run：未提交。确认无误后加 --apply 再运行。")
+            await conn.rollback()
+            return 0
+
+        await conn.commit()
+        after = (await conn.execute(_GROUP)).mappings().all()
+        print("改后:", [dict(row) for row in after] or "（空表）")
+        for row in (await conn.execute(_DETAIL)).mappings().all():
+            print("  ", dict(row))
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(asyncio.run(main("--apply" in sys.argv)))

tools/setup_milvus_profile_collection.py

@@ -0,0 +1,144 @@
+"""幂等创建长期记忆画像向量集合 `user_long_term_memory_v1`。
+
+写入侧是 `app/infrastructure/milvus_profile_projection.py`，字段必须与之一致。
+
+用法：
+    .\\.venv\\Scripts\\python.exe tools\\setup_milvus_profile_collection.py
+
+安全口径（与 `setup_milvus_knowledge_collections.py` 相同：这是**共享** Milvus 实例，
+里面还有别的项目在用的集合）：
+- 只碰 `user_long_term_memory_v1` 这一个集合，绝不 list 后批量删除；
+- 集合已存在时**只做结构比对并报告**，不覆盖、不删重建；
+- 结构不一致时明确报错退出，由人决定怎么处理，避免静默丢数据。
+"""
+
+import asyncio
+import sys
+from typing import Any
+
+COLLECTION = "user_long_term_memory_v1"
+VECTOR_FIELD = "embedding"
+VECTOR_DIM = 1024
+PRIMARY_FIELD = "memory_uuid"
+
+#: (字段名, 最大长度)。`memory_uuid` 是主键（UUID 字符串）。
+VARCHAR_FIELDS: tuple[tuple[str, int], ...] = (
+    ("memory_uuid", 64),
+    ("content", 2048),
+    ("memory_type", 32),
+    ("memory_key", 64),
+    ("status", 16),
+)
+#: INT64 字段；可空，因为 `valid_until_ts` 对永久记忆为空。
+INT64_FIELDS: tuple[str, ...] = (
+    "customer_id",
+    "version",
+    "valid_until_ts",
+    "updated_at_ts",
+)
+FLOAT_FIELDS: tuple[str, ...] = ("confidence",)
+
+
+def _build_schema() -> Any:
+    from pymilvus import DataType, MilvusClient  # type: ignore[import-untyped]
+
+    schema = MilvusClient.create_schema(auto_id=False, enable_dynamic_field=False)
+    for name, max_length in VARCHAR_FIELDS:
+        schema.add_field(
+            field_name=name,
+            datatype=DataType.VARCHAR,
+            max_length=max_length,
+            is_primary=(name == PRIMARY_FIELD),
+            nullable=False,
+        )
+    for name in INT64_FIELDS:
+        # 可空：`valid_until_ts` 对永久记忆必须能不写。
+        schema.add_field(
+            field_name=name, datatype=DataType.INT64, nullable=name != "customer_id"
+        )
+    for name in FLOAT_FIELDS:
+        schema.add_field(field_name=name, datatype=DataType.DOUBLE, nullable=False)
+    schema.add_field(
+        field_name=VECTOR_FIELD, datatype=DataType.FLOAT_VECTOR, dim=VECTOR_DIM
+    )
+    return schema
+
+
+def _build_index_params() -> Any:
+    from pymilvus import MilvusClient  # type: ignore[import-untyped]
+
+    index_params = MilvusClient.prepare_index_params()
+    index_params.add_index(
+        field_name=VECTOR_FIELD,
+        index_name="profile_vector_index",
+        index_type="AUTOINDEX",
+        metric_type="COSINE",
+    )
+    return index_params
+
+
+def describe_mismatch(described: dict[str, Any]) -> list[str]:
+    """比对已存在集合与期望结构，返回差异列表（一致时为空）。"""
+    problems: list[str] = []
+    actual = {field["name"]: field for field in described.get("fields", [])}
+    expected_names = (
+        [name for name, _ in VARCHAR_FIELDS] + list(INT64_FIELDS)
+        + list(FLOAT_FIELDS) + [VECTOR_FIELD]
+    )
+    for name in expected_names:
+        if name not in actual:
+            problems.append(f"缺少字段 {name}")
+    for name, _ in VARCHAR_FIELDS:
+        if name in actual and not actual[name].get("is_primary") and name == PRIMARY_FIELD:
+            problems.append(f"{name} 不是主键")
+    vector = actual.get(VECTOR_FIELD)
+    if vector is not None:
+        params = vector.get("params") or {}
+        dim = params.get("dim")
+        if dim is not None and int(dim) != VECTOR_DIM:
+            problems.append(f"{VECTOR_FIELD} 维度是 {dim}，期望 {VECTOR_DIM}")
+    return problems
+
+
+async def ensure_collection(uri: str, token: str = "") -> str:
+    """返回 `created` / `exists` / `conflict:<原因>`。"""
+    from pymilvus import AsyncMilvusClient  # type: ignore[import-untyped]
+
+    client = AsyncMilvusClient(uri=uri, token=token or None)
+    try:
+        if await client.has_collection(COLLECTION):
+            described = await client.describe_collection(COLLECTION)
+            problems = describe_mismatch(described)
+            if problems:
+                return "conflict:" + "; ".join(problems)
+            return "exists"
+        await client.create_collection(
+            collection_name=COLLECTION,
+            schema=_build_schema(),
+            index_params=_build_index_params(),
+        )
+        return "created"
+    finally:
+        close = getattr(client, "close", None)
+        if close is not None:
+            await close()
+
+
+async def main() -> int:
+    from app.core.config import get_settings
+
+    settings = get_settings()
+    uri = settings.milvus_uri
+    if not uri:
+        print("未配置 milvus_uri，无法创建集合")
+        return 1
+    outcome = await ensure_collection(uri, settings.milvus_token or "")
+    print(f"{COLLECTION}: {outcome}")
+    if outcome.startswith("conflict:"):
+        print("结构不一致，未做任何修改。请人工确认后再处理。")
+        return 2
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(asyncio.run(main()))