diff --git a/app/service/customer_profile_candidate_service.py b/app/service/customer_profile_candidate_service.py index bb2758f..dcfc851 100644 --- a/app/service/customer_profile_candidate_service.py +++ b/app/service/customer_profile_candidate_service.py @@ -194,6 +194,24 @@ class CustomerProfileCandidateService: ) session.add(created) await session.flush() + active_memories = list(await session.scalars( + select(MemoryUnit).where( + MemoryUnit.customer_id == candidate.customer_id, + MemoryUnit.status == "active", + ) + )) + memory_sources = [ + { + "memory_uuid": item.memory_uuid, + "memory_key": item.memory_key, + "content": item.content, + "memory_type": item.memory_type, + "confidence": float(item.confidence), + "version": int(item.version), + "valid_until": item.valid_until.isoformat() if item.valid_until else None, + } + for item in active_memories + ] for target_store in ("milvus", "neo4j"): session.add(MemorySyncOutbox( event_uuid=str(uuid4()), aggregate_type="profile_snapshot", @@ -204,6 +222,7 @@ class CustomerProfileCandidateService: "profile_uuid": profile_uuid, "profile_version": version, "snapshot": snapshot, + "memory_sources": memory_sources, }, status="pending", retry_count=0, created_at=now, )) diff --git a/tests/unit/service/test_customer_profile_candidate_service.py b/tests/unit/service/test_customer_profile_candidate_service.py index c1c441f..d80852a 100644 --- a/tests/unit/service/test_customer_profile_candidate_service.py +++ b/tests/unit/service/test_customer_profile_candidate_service.py @@ -150,21 +150,26 @@ async def test_profile_snapshot_and_projection_events_are_created_on_promotion( ) -> None: """批准候选必须生成新画像版本和两个待投影事件。""" class FakeSession: - def __init__(self) -> None: + def __init__(self, memory_units: list[MemoryUnit]) -> None: self.added: list[object] = [] + self.memory_units = memory_units async def scalar(self, statement: object) -> None: del statement return None + async def scalars(self, statement: object) -> list[object]: + del statement + return self.memory_units + def add(self, item: object) -> None: self.added.append(item) async def flush(self) -> None: return None - session = FakeSession() target = candidate() + session = FakeSession([target]) monkeypatch.setattr( "app.service.customer_profile_candidate_service.get_memory_cache_adapter", lambda: None, @@ -180,3 +185,5 @@ async def test_profile_snapshot_and_projection_events_are_created_on_promotion( assert value == "稳健型" events = [item for item in session.added if isinstance(item, MemorySyncOutbox)] assert {item.target_store for item in events} == {"milvus", "neo4j"} + for event in events: + assert event.payload["memory_sources"][0]["memory_uuid"] == target.memory_uuid