From 852bafb37427cb135f70d0fdef8c69ac2f3d3a47 Mon Sep 17 00:00:00 2001 From: Windows Date: Fri, 11 Sep 2026 16:42:31 +0800 Subject: [PATCH] feat: add profile drift governance workflow --- app/api/controllers/admin.py | 27 +++ app/api/controllers/asset_allocation.py | 4 +- app/api/controllers/portfolio_analysis.py | 4 +- app/api/controllers/recommendations.py | 4 +- app/core/profile_governance_contracts.py | 12 ++ app/service/asset_allocation_service.py | 15 +- app/service/portfolio_analysis_service.py | 9 +- app/service/product_recommendation_service.py | 9 +- app/service/profile_governance_service.py | 199 ++++++++++++++++++ docs/21-投顾Agent迁移TODO.md | 59 +++--- .../test_profile_governance_service.py | 125 +++++++++++ 11 files changed, 436 insertions(+), 31 deletions(-) create mode 100644 app/core/profile_governance_contracts.py create mode 100644 app/service/profile_governance_service.py create mode 100644 tests/unit/service/test_profile_governance_service.py diff --git a/app/api/controllers/admin.py b/app/api/controllers/admin.py index 9a0498e..a5235ec 100644 --- a/app/api/controllers/admin.py +++ b/app/api/controllers/admin.py @@ -19,8 +19,10 @@ from app.api.schemas.admin import ( ) from app.core.advisor_backtest_contracts import AllocationBacktestQuery from app.core.contracts import RequestContext +from app.core.profile_governance_contracts import ProfileDriftReviewRequest from app.service.admin_service import AdminService from app.service.allocation_backtest_service import AllocationBacktestService +from app.service.profile_governance_service import ProfileGovernanceService router = APIRouter( prefix="/api/v1/admin", tags=["platform-admin"], dependencies=[Depends(enforce_rate_limit)] @@ -36,6 +38,31 @@ async def run_asset_allocation_backtest( return await AllocationBacktestService().run(payload, context, key) +@router.get("/advisor/profile-tags") +async def list_profile_tags( + customer_id: int = Query(gt=0), + context: RequestContext = Depends(build_request_context), # noqa: B008 +) -> dict[str, object]: + return await ProfileGovernanceService().tags(customer_id, context) + + +@router.get("/advisor/profile-drift-reviews") +async def list_profile_drift_reviews( + context: RequestContext = Depends(build_request_context), # noqa: B008 +) -> dict[str, object]: + return await ProfileGovernanceService().pending_reviews(context) + + +@router.post("/advisor/profile-drift-reviews/{review_id}/reviews") +async def review_profile_drift( + payload: ProfileDriftReviewRequest, + review_id: int = Path(gt=0), + context: RequestContext = Depends(build_request_context), # noqa: B008 + key: str | None = Header(default=None, alias="Idempotency-Key"), +) -> dict[str, object]: + return await ProfileGovernanceService().review(review_id, payload, context, key) + + def register_resource( resource: str, schema: type[BaseModel], diff --git a/app/api/controllers/asset_allocation.py b/app/api/controllers/asset_allocation.py index 642b620..f6ef79d 100644 --- a/app/api/controllers/asset_allocation.py +++ b/app/api/controllers/asset_allocation.py @@ -19,4 +19,6 @@ async def generate_asset_allocation( payload: AssetAllocationQuery, context: RequestContext = Depends(build_request_context), # noqa: B008 ) -> dict[str, object]: - return await AssetAllocationService().generate_for_agent(payload, context) + return await AssetAllocationService(enforce_profile_governance=True).generate_for_agent( + payload, context + ) diff --git a/app/api/controllers/portfolio_analysis.py b/app/api/controllers/portfolio_analysis.py index c3f41be..579c876 100644 --- a/app/api/controllers/portfolio_analysis.py +++ b/app/api/controllers/portfolio_analysis.py @@ -19,4 +19,6 @@ async def analyze_portfolio( payload: PortfolioAnalysisQuery, context: RequestContext = Depends(build_request_context), # noqa: B008 ) -> dict[str, object]: - return await PortfolioAnalysisService().analyze_for_agent(payload, context) + return await PortfolioAnalysisService(enforce_profile_governance=True).analyze_for_agent( + payload, context + ) diff --git a/app/api/controllers/recommendations.py b/app/api/controllers/recommendations.py index b118a33..804ce01 100644 --- a/app/api/controllers/recommendations.py +++ b/app/api/controllers/recommendations.py @@ -28,7 +28,9 @@ async def generate_recommendation( context: RequestContext = Depends(build_request_context), # noqa: B008 key: str | None = Header(default=None, alias="Idempotency-Key"), ) -> dict[str, object]: - return await ProductRecommendationService().generate(payload, context, key) + return await ProductRecommendationService(enforce_profile_governance=True).generate( + payload, context, key + ) @advisor_router.get("/recommendations/published") diff --git a/app/core/profile_governance_contracts.py b/app/core/profile_governance_contracts.py new file mode 100644 index 0000000..ab1f47a --- /dev/null +++ b/app/core/profile_governance_contracts.py @@ -0,0 +1,12 @@ +"""Contracts for internal customer-profile drift review.""" + +from typing import Literal + +from pydantic import BaseModel, ConfigDict, Field + + +class ProfileDriftReviewRequest(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + decision: Literal["approved", "rejected"] + comment: str = Field(default="", max_length=1000) diff --git a/app/service/asset_allocation_service.py b/app/service/asset_allocation_service.py index 9ef381c..b4c63b6 100644 --- a/app/service/asset_allocation_service.py +++ b/app/service/asset_allocation_service.py @@ -19,6 +19,7 @@ from app.service.dynamic_allocation_optimizer import ( ) from app.service.investment_goal_service import InvestmentGoalService from app.service.product_governance_monitor_service import SALES_INSTITUTION +from app.service.profile_governance_service import ProfileGovernanceService from app.service.suitability_service import SuitabilityService BASE_ALLOCATIONS = { @@ -36,13 +37,21 @@ ASSET_LABELS = { class AssetAllocationService: - def __init__(self, *, session_factory: Callable[[], Any] = SessionFactory) -> None: + def __init__( + self, + *, + session_factory: Callable[[], Any] = SessionFactory, + enforce_profile_governance: bool = False, + ) -> None: self.session_factory = session_factory + self.enforce_profile_governance = enforce_profile_governance async def generate_for_agent( self, _arguments: AssetAllocationQuery, context: RequestContext ) -> dict[str, object]: await AuthorizationService.require(context, "asset-allocation:generate:self") + if self.enforce_profile_governance: + await ProfileGovernanceService().require_operable(int(context.user_id)) authority = await SuitabilityService().authority_for_customer(int(context.user_id)) if authority.customer_risk_level is None: return {"status": "profile_required"} @@ -180,4 +189,6 @@ class AssetAllocationService: async def asset_allocation_tool( arguments: AssetAllocationQuery, context: RequestContext ) -> dict[str, object]: - return await AssetAllocationService().generate_for_agent(arguments, context) + return await AssetAllocationService(enforce_profile_governance=True).generate_for_agent( + arguments, context + ) diff --git a/app/service/portfolio_analysis_service.py b/app/service/portfolio_analysis_service.py index 8c87887..378dc3f 100644 --- a/app/service/portfolio_analysis_service.py +++ b/app/service/portfolio_analysis_service.py @@ -15,6 +15,7 @@ from app.model.advisor_product import AdvisorProductIndustryExposure, AdvisorPro from app.model.fund import FundHolding, FundProduct from app.repository.portfolio_analysis_repository import PortfolioAnalysisRepository from app.service.authorization_service import AuthorizationService +from app.service.profile_governance_service import ProfileGovernanceService from app.service.relationship_service import RelationshipService HUNDRED = Decimal("100") @@ -26,16 +27,20 @@ class PortfolioAnalysisService: *, session_factory: Callable[[], Any] = SessionFactory, graph_service: RelationshipService | None = None, + enforce_profile_governance: bool = False, ) -> None: self.session_factory = session_factory self.graph_service = graph_service or RelationshipService( Neo4jGraphDriver(get_settings()) ) + self.enforce_profile_governance = enforce_profile_governance async def analyze_for_agent( self, _arguments: PortfolioAnalysisQuery, context: RequestContext ) -> dict[str, object]: await AuthorizationService.require(context, "portfolio-analysis:read:self") + if self.enforce_profile_governance: + await ProfileGovernanceService().require_operable(int(context.user_id)) async with self.session_factory() as session: repository = PortfolioAnalysisRepository(session) positions = await repository.positions(int(context.user_id)) @@ -210,4 +215,6 @@ class PortfolioAnalysisService: async def portfolio_analysis_tool( arguments: PortfolioAnalysisQuery, context: RequestContext ) -> dict[str, object]: - return await PortfolioAnalysisService().analyze_for_agent(arguments, context) + return await PortfolioAnalysisService(enforce_profile_governance=True).analyze_for_agent( + arguments, context + ) diff --git a/app/service/product_recommendation_service.py b/app/service/product_recommendation_service.py index 2c33d8f..41e9a3a 100644 --- a/app/service/product_recommendation_service.py +++ b/app/service/product_recommendation_service.py @@ -22,6 +22,7 @@ from app.service.api_transaction_service import ApiTransactionService from app.service.authorization_service import AuthorizationService from app.service.investment_goal_service import InvestmentGoalService from app.service.product_governance_monitor_service import SALES_INSTITUTION +from app.service.profile_governance_service import ProfileGovernanceService from app.service.relationship_service import RelationshipService from app.service.suitability_service import SuitabilityService @@ -34,16 +35,20 @@ class ProductRecommendationService: *, session_factory: Callable[[], Any] = SessionFactory, relationship_service: RelationshipService | None = None, + enforce_profile_governance: bool = False, ) -> None: self.session_factory = session_factory self.relationship_service = relationship_service or RelationshipService( Neo4jGraphDriver(get_settings()) ) + self.enforce_profile_governance = enforce_profile_governance async def generate( self, payload: ProductRecommendationQuery, context: RequestContext, key: str | None ) -> dict[str, object]: await AuthorizationService.require(context, "product-recommendation:generate:self") + if self.enforce_profile_governance: + await ProfileGovernanceService().require_operable(int(context.user_id)) authority = await SuitabilityService().authority_for_customer(int(context.user_id)) if authority.customer_risk_level is None: return {"status": "profile_required"} @@ -342,4 +347,6 @@ class ProductRecommendationService: async def product_recommendation_tool( arguments: ProductRecommendationQuery, context: RequestContext ) -> dict[str, object]: - return await ProductRecommendationService().generate(arguments, context, None) + return await ProductRecommendationService(enforce_profile_governance=True).generate( + arguments, context, None + ) diff --git a/app/service/profile_governance_service.py b/app/service/profile_governance_service.py new file mode 100644 index 0000000..7f9c75a --- /dev/null +++ b/app/service/profile_governance_service.py @@ -0,0 +1,199 @@ +"""Profile-tag drift review and advisory-operation gating.""" + +from datetime import UTC, datetime +from typing import Any, Protocol +from uuid import uuid4 + +from sqlalchemy import update +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.contracts import RequestContext +from app.core.errors import GenericResourceNotFoundError, InvalidStateError +from app.core.profile_governance_contracts import ProfileDriftReviewRequest +from app.infrastructure.db import SessionFactory +from app.model.audit import InteractionAudit +from app.model.memory import MemorySyncOutbox +from app.model.profile_tag import AdvisorProfileDriftReview, AdvisorProfileTag +from app.model.risk_questionnaire import ProfileSnapshot +from app.repository.risk_questionnaire_repository import RiskQuestionnaireRepository +from app.service.api_transaction_service import ApiTransactionService, digest +from app.service.authorization_service import AuthorizationService + + +class SessionFactoryLike(Protocol): + def __call__(self) -> AsyncSession: ... + + +class ProfileGovernanceService: + """Internal-only profile governance; no customer-facing projection is returned.""" + + def __init__(self, session_factory: SessionFactoryLike = SessionFactory) -> None: + self.session_factory = session_factory + + async def require_operable(self, customer_id: int) -> None: + """Block advisory decisions while a changed profile is awaiting review.""" + async with self.session_factory() as session: + pending = await RiskQuestionnaireRepository(session).pending_drift_review(customer_id) + if pending is not None: + raise InvalidStateError("画像标签漂移正在复核，暂不能执行投顾分析") + + async def tags(self, customer_id: int, context: RequestContext) -> dict[str, object]: + await AuthorizationService.require(context, "profile-governance:read", admin=True) + async with self.session_factory() as session: + rows = await RiskQuestionnaireRepository(session).tags(customer_id) + return { + "data": [self._tag_view(row) for row in rows], + "meta": {"trace_id": context.trace_id}, + } + + async def pending_reviews(self, context: RequestContext) -> dict[str, object]: + await AuthorizationService.require(context, "profile-governance:read", admin=True) + async with self.session_factory() as session: + rows = await RiskQuestionnaireRepository(session).pending_reviews(limit=100) + return { + "data": [self._review_view(row) for row in rows], + "meta": {"trace_id": context.trace_id}, + } + + async def review( + self, + review_id: int, + payload: ProfileDriftReviewRequest, + context: RequestContext, + key: str | None, + ) -> dict[str, object]: + await AuthorizationService.require(context, "profile-governance:review", admin=True) + + async def operation(session: AsyncSession) -> dict[str, Any]: + repository = RiskQuestionnaireRepository(session) + review = await repository.drift_review(review_id, lock=True) + if review is None: + raise GenericResourceNotFoundError("画像漂移复核记录不存在") + if review.status != "pending_review": + raise InvalidStateError("画像漂移复核记录当前不能审核") + now = datetime.now(UTC).replace(tzinfo=None) + tags = await repository.tags_for_review(review_id, lock=True) + if payload.decision == "approved": + await self._approve(repository, review, tags, now) + else: + await session.execute( + update(AdvisorProfileTag) + .where(AdvisorProfileTag.drift_review_id == review_id) + .values(status="rejected", active_customer_tag=None, updated_at=now) + ) + review.status = payload.decision + review.reviewer_user_id = int(context.user_id) + review.reviewed_at = now + review.review_comment = payload.comment + review.updated_at = now + session.add(InteractionAudit( + actor_type="user", actor_id=int(context.user_id), + target_customer_id=review.customer_id, portal="admin", + action_type="advisor.profile_drift_reviewed", + detail={"review_id": review_id, "decision": payload.decision, + "trace_id": context.trace_id}, created_at=now, + )) + await session.flush() + return { + "data": {"review_id": str(review.id), "status": review.status}, + "meta": {"trace_id": context.trace_id}, + } + + return await ApiTransactionService().execute( + context, f"advisor:profile-drift-review:{review_id}", key, + payload.model_dump(mode="json"), operation, + ) + + async def _approve( + self, + repository: RiskQuestionnaireRepository, + review: AdvisorProfileDriftReview, + tags: list[AdvisorProfileTag], + now: datetime, + ) -> None: + await repository.deactivate_current_profile(review.customer_id, now) + active = await repository.active_tags(review.customer_id, lock=True) + await repository.supersede_active_tags( + review.customer_id, tuple(tag.tag_key for tag in active), now + ) + repository.add_profile(ProfileSnapshot( + profile_uuid=review.candidate_profile_uuid, + customer_id=review.customer_id, + version=review.candidate_profile_version, + snapshot=review.candidate_snapshot, + generation_basis=review.candidate_generation_basis, + snapshot_hash=digest(review.candidate_snapshot), + is_current=True, + generated_at=now, + created_at=now, + updated_at=now, + )) + tag_keys = {tag.tag_key for tag in tags} + if tag_keys: + await repository.session.execute( + update(AdvisorProfileTag) + .where( + AdvisorProfileTag.drift_review_id == review.id, + AdvisorProfileTag.status == "pending_review", + ) + .values( + status="active", + active_customer_tag=( + # A single SQL value cannot vary by tag; update individually below. + None + ), + updated_at=now, + ) + ) + for tag in tags: + tag.status = "active" + tag.active_customer_tag = f"{review.customer_id}:{tag.tag_key}" + tag.updated_at = now + self._add_sync_events(repository, review, now) + + @staticmethod + def _add_sync_events( + repository: RiskQuestionnaireRepository, + review: AdvisorProfileDriftReview, + now: datetime, + ) -> None: + # The same event UUID across targets is intentional; the baseline unique key is + # (event_uuid, target_store), so both projections share one logical change. + event_uuid = str(uuid4()) + payload = { + "customer_id": str(review.customer_id), + "profile_uuid": review.candidate_profile_uuid, + "version": review.candidate_profile_version, + "profile": review.candidate_snapshot, + } + for target_store in ("milvus", "neo4j"): + repository.add_sync_event(MemorySyncOutbox( + event_uuid=event_uuid, aggregate_type="profile", + aggregate_uuid=review.candidate_profile_uuid, + aggregate_version=review.candidate_profile_version, + target_store=target_store, operation="upsert", payload=payload, + status="pending", retry_count=0, next_retry_at=None, last_error=None, + created_at=now, processed_at=None, + )) + + @staticmethod + def _tag_view(row: AdvisorProfileTag) -> dict[str, object]: + return { + "tag_id": str(row.id), "customer_id": str(row.customer_id), + "tag_key": row.tag_key, "tag_value": row.tag_value, + "confidence": str(row.confidence), "source_type": row.source_type, + "source_reference": row.source_reference, + "source_confidence": str(row.source_confidence), + "profile_version": row.profile_version, "status": row.status, + "drift_review_id": str(row.drift_review_id) if row.drift_review_id else None, + } + + @staticmethod + def _review_view(row: AdvisorProfileDriftReview) -> dict[str, object]: + return { + "review_id": str(row.id), "drift_no": row.drift_no, + "customer_id": str(row.customer_id), + "candidate_profile_version": row.candidate_profile_version, + "changed_tags": row.changed_tags, "status": row.status, + "created_at": row.created_at.isoformat() + "Z", + } diff --git a/docs/21-投顾Agent迁移TODO.md b/docs/21-投顾Agent迁移TODO.md index 4b038c1..f9c3a0f 100644 --- a/docs/21-投顾Agent迁移TODO.md +++ b/docs/21-投顾Agent迁移TODO.md @@ -124,6 +124,17 @@ Ruff 通过，MyPy（138 个源文件）通过。阶段十提交：`0d42779`； 问题（`chk_config_release_separation` 未按新底座迁移撤下、测试账号外键缺失，以及 UTC 测试依赖的数据库状态），不是本阶段代码回归；独立迁移库和端到端会话验收仍待执行。 +阶段十二已完成画像标签治理闭环：开户问卷继续生成带置信度、来源类型、来源引用和画像版本 +的标签；标签值、来源变化和置信度明显下降会生成待复核候选画像。新增画像治理 Service 和 +管理员接口，可查看标签证据、查看待复核队列并审核通过/拒绝。审核通过在同一事务中切换当前 +画像、激活候选标签并写入 Milvus/Neo4j 两条 `memory_sync_outbox` 事件；审核拒绝保留旧画像。 +推荐、资产配置和持仓分析的 API 与 Agent 工具均接入漂移复核暂停闸门，客户接口不暴露内部 +标签、置信度、来源和审核信息。当前项目没有独立的调仓模拟入口，该项未虚报完成。 + +阶段十二测试结果：专项 `12 passed`；单元/契约 `499 passed, 3 warnings`；Ruff 通过；MyPy +（142 个源文件）通过；OpenAPI 已确认新增 3 个管理员画像治理路由。数据库结构未新增迁移， +复用阶段四已建立的画像标签、漂移复核和 Outbox 表；真实库端到端审核验收待独立迁移库执行。 + ## 一、迁移准备 - [ ] 确认远程仓库可访问。（当前失败：连接 `47.106.207.27:3000` 被拒绝） @@ -404,31 +415,31 @@ python tools/audit_constraints.py ## 十二、画像标签和漂移复核 -- [ ] 迁移画像标签模型。 -- [ ] 迁移标签值保存。 -- [ ] 迁移标签置信度保存。 -- [ ] 迁移来源类型保存。 -- [ ] 迁移来源引用保存。 -- [ ] 迁移来源置信度保存。 -- [ ] 迁移画像版本保存。 -- [ ] 迁移标签生效状态。 -- [ ] 迁移标签值变化检测。 -- [ ] 迁移来源变化检测。 -- [ ] 迁移置信度下降检测。 -- [ ] 迁移候选画像保存。 -- [ ] 迁移漂移复核队列。 -- [ ] 迁移管理员查看标签证据接口。 -- [ ] 迁移管理员审核通过接口。 -- [ ] 迁移管理员审核拒绝接口。 -- [ ] 审核通过后切换当前画像。 -- [ ] 审核通过后创建 Milvus 同步事件。 -- [ ] 审核通过后创建 Neo4j 同步事件。 -- [ ] 审核期间暂停产品推荐。 -- [ ] 审核期间暂停资产配置。 -- [ ] 审核期间暂停持仓分析。 +- [x] 迁移画像标签模型。（复用新增 `advisor_profile_tag`） +- [x] 迁移标签值保存。 +- [x] 迁移标签置信度保存。 +- [x] 迁移来源类型保存。 +- [x] 迁移来源引用保存。 +- [x] 迁移来源置信度保存。 +- [x] 迁移画像版本保存。 +- [x] 迁移标签生效状态。 +- [x] 迁移标签值变化检测。 +- [x] 迁移来源变化检测。 +- [x] 迁移置信度下降检测。 +- [x] 迁移候选画像保存。（复核记录保存候选快照） +- [x] 迁移漂移复核队列。 +- [x] 迁移管理员查看标签证据接口。 +- [x] 迁移管理员审核通过接口。 +- [x] 迁移管理员审核拒绝接口。 +- [x] 审核通过后切换当前画像。 +- [x] 审核通过后创建 Milvus 同步事件。 +- [x] 审核通过后创建 Neo4j 同步事件。 +- [x] 审核期间暂停产品推荐。 +- [x] 审核期间暂停资产配置。 +- [x] 审核期间暂停持仓分析。 - [ ] 审核期间暂停调仓模拟。 -- [ ] 确认客户不能读取内部标签、置信度和审核信息。 -- [ ] 完成画像治理提交 `advisor/profile-governance`。 +- [x] 确认客户不能读取内部标签、置信度和审核信息。（无客户侧标签路由） +- [x] 完成画像治理提交 `advisor/profile-governance`。（真实库端到端审核验收待完成） ## 十三、全端测试 diff --git a/tests/unit/service/test_profile_governance_service.py b/tests/unit/service/test_profile_governance_service.py new file mode 100644 index 0000000..2e83c57 --- /dev/null +++ b/tests/unit/service/test_profile_governance_service.py @@ -0,0 +1,125 @@ +from datetime import datetime +from decimal import Decimal + +import pytest + +from app.core.errors import InvalidStateError +from app.model.profile_tag import AdvisorProfileDriftReview, AdvisorProfileTag +from app.service.profile_governance_service import ProfileGovernanceService + + +class _AsyncContext: + async def __aenter__(self) -> object: + return None + + async def __aexit__(self, *_args: object) -> bool: + return False + + +class _Session: + def __init__(self, pending: object = None) -> None: + self.pending = pending + + def begin(self) -> _AsyncContext: + return _AsyncContext() + + +class _SessionFactory: + def __init__(self, session: _Session) -> None: + self.session = session + + def __call__(self) -> "_SessionContext": + return _SessionContext(self.session) + + +class _SessionContext: + def __init__(self, session: _Session) -> None: + self.session = session + + async def __aenter__(self) -> _Session: + return self.session + + async def __aexit__(self, *_args: object) -> bool: + return False + + +@pytest.mark.asyncio +async def test_pending_profile_blocks_advisory_operation(monkeypatch) -> None: + session = _Session(pending=object()) + + async def pending(_self, _customer_id: int, *, lock: bool = False) -> object: + del lock + return session.pending + + monkeypatch.setattr( + "app.service.profile_governance_service.RiskQuestionnaireRepository.pending_drift_review", + pending, + ) + + with pytest.raises(InvalidStateError, match="漂移正在复核"): + await ProfileGovernanceService(_SessionFactory(session)).require_operable(7) + + +class _Repository: + def __init__(self) -> None: + self.session = self + self.added: list[object] = [] + self.sync_targets: list[str] = [] + + async def execute(self, _statement: object) -> None: + return None + + async def deactivate_current_profile(self, _customer_id: int, _now: datetime) -> None: + return None + + async def active_tags( + self, _customer_id: int, *, lock: bool = False + ) -> list[AdvisorProfileTag]: + del lock + return [] + + async def supersede_active_tags( + self, _customer_id: int, _tag_keys: tuple[str, ...], _now: datetime + ) -> None: + return None + + def add_profile(self, profile: object) -> None: + self.added.append(profile) + + def add_sync_event(self, event: object) -> None: + self.added.append(event) + self.sync_targets.append(event.target_store) + + +def _review() -> AdvisorProfileDriftReview: + now = datetime(2026, 9, 11) + return AdvisorProfileDriftReview( + id=4, drift_no="drift-4", customer_id=7, source_assessment_id=9, + candidate_profile_uuid="candidate-uuid", candidate_profile_version=2, + candidate_snapshot={"risk_level": "C3"}, + candidate_generation_basis={"source": "assessment"}, + changed_tags=[], status="pending_review", reviewer_user_id=None, + reviewed_at=None, review_comment=None, created_at=now, updated_at=now, + ) + + +@pytest.mark.asyncio +async def test_approved_review_activates_candidate_and_enqueues_both_projections() -> None: + repository = _Repository() + tag = AdvisorProfileTag( + id=11, tag_uuid="tag-11", customer_id=7, tag_key="risk_level", tag_value="C3", + tag_value_hash="x", confidence=Decimal("0.9000"), source_type="assessment", + source_reference="assessment:9", source_confidence=Decimal("1.0000"), + profile_version=2, drift_review_id=4, previous_tag_id=1, drift_reason="value_changed", + status="pending_review", active_customer_tag=None, + created_at=datetime(2026, 9, 11), updated_at=datetime(2026, 9, 11), + ) + + await ProfileGovernanceService()._approve( + repository, _review(), [tag], datetime(2026, 9, 11) + ) + + assert tag.status == "active" + assert tag.active_customer_tag == "7:risk_level" + assert repository.added[0].is_current is True + assert repository.sync_targets == ["milvus", "neo4j"]