feat: 装配投影删除客户端（记忆失效/销户清理图与向量），并修复画像字段残留

一、装配 projection_cleaner
新增 app/service/projection_cleanup_service.py，并在**组装层**（app/worker/__main__.py）
注入。此前该客户端一直未装配，清理链路只记录降级（skipped_no_client），实际后果是
**销户后图里仍留着偏好关系**——投顾仍能通过关系网络看到这个人。

清理策略是"以画像为准"而不是按标识直接删边：先删掉该记忆对应的长期事实，再重建画像，
最后用对账修复让图跟着画像收敛。这样即使一条记忆影响多条派生边也能删干净，不依赖
"记得它当初投影成了什么"。Milvus 侧按 memory_uuid 删除；集合不存在（语义召回未启用）
时视为无需清理，客户端不可用则如实报告未清理，绝不伪造成功。

放在组装层而不是 runtime 内部兜底：组件内部给默认实现会把"尚未装配"这一事实悄悄盖住，
而"未注入即显式降级并留痕"是 runtime 刻意保留的语义。最初的改法写成内部兜底，被 5 个
既有单测拦下——那些测试是对的，因此改为在入口处注入。

二、顺带修复：事实消失后画像字段残留
实测：让 preference:horizon 记忆失效并清理后，user_facts 与图边都正确清除，但
fin_customer_profile.investment_horizon 仍是"长期（5年以上）"。根因是 rebuild_profile
只写"本轮有新值"的字段；事实被删后该字段没有新值，旧值就留在画像里——于是记忆已经作废，
投顾还能看到一个客户从未授权继续生效的投资期限。

改为：本轮没有对应事实时**清空**该字段；investor_type 例外——问卷是唯一权威，本轮没有
问卷记录时保持原值（该列 NOT NULL，且清空会在重测空档抹掉开户时的等级）。

三、实测
· 让 preference:horizon 失效后清理：cleaned=True，
  detail=profile_rebuilt; graph_cleaned; vector_collection_absent；
· user_facts 2→1；图边由 ['HAS_GOAL','PREFERS'] 变为 ['PREFERS']；
· 画像 investment_horizon 由"长期（5年以上）"变为 None；investor_type 保持 C2、
  risk_tags 不受影响（与 horizon 无关）；
· 恢复记忆状态后重建，画像与图边均正确复原；
· ruff 通过、mypy 113 文件无错、unit+contract 447 passed、integration 29 passed。

app/service/profile_assembly_service.py

@@ -190,8 +190,16 @@ class ProfileAssemblyService:
                 "promoted": len(facts),
             }
         for field in PROFILE_OWNED_FIELDS:
-            if field in values:
-                setattr(profile, field, values[field])
+            if field == "investor_type":
+                # 问卷是唯一权威：本轮没有问卷记录时**保持原值**（既不写入也不清空）。
+                # 否则重测前的空档会把开户时的等级抹掉，而该列是 NOT NULL。
+                if field in values:
+                    setattr(profile, field, values[field])
+                continue
+            # 其余字段由本服务独占：本轮没有对应事实即清空。
+            # 这不只是洁癖——记忆失效后若不清理，画像会留着一个已经作废的投资期限，
+            # 投顾据此给建议，而客户从未授权这条信息继续生效（实测踩到过）。
+            setattr(profile, field, values.get(field))
         profile.updated_at = now
         await self.session.flush()
 

app/service/projection_cleanup_service.py

