wangjianlong_0626
8a0cbab636
背景：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，会被消费至死信，待架构师确认是否投影。
343 lines
14 KiB
Python
343 lines
14 KiB
Python
"""画像版本生成器单元测试（`docs/00` §6.4.6 的四条规则）。
|
||
|
||
覆盖四个方向，都要有牙：
|
||
|
||
1. **同事务两条事件**：一次生成必须同时投 `MILVUS` 与 `NEO4J` 两条，共用同一 `event_uuid`。
|
||
2. **幂等**：`snapshot_hash` 未变 → **不升版本、不投事件**（否则每次调用都灌 outbox）。
|
||
3. **旧版本置 0**：清旧当前标记必须发生在插新版本**之前**（唯一键约束）。
|
||
4. **失败关闭**：客户主表没有该行 → 抛错，不凭空造画像。
|
||
5. **字段对齐**：写入快照必须覆盖读取侧白名单所需字段（否则端点会返回空画像）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
from datetime import UTC, datetime, timedelta
|
||
from typing import Any
|
||
|
||
import pytest
|
||
|
||
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,
|
||
compute_hash,
|
||
)
|
||
|
||
NOW = datetime(2026, 9, 11, 12, 0, tzinfo=UTC)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 替身：只实现 ProfileRepository 用到的方法
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class FakeRepository:
|
||
"""记录调用顺序与写入内容；`executed` 用来断言"先清旧、后插新"。"""
|
||
|
||
def __init__(
|
||
self,
|
||
*,
|
||
profile: dict[str, Any] | None = None,
|
||
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]] = []
|
||
|
||
async def profile_row(self, _cid: int) -> dict[str, Any] | None:
|
||
self.executed.append("profile_row")
|
||
return self._profile
|
||
|
||
async def latest_assessment(self, _cid: int) -> dict[str, Any] | None:
|
||
self.executed.append("latest_assessment")
|
||
return self._assessment
|
||
|
||
async def current_snapshot(self, _cid: int) -> dict[str, Any] | None:
|
||
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
|
||
|
||
async def clear_current(self, _cid: int, *, now: datetime) -> None:
|
||
self.executed.append("clear_current")
|
||
|
||
async def insert_snapshot(self, **kwargs: Any) -> None:
|
||
self.executed.append("insert_snapshot")
|
||
self.inserted_snapshot = kwargs
|
||
|
||
def add_sync_event(self, **kwargs: Any) -> None:
|
||
self.executed.append("add_sync_event")
|
||
self.sync_events.append(kwargs)
|
||
|
||
|
||
def service(monkeypatch: pytest.MonkeyPatch, repo: FakeRepository) -> ProfileGenerationService:
|
||
monkeypatch.setattr(
|
||
"app.service.profile_generation_service.ProfileRepository", lambda _session: repo
|
||
)
|
||
return ProfileGenerationService(session=object()) # type: ignore[arg-type]
|
||
|
||
|
||
def profile_row(**overrides: Any) -> dict[str, Any]:
|
||
row: dict[str, Any] = {
|
||
"customer_id": 9102,
|
||
"investor_type": "C3",
|
||
"investment_horizon": "medium_term",
|
||
"trading_frequency": "medium",
|
||
"preferred_asset_class": ["bond_fund"],
|
||
"risk_tags": ["balanced"],
|
||
"behavior_score": 61,
|
||
"total_asset": "860000.00",
|
||
"last_active_at": NOW,
|
||
}
|
||
row.update(overrides)
|
||
return row
|
||
|
||
|
||
def assessment_row(valid_days: int = 200) -> dict[str, Any]:
|
||
return {
|
||
"id": 99102,
|
||
"questionnaire_version": "v2026.1",
|
||
"investor_type": "C3",
|
||
"assessed_at": NOW - timedelta(days=165),
|
||
"valid_until": (NOW + timedelta(days=valid_days)).replace(tzinfo=None),
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 1. 快照构建
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_snapshot_covers_every_field_the_read_side_needs() -> None:
|
||
"""写侧必须覆盖读取侧白名单所需字段，否则端点会返回空画像。"""
|
||
snapshot = build_snapshot(profile_row(), assessment_row(), now=NOW)
|
||
missing = [f for f in REQUIRED_SNAPSHOT_FIELDS if f not in snapshot]
|
||
assert not missing, f"快照缺少读取侧需要的字段：{missing}"
|
||
|
||
|
||
def test_snapshot_is_projectable_by_the_read_path() -> None:
|
||
"""端到端对齐：生成出来的快照经 `project_profile` 能得到非空投影。"""
|
||
snapshot = build_snapshot(profile_row(), assessment_row(), now=NOW)
|
||
projected = project_profile(snapshot, now=NOW)
|
||
assert projected["investor_type"] == "C3"
|
||
assert projected["assessment_expired"] is False
|
||
|
||
|
||
def test_expired_assessment_is_marked_expired_in_snapshot() -> None:
|
||
"""过期测评必须在快照里标成过期（与读取侧"按当前时间重算"同向）。"""
|
||
snapshot = build_snapshot(profile_row(), assessment_row(valid_days=-45), now=NOW)
|
||
assert snapshot["assessment_expired"] is True
|
||
assert project_profile(snapshot, now=NOW)["assessment_expired"] is True
|
||
|
||
|
||
def test_no_assessment_means_no_fabricated_validity() -> None:
|
||
"""没有测评行时**不写**有效期/过期标记，而不是伪造一个。"""
|
||
snapshot = build_snapshot(profile_row(), None, now=NOW)
|
||
assert "assessment_valid_until" not in snapshot
|
||
assert "assessment_expired" not in snapshot
|
||
|
||
|
||
def test_hash_is_stable_against_key_order() -> None:
|
||
"""哈希是幂等判据，不能被字典插入顺序影响。"""
|
||
a = {"investor_type": "C3", "behavior_score": 61}
|
||
b = {"behavior_score": 61, "investor_type": "C3"}
|
||
assert compute_hash(a) == compute_hash(b)
|
||
assert compute_hash({"x": 1}) != compute_hash({"x": 2})
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 2. 首次生成：两条事件 + 顺序
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_first_generation_writes_two_events_and_clears_old_current(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
repo = FakeRepository(profile=profile_row(), assessment=assessment_row(), current=None)
|
||
result = await service(monkeypatch, repo).generate(9102, now=NOW)
|
||
|
||
assert result.changed is True
|
||
assert result.version == 2
|
||
assert result.sync_events == 2
|
||
# 两个目标存储各一条，且**共用同一个 event_uuid**（唯一键是 (event_uuid, target_store)）
|
||
targets = [e["target_store"] for e in repo.sync_events]
|
||
assert targets == list(SYNC_TARGETS)
|
||
assert len({e["event_uuid"] for e in repo.sync_events}) == 1
|
||
# `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)
|
||
# 断言取值本身，且**与消费端领取条件对齐**：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")
|
||
# 新版本标为当前
|
||
assert repo.inserted_snapshot is not None
|
||
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）。"""
|
||
old = {"profile_uuid": "old-uuid", "version": 1, "snapshot_hash": "different"}
|
||
repo = FakeRepository(profile=profile_row(), assessment=assessment_row(), current=old)
|
||
result = await service(monkeypatch, repo).generate(9102, now=NOW)
|
||
|
||
assert result.profile_uuid != "old-uuid"
|
||
assert repo.inserted_snapshot is not None
|
||
assert repo.inserted_snapshot["profile_uuid"] == result.profile_uuid
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 3. 幂等
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_unchanged_content_does_not_bump_version_or_emit_events(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
snapshot = build_snapshot(profile_row(), assessment_row(), now=NOW)
|
||
current = {
|
||
"profile_uuid": "u1",
|
||
"version": 7,
|
||
"snapshot_hash": compute_hash(snapshot),
|
||
}
|
||
repo = FakeRepository(profile=profile_row(), assessment=assessment_row(), current=current)
|
||
result = await service(monkeypatch, repo).generate(9102, now=NOW)
|
||
|
||
assert result.changed is False
|
||
assert result.version == 7 # 不升版本
|
||
assert result.sync_events == 0 # 不投事件
|
||
assert "insert_snapshot" not in repo.executed
|
||
assert repo.sync_events == []
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_changed_content_bumps_version_and_reemits(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
current = {"profile_uuid": "u1", "version": 3, "snapshot_hash": "stale-hash"}
|
||
repo = FakeRepository(
|
||
profile=profile_row(total_asset="999999.00"),
|
||
assessment=assessment_row(),
|
||
current=current,
|
||
next_version=4,
|
||
)
|
||
result = await service(monkeypatch, repo).generate(9102, now=NOW)
|
||
|
||
assert result.changed is True
|
||
assert result.version == 4
|
||
assert result.sync_events == 2
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 4. 失败关闭
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_missing_customer_fails_closed(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
"""主表没有该客户 → 抛错，不凭空造画像。"""
|
||
repo = FakeRepository(profile=None)
|
||
with pytest.raises(ValidationAgentError, match="客户不存在"):
|
||
await service(monkeypatch, repo).generate(99999, now=NOW)
|
||
assert repo.inserted_snapshot is None
|
||
assert repo.sync_events == []
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 5. 事件载荷
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_event_payload_carries_version_and_hash(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
repo = FakeRepository(profile=profile_row(), assessment=assessment_row(), current=None)
|
||
result = await service(monkeypatch, repo).generate(9102, now=NOW)
|
||
|
||
payload = repo.sync_events[0]["payload"]
|
||
assert payload["version"] == result.version
|
||
assert payload["snapshot_hash"] == result.snapshot_hash
|
||
assert payload["customer_id"] == "9102"
|
||
assert payload["snapshot"]["investor_type"] == "C3"
|
||
# 载荷必须可 JSON 序列化（outbox 是 JSON 列）
|
||
json.dumps(payload, ensure_ascii=False)
|