diff --git a/AGENTS.md b/AGENTS.md
index 1c14b00..fcda30b 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -112,7 +112,13 @@
   （唯一键 `(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 记忆投影链路落地后实测；
+- ⚠️ **一张表只能有一个 ORM 类**：`app/model/` 下曾出现**两个类都映射 `profile_snapshots`**
+  （`profile.py` 与 `risk_questionnaire.py`），各自单独导入都没事，**同时导入即抛**
+  `InvalidRequestError: Table 'profile_snapshots' is already defined for this MetaData instance`
+  —— Worker 既要重建画像又要处理投顾问卷，因此**真的被打挂过**（库里 `memory_sync_outbox`
+  留下 `last_error='InvalidRequestError'` 的行）。2026-09-12 已修为 re-export，见 `docs/32` §6.2。
+  **新增模型前先搜一遍 `__tablename__` 有没有被占用。**
+- 测试基线：`2 failed, 1312 passed, 2 skipped`（2026-09-12 记忆投影链路落地后实测；
   此前为 `3 failed, 1219 passed`——第 3 个失败
   `tests/unit/repository/test_fund_readonly_contract.py` 已由投顾线合入主干时修复，**不要再当既有缺陷引用**）。
   剩下 2 个失败**都不是代码缺陷**，接手时不要"修"它们：
diff --git a/app/model/risk_questionnaire.py b/app/model/risk_questionnaire.py
index 6bd96bb..def63e7 100644
--- a/app/model/risk_questionnaire.py
+++ b/app/model/risk_questionnaire.py
@@ -1,30 +1,39 @@
-"""Additive projection model for opening-risk questionnaire profiles."""
+"""开户风险问卷画像的模型出口（re-export）。
 
-from datetime import datetime
-from typing import Any
+⚠️ 2026-09-12 修复的**重复定义缺陷**
+===================================
 
-from sqlalchemy import JSON, BigInteger, DateTime, String
-from sqlalchemy.orm import Mapped, mapped_column
+本模块原先**自己定义**了一个 `ProfileSnapshot` 类，映射的却是主干
+`app/model/profile.py` 里已有的同名表 `profile_snapshots`。
+
+SQLAlchemy 不允许两个类映射同一张表，因此**任何同时导入本模块与
+`app.model.profile` 的进程都会抛**：
+
+    InvalidRequestError: Table 'profile_snapshots' is already defined
+    for this MetaData instance.
+
+**实测影响（不是理论风险）**：
+
+- 单独导入 `app.main` / `app.worker.runtime` 都正常；但
+  `profile_assembly_service` 与 `risk_questionnaire_service` **同时**导入即崩。
+- Worker 在同一个进程里既要处理 `profile.rebuild_requested`（走 `app.model.profile`），
+  又要处理投顾风险问卷（走本模块）——因此这是**会打挂 Worker 的缺陷**。
+- 库里已留痕：`memory_sync_outbox` 中 `target_store='neo4j'` 的行
+  `last_error='InvalidRequestError'` 就是这个原因，不是图库故障。
+
+**修法**：本模块不再重复定义，改为从 `app.model.profile` 转出（re-export）。
+因此 `from app.model.risk_questionnaire import ProfileSnapshot` 的既有调用点
+**无需改动**（4 处：`risk_questionnaire_repository`、`profile_governance_service`、
+`risk_questionnaire_service` 与对应单测）。
+
+**字段等价性核查**：原定义比 `app.model.profile` 多映射了一个 `current_customer_id`。
+经全仓核查**无人使用**该属性（投顾线只用到 `id` / `customer_id` / `version` / `is_current`），
+且 `app.model.profile` 明确注明该列由数据库维护、**故意不映射**，故不保留。
+`app.model.profile` 的列类型（`CHAR(36)` / `CHAR(64)` / `Boolean`）与库中实际 DDL 一致，
+比原定义的 `String(36)` / `String(64)` 更准确。
+"""
 
-from app.model.base import Base
 from app.model.fund import FundRiskAssessment as RiskAssessment
-
-
-class ProfileSnapshot(Base):
-    __tablename__ = "profile_snapshots"
-
-    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
-    profile_uuid: Mapped[str | None] = mapped_column(String(36), unique=True)
-    customer_id: Mapped[int] = mapped_column(BigInteger, nullable=False)
-    version: Mapped[int] = mapped_column(BigInteger, nullable=False)
-    snapshot: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False)
-    generation_basis: Mapped[dict[str, Any] | None] = mapped_column(JSON)
-    snapshot_hash: Mapped[str | None] = mapped_column(String(64))
-    is_current: Mapped[bool] = mapped_column(nullable=False, default=False)
-    current_customer_id: Mapped[int | None] = mapped_column(BigInteger)
-    generated_at: Mapped[datetime | None] = mapped_column(DateTime)
-    created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False)
-    updated_at: Mapped[datetime] = mapped_column(DateTime, nullable=False)
-
+from app.model.profile import ProfileSnapshot
 
 __all__ = ["ProfileSnapshot", "RiskAssessment"]
diff --git a/app/worker/runtime.py b/app/worker/runtime.py
index abb8394..9d968b8 100644
--- a/app/worker/runtime.py
+++ b/app/worker/runtime.py
@@ -485,8 +485,9 @@ class WorkerRuntime:
         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(payload)
+            await projection.upsert(effective)
 
         async def project_neo4j(payload: dict[str, Any]) -> None:
             raw_customer_id = payload.get("customer_id")
@@ -518,6 +519,66 @@ class WorkerRuntime:
             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]:
         """向量化一条记忆正文；端点走与知识向量化同一套已批准端点解析。
 
diff --git a/docs/32-记忆投影链路实现说明.md b/docs/32-记忆投影链路实现说明.md
index 9356ac8..66e4145 100644
--- a/docs/32-记忆投影链路实现说明.md
+++ b/docs/32-记忆投影链路实现说明.md
@@ -167,6 +167,51 @@
 向量化复用与知识向量同一套已批准端点解析（`agent_type="memory_recall"`,
 `task_type="embedding"`）。
 
+### 6.1 `memory_sources` 缺失时的兜底（`_with_memory_sources`）
+
+`memory_sources` 是本线新增的投影入参，而**投顾线两处生产者**
+（`profile_governance_service` / `risk_questionnaire_service`）发的 payload 是
+`{customer_id, profile_uuid, version, profile}`，**没有**这个键。若不处理，它们每次画像
+变更都会因 `memory_sources is invalid` 失败重试直至死信。
+
+消费端因此做了兜底：**键缺失或为 `None`** 时，回退为查询该客户 `memory_unit` 中
+`status='active'` 的记忆，并**记一条 warning**（使"谁没提供"保持可见）。
+
+为什么允许兜底：缺失表示生产者不知道要提供，属契约演进期的正常情况，且**语义成立**
+——长期记忆是**客户级**的、不是画像版本级的，每条记忆自带 `version`，适配器按
+`memory_uuid + version` 幂等，"用的是哪一版"仍然确定。
+
+**兜底不掩盖真错误**（有测试守着）：键**存在但格式不对**（例如是字符串）时**不兜底**，
+原样放行交给适配器失败关闭。实现上用**键存在性**判断而不是 `isinstance`——后者会把
+"缺失"与"格式错"混为一谈，那正是本模块初版实现里的一个真 bug，被测试抓出来后修正。
+
+---
+
+## 6.2 顺带修掉的独立缺陷：`profile_snapshots` 被重复定义
+
+**发现路径**：验证 `neo4j` 分支时，库里那行 `last_error='InvalidRequestError'`。
+原以为是图库故障，追下去发现是**模型层缺陷**：
+
+- `app/model/profile.py` → `ProfileSnapshot` 映射 `profile_snapshots`
+- `app/model/risk_questionnaire.py` → **另一个** `ProfileSnapshot` 也映射 `profile_snapshots`
+
+SQLAlchemy 不允许两个类映射同一张表。实测：
+
+| 场景 | 结果 |
+|---|---|
+| 单独导入 `app.main` / `app.worker.runtime` | 正常 |
+| 单独导入 `profile_assembly_service` / `risk_questionnaire_service` | 正常 |
+| **两者同时导入** | `InvalidRequestError: Table 'profile_snapshots' is already defined` |
+
+**影响**：Worker 在同一个进程里既要处理 `profile.rebuild_requested`（走 `app.model.profile`），
+又要处理投顾风险问卷（走 `risk_questionnaire.py`）——所以这是**会打挂 Worker 的缺陷**，
+不是理论风险。
+
+**修法**：`app/model/risk_questionnaire.py` 不再重复定义，改为从 `app.model.profile`
+转出（re-export），既有 4 处 `from app.model.risk_questionnaire import ProfileSnapshot`
+**无需改动**。原定义多映射的 `current_customer_id` 经全仓核查**无人使用**，故不保留
+（`app.model.profile` 明确注明该列由数据库维护、故意不映射）。
+
 ---
 
 ## 7. 真机验证证据（2026-09-12）
@@ -178,8 +223,9 @@
 | 真实写入 + 回读 | 2 条可投影键写入成功并可回读（内容/版本正确） |
 | 不可投影键 | `constraint:liquidity` **未写入**（跳过生效，未毒死整批） |
 | 测试数据清理 | 已按 `customer_id=999999` 删除，集合残留 **0** 条 |
-| 单元测试 | 新增 17 个（适配器 10 + worker 5 + 生产端 2），全过 |
-| 全量回归 | `2 failed, 1307 passed, 2 skipped` —— 与基线一致，**无新增失败** |
+| **消费端全路径**（补验） | 见下方 7.1 |
+| 单元测试 | 新增 **22** 个（适配器 10 + worker 5 + 生产端 2 + 消费端兜底 5），全过 |
+| 全量回归 | `2 failed, 1312 passed, 2 skipped` —— 与基线一致，**无新增失败** |
 | mypy | `Success: no issues found in 227 source files` |
 
 > 全量的 2 个失败是既有环境相关项（`test_offsite_document_recognition_adapter.py`
@@ -188,24 +234,54 @@
 > 真机注意：Milvus 写入后**短时间内可能查不到**（索引尚未可见），
 > 验证脚本按重试处理；同理删除后立即查询可能仍返回旧行，需重查确认。
 
+### 7.1 消费端全路径验证（2026-09-12 补做）
+
+此前"整合验证"是**直接调适配器**，跳过了 outbox 的领取→分派→状态更新。
+后补做了两轮，覆盖失败分支与成功分支：
+
+**失败分支**（Milvus 断开时实测）：
+
+| id | `last_error` | 说明 |
+|---|---|---|
+| 5 | `ValueError` | payload 缺 `memory_sources`（兜底上线前的旧行） |
+| 6 | `InvalidRequestError` | 模型重复定义缺陷（见 §6.2），**修复后此错误消失** |
+| 9 | `RecoverableAgentError` | Milvus 不可达——如实失败，不伪造成功 |
+
+这证明全路径都工作：行被领取 ✓、按 `target_store` 分派 handler ✓、
+handler 异常被捕获 ✓、`status`/`retry_count`/`last_error`/`next_retry_at` 正确落库 ✓。
+
+**成功分支**（注入替身向量客户端，不依赖真实 Milvus）：
+
+- outbox 行 → `status=processed`、`processed_at` 已写、`last_error` 清空 ✓
+- 不可投影的 `constraint:liquidity` **被跳过**（只写 1 行而非 2 行）✓
+- 向量维度 1024 ✓；字符串客户号 `"999996"` → int ✓
+- **手机号脱敏生效**：`稳健型投资者，手机号 [手机号已隐藏] 请勿外泄` ✓
+
+**兜底的实证**：历史行 `id=5`（payload 无 `memory_sources`）经兜底回退查询后
+成功投递为 `processed`，日志留
+`profile projection payload has no memory_sources (customer_id=9102); fell back to 0 active memories`。
+
+> 补验时的环境限制：Docker Desktop 中途崩溃（`milvus-standalone` 内嵌 etcd panic、
+> Neo4j `Exited(1)`），因此 `id=6`（neo4j 分支）停在 `failed`/`RecoverableAgentError`。
+> **那是环境不可用，不是代码缺陷**——图库不可用时如实失败、不伪造成功正是设计口径。
+
 ---
 
 ## 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 的取值栏与实现不一致：按评审要求**未改基线文档**，
+1. **常驻 Worker 未运行**：`memory_unit`、`user_facts`、`episodes` 目前都是 **0 行**。
+   代码链路是通的（§7.1 已用真实 outbox 行验证消费端），但没有 Worker 在跑，
+   所以记忆永远不会被抽取出来。积压量（2026-09-12 实测）：
+   `agent.run_requested` **431**、`profile.rebuild_requested` **220**、
+   `agent.run_completed` **61**、`memory.extraction_requested` **58**。
+   起 `python -X utf8 -m app.worker --once`（或常驻）即开始消费。
+   ⚠️ 会派发真实 agent 任务、产生模型调用费用，故未擅自启动。
+2. **投顾线两处生产者的 `memory_sources`**：其 payload 仍**没有**这个键。
+   本线已在消费端加了兜底（§6.1），因此**不再会死信**；但根治仍应由投顾线补上
+   （或明确这两个来源是否也要投影长期记忆）。属架构师线，本线未改其生产者代码。
+3. **`profile_snapshots` 重复定义缺陷**（§6.2）：本线已修，但它源自投顾线的模型文件，
+   需让架构师知晓，以免在别处再引入同名定义。
+4. `docs/00` §6.4.6 的取值栏与实现不一致：按评审要求**未改基线文档**，
    实际口径以本文第 2 节为准。
 
 ---
@@ -218,12 +294,16 @@
 - `app/infrastructure/milvus_profile_vector_client.py`
 - `app/worker/memory_sync_outbox_worker.py`（移植）
 - `tools/setup_milvus_profile_collection.py`
+- `tools/normalize_memory_sync_outbox.py`（历史取值订正，默认 dry-run、幂等）
 - `tests/unit/infrastructure/test_milvus_profile_projection.py`
 - `tests/unit/worker/test_memory_sync_outbox_worker.py`
+- `tests/unit/worker/test_runtime_profile_projection.py`（消费端兜底 5 用例）
 
 **修改**
 - `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` 接线）
+- `app/worker/runtime.py`（`consume_profile_projections()`、`_with_memory_sources()` 兜底、`run_once` 接线）
+- `app/model/risk_questionnaire.py`（**修重复定义**：改为 re-export `app.model.profile` 的 `ProfileSnapshot`，见 §6.2）
 - `tests/unit/service/test_profile_generation_service.py`（+2 用例、断言改引用常量）
+- `AGENTS.md`（新增 `memory_sync_outbox` 取值口径与 Windows 中文输出两条易错点；校正测试基线/mypy 数字）
diff --git a/tests/unit/worker/test_runtime_profile_projection.py b/tests/unit/worker/test_runtime_profile_projection.py
new file mode 100644
index 0000000..df28268
--- /dev/null
+++ b/tests/unit/worker/test_runtime_profile_projection.py
@@ -0,0 +1,130 @@
+"""画像投影入参的契约与兜底：`WorkerRuntime._with_memory_sources`。
+
+背景（本测试要拦住的真实故障）：`memory_sources` 是本仓新增的投影入参，而投顾线两处
+生产者（`profile_governance_service` / `risk_questionnaire_service`）发的 payload 是
+`{customer_id, profile_uuid, version, profile}`，**没有**这个键。若不兜底，它们每次画像
+变更都会因 `memory_sources is invalid` 失败重试直至死信（库里已留 `ValueError` 行痕）。
+
+同时要钉住"兜底不得掩盖格式错"：已提供但格式不合法时，仍由适配器失败关闭。
+"""
+
+from typing import Any
+
+import pytest
+
+from app.repository.profile_repository import ProfileRepository
+from app.worker.runtime import WorkerRuntime
+
+
+class FakeSession:
+    def __init__(self) -> None:
+        self.closed = False
+
+    async def __aenter__(self) -> "FakeSession":
+        return self
+
+    async def __aexit__(self, *args: object) -> None:
+        self.closed = True
+
+
+def memory_row() -> dict[str, Any]:
+    return {
+        "memory_uuid": "11111111-2222-3333-4444-555555555555",
+        "memory_key": "preference:risk_level",
+        "content": "稳健型",
+        "memory_type": "preference",
+        "confidence": 0.9,
+        "version": 3,
+        "valid_until": None,
+    }
+
+
+@pytest.mark.asyncio
+async def test_provided_memory_sources_is_passed_through_unchanged() -> None:
+    """生产端已提供时必须**原样**使用事件里的确定快照，不回查数据库。"""
+    runtime = WorkerRuntime()
+    sources = [{"memory_uuid": "u-1", "memory_key": "preference:horizon"}]
+    payload = {"customer_id": 9102, "profile_version": 2, "memory_sources": sources}
+
+    result = await runtime._with_memory_sources(payload)
+
+    assert result is payload
+    assert result["memory_sources"] is sources
+
+
+@pytest.mark.asyncio
+async def test_missing_memory_sources_falls_back_to_active_memories(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """缺失时回退为查询当前有效记忆，并把它组装成适配器认得的形状。"""
+    runtime = WorkerRuntime()
+    calls: list[int] = []
+
+    async def fake_active_memories(self: Any, customer_id: int) -> list[dict[str, Any]]:
+        calls.append(customer_id)
+        return [memory_row()]
+
+    monkeypatch.setattr(ProfileRepository, "active_memories", fake_active_memories)
+    monkeypatch.setattr("app.worker.runtime.SessionFactory", FakeSession)
+
+    payload = {"customer_id": "9102", "profile_version": 2}  # 字符串客户号，复现生产端形态
+    result = await runtime._with_memory_sources(payload)
+
+    assert calls == [9102]
+    sources = result["memory_sources"]
+    assert len(sources) == 1
+    assert sources[0]["memory_key"] == "preference:risk_level"
+    assert sources[0]["memory_uuid"] == memory_row()["memory_uuid"]
+    assert sources[0]["version"] == 3
+    # 原 payload 的其余键必须保留（适配器还要用 profile_version 等）
+    assert result["profile_version"] == 2
+
+
+@pytest.mark.asyncio
+async def test_empty_active_memories_yields_empty_list_not_missing_key(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """没有有效记忆时补**空列表**（不是删掉键）。
+
+    空列表是"确实没有要投影的记忆"，适配器接受并写 0 行；缺键则会 `memory_sources is
+    invalid` 报错、退避重试到死信。两者语义不同，必须区分。
+    """
+    runtime = WorkerRuntime()
+
+    async def none_active(self: Any, customer_id: int) -> list[dict[str, Any]]:
+        return []
+
+    monkeypatch.setattr(ProfileRepository, "active_memories", none_active)
+    monkeypatch.setattr("app.worker.runtime.SessionFactory", FakeSession)
+
+    result = await runtime._with_memory_sources({"customer_id": 9102})
+
+    assert result["memory_sources"] == []
+
+
+@pytest.mark.asyncio
+async def test_payload_without_customer_id_is_left_alone() -> None:
+    """没有客户号时无法兜底，原样返回、交给适配器失败关闭（不在此处静默造数据）。"""
+    runtime = WorkerRuntime()
+    payload = {"profile_version": 1}
+
+    result = await runtime._with_memory_sources(payload)
+
+    assert result is payload
+    assert "memory_sources" not in result
+
+
+@pytest.mark.asyncio
+async def test_non_list_memory_sources_is_not_silently_replaced() -> None:
+    """已提供但格式错（不是列表）时不兜底——那是真错误，必须由适配器报出来。
+
+    这条守住"兜底不得掩盖格式错"：否则生产者写错字段类型会被悄悄修好，
+    线上永远看不到问题。
+    """
+    runtime = WorkerRuntime()
+    payload = {"customer_id": 9102, "memory_sources": "not-a-list"}
+
+    result = await runtime._with_memory_sources(payload)
+
+    assert result is payload
+    assert result["memory_sources"] == "not-a-list"