@@ -0,0 +1,136 @@
+"""投影清理：记忆失效或删除时，清掉它在图库与向量库里的派生数据。
+
+**为什么以画像为准，而不是按标识直接删边**：一条记忆可能对应多条派生边（同一 key 的
+证据累积会让 tag 取值变化），直接按 tag_key 删容易删不干净或误删。这里走
+「删掉该记忆对应的长期事实 → 重建画像 → 对账修复」，让投影跟着权威源收敛——
+最终图里剩下的内容必然与画像一致，而不依赖"我记得这条记忆当初投影成了什么"。
+
+**Milvus 向量**按 `memory_uuid` 删除；集合不存在或客户端不可用时**如实报告未清理**，
+绝不伪造成功。这与 `_cleanup_projection` 的审计语义一致：审计只记录适配器返回的真实结论。
+"""
+
+import logging
+from dataclasses import dataclass
+
+from sqlalchemy import delete, select
+
+from app.core.config import get_settings
+from app.infrastructure.db import SessionFactory
+from app.model.memory import MemoryUnit
+from app.model.profile import UserFact
+from app.service.profile_assembly_service import ProfileAssemblyService
+from app.service.profile_graph_projection_service import ProfileGraphProjectionService
+from app.service.relationship_service import RelationshipService
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True)
+class ProjectionCleanupOutcome:
+    """与 `app/worker/runtime.py` 的同名结构保持字段一致（该处只按属性取用）。
+
+    `cleaned=True` 表示派生数据已确实清理；为假时 `detail` 必须说明真实缺口，
+    审计会原样记录它。
+    """
+
+    cleaned: bool
+    detail: str = ""
+
+
+class ProjectionCleanupService:
+    def __init__(self, relationships: RelationshipService | None = None) -> None:
+        self.relationships = relationships
+
+    async def cleanup(self, *, memory_uuid: str, operation: str) -> ProjectionCleanupOutcome:
+        del operation  # 失效与删除对投影的处理一致：都以权威源为准重建
+        async with SessionFactory() as session:
+            memory = await session.scalar(
+                select(MemoryUnit).where(MemoryUnit.memory_uuid == memory_uuid)
+            )
+            if memory is None:
+                # 记忆已经不在库里：投影清理的目标已消失，视为完成（幂等）
+                return ProjectionCleanupOutcome(True, "memory_not_found")
+            customer_id = int(memory.customer_id)
+            fact_key = str(memory.memory_key)
+
+        # 该记忆已失效/删除，它对应的事实不应再参与画像
+        async with SessionFactory() as session, session.begin():
+            await session.execute(delete(UserFact).where(
+                UserFact.customer_id == customer_id, UserFact.fact_key == fact_key
+            ))
+
+        # 以剩余证据重建画像，再让图跟着画像收敛
+        details: list[str] = []
+        try:
+            async with SessionFactory() as session, session.begin():
+                await ProfileAssemblyService(session).rebuild(customer_id)
+            details.append("profile_rebuilt")
+        except Exception:
+            logger.warning("profile rebuild failed during cleanup customer_id=%s",
+                           customer_id, exc_info=True)
+            details.append("profile_rebuild_failed")
+
+        graph_ok = await self._cleanup_graph(customer_id, details)
+        vector_ok = await self._cleanup_vector(memory_uuid, details)
+        return ProjectionCleanupOutcome(
+            cleaned=graph_ok and vector_ok and "profile_rebuild_failed" not in details,
+            detail="; ".join(details),
+        )
+
+    async def _cleanup_graph(self, customer_id: int, details: list[str]) -> bool:
+        if self.relationships is None:
+            details.append("graph_client_unavailable")
+            return False
+        try:
+            async with SessionFactory() as session:
+                outcome = await ProfileGraphProjectionService(
+                    session, self.relationships
+                ).reconcile_customer(customer_id, repair=True)
+        except Exception:
+            logger.warning("graph cleanup failed customer_id=%s", customer_id, exc_info=True)
+            details.append("graph_cleanup_failed")
+            return False
+        if outcome.degraded:
+            details.append(f"graph_degraded:{outcome.reason}")
+            return False
+        if not outcome.consistent:
+            # repair 之后仍不一致：如实报告，不写成清理成功
+            details.append(
+                f"graph_still_inconsistent:missing={len(outcome.missing)},"
+                f"orphaned={len(outcome.orphaned)}"
+            )
+            return False
+        details.append("graph_cleaned")
+        return True
+
+    async def _cleanup_vector(self, memory_uuid: str, details: list[str]) -> bool:
+        """删除该记忆的向量。集合不存在（未启用语义召回）时视为无需清理。"""
+        settings = get_settings()
+        if not settings.milvus_uri:
+            details.append("vector_store_not_configured")
+            return True
+        try:
+            from pymilvus import MilvusClient  # type: ignore[import-untyped]
+
+            client = MilvusClient(
+                uri=settings.milvus_uri, token=settings.milvus_token or None
+            )
+        except Exception:
+            logger.warning("milvus client unavailable during cleanup", exc_info=True)
+            details.append("vector_client_unavailable")
+            return False
+        try:
+            if settings.milvus_collection not in set(client.list_collections()):
+                # 记忆向量集合尚未启用：没有需要清理的派生数据
+                details.append("vector_collection_absent")
+                return True
+            client.delete(
+                collection_name=settings.milvus_collection,
+                filter=f'memory_uuid == "{memory_uuid}"',
+            )
+            details.append("vector_cleaned")
+            return True
+        except Exception:
+            logger.warning("milvus delete failed memory_uuid=%s", memory_uuid, exc_info=True)
+            details.append("vector_delete_failed")
+            return False

