feat: add profile candidate review workflow

2026-09-11 17:47:06 +08:00
parent 3a21afa847
commit a769130658
7 changed files with 437 additions and 14 deletions
@@ -0,0 +1,180 @@
+"""客户画像候选的确认、审核与晋升服务。
+
+候选与正式记忆共用 ``memory_unit``，但状态转换必须经过本服务；客服 Agent 不具备
+调用权限。用户确认只把候选标记为 ``verified``，管理员批准后才切换为 ``active``。
+"""
+
+from datetime import UTC, datetime
+from typing import Any, Literal
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+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.service.agent.bootstrap import get_memory_cache_adapter
+from app.service.authorization_service import AuthorizationService
+from app.service.memory_service import MemoryService
+
+CandidateDecision = Literal["confirmed", "rejected"]
+ReviewDecision = Literal["approved", "rejected"]
+USER_CONFIRM_PERMISSION = "memory:candidate:confirm"
+ADMIN_REVIEW_PERMISSION = "memory:candidate:review"
+
+
+class CustomerProfileCandidateService:
+    """候选状态机的唯一应用服务入口。"""
+
+    async def list_for_customer(
+        self, context: RequestContext, *, limit: int = 20
+    ) -> dict[str, Any]:
+        """返回当前登录用户自己的候选，不暴露证据原文或其他客户数据。"""
+        await AuthorizationService.require(context, "memory:read:self")
+        customer_id = int(context.user_id)
+        async with SessionFactory() as session:
+            rows = await session.scalars(
+                select(MemoryUnit)
+                .where(
+                    MemoryUnit.customer_id == customer_id,
+                    MemoryUnit.status.in_(("candidate", "verified")),
+                )
+                .order_by(MemoryUnit.updated_at.desc())
+                .limit(max(1, min(limit, 100)))
+            )
+            data = [self._view(item) for item in rows]
+        return {"data": data, "meta": {"trace_id": context.trace_id}}
+
+    async def list_for_admin(
+        self, context: RequestContext, *, limit: int = 20
+    ) -> dict[str, Any]:
+        """管理员查看所有待处理候选；列表仍不返回证据原文。"""
+        await AuthorizationService.require(context, ADMIN_REVIEW_PERMISSION, admin=True)
+        async with SessionFactory() as session:
+            rows = await session.scalars(
+                select(MemoryUnit)
+                .where(MemoryUnit.status.in_(("candidate", "verified")))
+                .order_by(MemoryUnit.updated_at.asc())
+                .limit(max(1, min(limit, 100)))
+            )
+            data = [self._view(item) for item in rows]
+        return {"data": data, "meta": {"trace_id": context.trace_id}}
+
+    async def decide_by_customer(
+        self, candidate_id: int, decision: CandidateDecision, context: RequestContext
+    ) -> dict[str, Any]:
+        """用户确认或拒绝自己的候选；确认不会直接激活正式记忆。"""
+        await AuthorizationService.require(context, USER_CONFIRM_PERMISSION)
+        customer_id = int(context.user_id)
+        target_status = "verified" if decision == "confirmed" else "rejected"
+        async with SessionFactory() as session, session.begin():
+            candidate = await self._locked_candidate(session, candidate_id, customer_id)
+            if candidate.status != "candidate":
+                raise InvalidStateError("候选已处理，不能重复确认")
+            candidate.status = target_status
+            candidate.updated_at = self._now()
+            session.add(self._audit(
+                context, customer_id, "memory.candidate_user_decision",
+                {"candidate_id": candidate_id, "decision": decision},
+            ))
+            await session.flush()
+            return {"data": self._view(candidate), "meta": {"trace_id": context.trace_id}}
+
+    async def review_by_admin(
+        self, candidate_id: int, decision: ReviewDecision, context: RequestContext,
+        *, comment: str = "",
+    ) -> dict[str, Any]:
+        """管理员审核候选；批准时处理同键正式记忆冲突并激活候选。"""
+        await AuthorizationService.require(context, ADMIN_REVIEW_PERMISSION, admin=True)
+        async with SessionFactory() as session, session.begin():
+            candidate = await self._locked_candidate(session, candidate_id, None)
+            if candidate.status not in {"candidate", "verified"}:
+                raise InvalidStateError("候选已处理，不能重复审核")
+            if decision == "rejected":
+                candidate.status = "rejected"
+            else:
+                await self._promote(session, candidate, int(context.user_id))
+            candidate.updated_at = self._now()
+            session.add(self._audit(
+                context, candidate.customer_id, "memory.candidate_admin_review",
+                {"candidate_id": candidate_id, "decision": decision, "comment": comment[:1000]},
+            ))
+            await session.flush()
+            return {"data": self._view(candidate), "meta": {"trace_id": context.trace_id}}
+
+    async def _locked_candidate(
+        self, session: AsyncSession, candidate_id: int, customer_id: int | None
+    ) -> MemoryUnit:
+        """按身份范围加锁读取候选，找不到时统一隐藏资源存在性。"""
+        conditions = [MemoryUnit.id == candidate_id]
+        if customer_id is not None:
+            conditions.append(MemoryUnit.customer_id == customer_id)
+        candidate = await session.scalar(select(MemoryUnit).where(*conditions).with_for_update())
+        if candidate is None:
+            raise GenericResourceNotFoundError("候选不存在")
+        return candidate
+
+    async def _promote(
+        self, session: AsyncSession, candidate: MemoryUnit, reviewer_id: int
+    ) -> None:
+        """同一客户同一键只保留一条 active，旧值失效并留下冲突审计记录。"""
+        current = await session.scalar(
+            select(MemoryUnit)
+            .where(
+                MemoryUnit.customer_id == candidate.customer_id,
+                MemoryUnit.memory_key == candidate.memory_key,
+                MemoryUnit.status == "active",
+            )
+            .with_for_update()
+        )
+        now = self._now()
+        if current is not None and current.id != candidate.id:
+            current.status = "invalidated"
+            current.updated_at = now
+            session.add(MemoryConflict(
+                left_memory_id=current.id, right_memory_id=candidate.id,
+                conflict_type="candidate_promoted", severity="medium", status="resolved",
+                resolution="管理员审核候选后替换旧正式记忆", winner_memory_id=candidate.id,
+                resolved_by=reviewer_id, resolved_at=now, created_at=now,
+            ))
+        candidate.status = "active"
+        candidate.promoted_at = now
+        candidate.version += 1
+        await MemoryService(session, cache=get_memory_cache_adapter()).invalidate_recall_cache(
+            int(candidate.customer_id)
+        )
+
+    @staticmethod
+    def _view(item: MemoryUnit) -> dict[str, Any]:
+        """只返回结构化候选值，不返回对话证据摘录。"""
+        return {
+            "candidate_id": int(item.id),
+            "customer_id": str(item.customer_id),
+            "memory_key": item.memory_key,
+            "value": item.content,
+            "memory_type": item.memory_type,
+            "confidence": float(item.confidence),
+            "status": item.status,
+            "version": item.version,
+            "created_at": item.created_at.isoformat(),
+            "updated_at": item.updated_at.isoformat(),
+        }
+
+    @staticmethod
+    def _audit(
+        context: RequestContext, customer_id: int, action_type: str, detail: dict[str, Any]
+    ) -> InteractionAudit:
+        """统一生成候选状态变更审计，不携带证据原文。"""
+        return InteractionAudit(
+            actor_type="user", actor_id=int(context.user_id), target_customer_id=customer_id,
+            session_id=None, portal=context.portal, action_type=action_type,
+            detail={**detail, "trace_id": context.trace_id},
+            created_at=CustomerProfileCandidateService._now(),
+        )
+
+    @staticmethod
+    def _now() -> datetime:
+        """使用 UTC 无时区值，与现有数据库 DATETIME 字段保持一致。"""
+        return datetime.now(UTC).replace(tzinfo=None)