feat: generate approved profile snapshots

app/model/memory.py

@@ -75,6 +75,25 @@ class MemorySyncOutbox(Base):
     processed_at: Mapped[datetime | None] = mapped_column(DateTime)
 
 
+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[int] = mapped_column(Integer, nullable=False, default=0)
+    current_customer_id: Mapped[int | None] = mapped_column(BigInteger, unique=True)
+    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)
+
+
 class MemoryConflict(Base):
     """记忆冲突；库列名为 `status`，`resolution`/`severity`/`resolved_by` 必须映射。"""
 

app/service/customer_profile_candidate_service.py

@@ -4,8 +4,11 @@
 调用权限。用户确认只把候选标记为 ``verified``，管理员批准后才切换为 ``active``。
 """
 
+import hashlib
+import json
 from datetime import UTC, datetime
 from typing import Any, Literal
+from uuid import uuid4
 
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -14,7 +17,7 @@ from app.core.contracts import RequestContext
 from app.core.errors import GenericResourceNotFoundError, InvalidStateError
 from app.infrastructure.db import SessionFactory
 from app.model.audit import InteractionAudit
-from app.model.memory import MemoryConflict, MemoryUnit
+from app.model.memory import MemoryConflict, MemorySyncOutbox, MemoryUnit, ProfileSnapshot
 from app.service.agent.bootstrap import get_memory_cache_adapter
 from app.service.authorization_service import AuthorizationService
 from app.service.memory_service import MemoryService
@@ -142,10 +145,69 @@ class CustomerProfileCandidateService:
         candidate.status = "active"
         candidate.promoted_at = now
         candidate.version += 1
+        await self._write_profile_snapshot(session, candidate, reviewer_id, now)
         await MemoryService(session, cache=get_memory_cache_adapter()).invalidate_recall_cache(
             int(candidate.customer_id)
         )
 
+    async def _write_profile_snapshot(
+        self, session: AsyncSession, candidate: MemoryUnit, reviewer_id: int, now: datetime
+    ) -> None:
+        """生成新的当前画像版本，并为两个派生存储写入可靠同步事件。"""
+        current = await session.scalar(
+            select(ProfileSnapshot)
+            .where(
+                ProfileSnapshot.customer_id == candidate.customer_id,
+                ProfileSnapshot.current_customer_id == candidate.customer_id,
+            )
+            .with_for_update()
+        )
+        previous = dict(current.snapshot) if current is not None else {}
+        preferences = dict(previous.get("customer_service_preferences", {}))
+        preferences[candidate.memory_key] = {
+            "value": candidate.content,
+            "memory_type": candidate.memory_type,
+            "confidence": float(candidate.confidence),
+        }
+        snapshot = {**previous, "customer_service_preferences": preferences}
+        version = (int(current.version) + 1) if current is not None else 1
+        profile_uuid = str(uuid4())
+        snapshot_hash = hashlib.sha256(
+            json.dumps(snapshot, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+            .encode("utf-8")
+        ).hexdigest()
+        if current is not None:
+            current.is_current = 0
+            current.current_customer_id = None
+            current.updated_at = now
+        created = ProfileSnapshot(
+            profile_uuid=profile_uuid, customer_id=candidate.customer_id,
+            version=version, snapshot=snapshot,
+            generation_basis={
+                "source": "customer_profile_candidate",
+                "candidate_id": candidate.id,
+                "reviewer_id": reviewer_id,
+            },
+            snapshot_hash=snapshot_hash, is_current=1,
+            current_customer_id=candidate.customer_id, generated_at=now,
+            created_at=now, updated_at=now,
+        )
+        session.add(created)
+        await session.flush()
+        for target_store in ("milvus", "neo4j"):
+            session.add(MemorySyncOutbox(
+                event_uuid=str(uuid4()), aggregate_type="profile_snapshot",
+                aggregate_uuid=profile_uuid, aggregate_version=version,
+                target_store=target_store, operation="upsert",
+                payload={
+                    "customer_id": candidate.customer_id,
+                    "profile_uuid": profile_uuid,
+                    "profile_version": version,
+                    "snapshot": snapshot,
+                },
+                status="pending", retry_count=0, created_at=now,
+            ))
+
     @staticmethod
     def _view(item: MemoryUnit) -> dict[str, Any]:
         """只返回结构化候选值，不返回对话证据摘录。"""

docs/客服Agent二期_客户画像候选流程_v1.md

@@ -2,7 +2,7 @@
 
 版本：v1.0  
 适用分支：`ZSY_develop2`  
-状态：候选提取、用户确认和管理员审核入口已实现，画像快照同步待后续迭代
+状态：候选提取、用户确认、管理员审核、画像快照生成和投影事件写入已实现
 
 ## 1. 业务边界
 
@@ -25,7 +25,7 @@
   -> 用户确认或拒绝
   -> 管理员审核
   -> 批准后处理同键冲突并晋升为 active
-  -> 后续流程再生成 profile_snapshots
+  -> 生成 profile_snapshots 并写入 memory_sync_outbox
 ```
 
 普通公开问答、闲聊、一次性操作问题不触发候选抽取；访客、非客服 Agent、非 self 数据范围
@@ -89,10 +89,13 @@
 - `app/api/controllers/public_platform.py`：用户候选查询和确认接口。
 - `app/api/controllers/admin.py`：管理员候选查询和审核接口。
 
+管理员批准后，服务会创建新的 `profile_snapshots` 当前版本，并为 `milvus`、`neo4j` 各写入
+一条 `memory_sync_outbox` 待投影事件；旧当前版本在同一事务内标记为非当前。
+
 ## 7. 尚未实现的后续能力
 
 1. 前端用户确认页面和管理员审核页面。
-2. 候选晋升后的 `profile_snapshots` 版本生成和 `memory_sync_outbox` 同步。
+2. `memory_sync_outbox` 到真实 Milvus/Neo4j 的消费者联调和失败重试验收。
 3. 候选撤回、过期、删除和隐私授权管理。
 4. 候选流程的 MySQL 集成测试和管理员端到端验收。
 
@@ -101,6 +104,6 @@
 ## 8. 验证结果
 
 - 客服画像候选专项测试：通过。
-- 一期单元与契约回归：`683 passed`。
+- 一期单元与契约回归：`687 passed`。
 - Ruff：通过。
 - Mypy：通过。

tests/unit/service/test_customer_profile_candidate_service.py

@@ -7,7 +7,7 @@ import pytest
 
 from app.core.contracts import RequestContext
 from app.core.errors import ForbiddenAgentError, InvalidStateError
-from app.model.memory import MemoryUnit
+from app.model.memory import MemorySyncOutbox, MemoryUnit, ProfileSnapshot
 from app.service.customer_profile_candidate_service import (
     ADMIN_REVIEW_PERMISSION,
     USER_CONFIRM_PERMISSION,
@@ -142,3 +142,41 @@ async def test_candidate_transition_rejects_repeated_processing(
         await CustomerProfileCandidateService().decide_by_customer(
             11, "confirmed", RequestContext(user_id="7", trace_id="candidate-test")
         )
+
+
+@pytest.mark.asyncio
+async def test_profile_snapshot_and_projection_events_are_created_on_promotion(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """批准候选必须生成新画像版本和两个待投影事件。"""
+    class FakeSession:
+        def __init__(self) -> None:
+            self.added: list[object] = []
+
+        async def scalar(self, statement: object) -> None:
+            del statement
+            return None
+
+        def add(self, item: object) -> None:
+            self.added.append(item)
+
+        async def flush(self) -> None:
+            return None
+
+    session = FakeSession()
+    target = candidate()
+    monkeypatch.setattr(
+        "app.service.customer_profile_candidate_service.get_memory_cache_adapter",
+        lambda: None,
+    )
+
+    await CustomerProfileCandidateService()._promote(session, target, reviewer_id=9003)
+
+    assert target.status == "active"
+    snapshot = next(item for item in session.added if isinstance(item, ProfileSnapshot))
+    assert snapshot.is_current == 1
+    assert snapshot.version == 1
+    value = snapshot.snapshot["customer_service_preferences"]["preference:risk_level"]["value"]
+    assert value == "稳健型"
+    events = [item for item in session.added if isinstance(item, MemorySyncOutbox)]
+    assert {item.target_store for item in events} == {"milvus", "neo4j"}