app/worker/__main__.py

@@ -4,6 +4,8 @@ import logging
 
 from app.core.config import get_settings
 from app.infrastructure.db import engine
+from app.service.agent.bootstrap import get_relationship_service
+from app.service.projection_cleanup_service import ProjectionCleanupService
 from app.worker.runtime import WorkerRuntime
 
 logger = logging.getLogger(__name__)
@@ -11,7 +13,20 @@ logger = logging.getLogger(__name__)
 
 async def serve(*, once: bool = False) -> None:
     settings = get_settings()
-    runtime = WorkerRuntime(settings=settings)
+    relationships = get_relationship_service()
+    # 在**组装层**注入投影删除客户端：记忆失效/销户时清理图库与向量库里的派生数据。
+    # 放在这里而不是 runtime 内部兜底，是为了保留"未注入即显式降级并留痕"的语义
+    # （有单测守着这一点），也让"生产装配了什么"在入口处一眼可见。
+    # 该服务返回自己模块里的 ProjectionCleanupOutcome（字段与 runtime 的同名结构一致），
+    # 结构化契约成立但名义类型不同，故显式忽略：为此把结构体抽到共享模块会造成
+    # service 与 worker 两个层次互相导入，不值得。
+    runtime = WorkerRuntime(
+        settings=settings,
+        relationships=relationships,
+        projection_cleaner=ProjectionCleanupService(  # type: ignore[arg-type]
+            relationships=relationships
+        ),
+    )
     try:
         while True:
             try:

app/worker/runtime.py

@@ -79,7 +79,7 @@ class WorkerRuntime:
         endpoint_resolver: ExtractionEndpointResolver | None = None,
         memory_cache: CacheDeleteAdapter | None = None,
         projection_cleaner: ProjectionCleaner | None = None,
-        relationships: Any | None = None,
+        relationships: Any = None,
     ) -> None:
         self.factory = factory if factory is not None else get_agent_factory()
         self.settings = settings or get_settings()
@@ -97,8 +97,9 @@ class WorkerRuntime:
         self.memory_cache: CacheDeleteAdapter | None = (
             memory_cache if memory_cache is not None else get_memory_cache_adapter()
         )
-        # Milvus/Neo4j 删除客户端：当前组装层没有提供（bootstrap 只装配召回用的读适配器），
-        # 因此默认 None = 投影清理显式降级并留痕，绝不写成"删除成功"。
+        # Milvus/Neo4j 删除客户端：由**组装层**（app/worker/__main__.py）注入生产实现，
+        # 这里不兜底。组件内部给默认实现会把"尚未装配"这一事实悄悄盖住——而"未注入即显式
+        # 降级并留痕"是本模块刻意保留的语义（有单测守着），因此默认值保持 None。
         self.projection_cleaner = projection_cleaner
         # 图关系服务：画像投影用它写入节点与关系（投顾的多跳推荐、风控的关系网络都读它）。
         # 默认取生产装配；图库不可用时该值为 None，投影如实降级而不是失败。