memory_manager.py

"""
记忆管理器 - 封装知识库操作，通过 metadata 实现用户隔离

核心功能:
- 记忆存储 (store_memory)
- 记忆召回 (recall_memories)
- 记忆删除 (forget_memory)
- 记忆列表 (list_memories)
- 智能更新 (smart_update_memory)
"""

from __future__ import annotations

import asyncio
import hashlib
import json
import math
import time
import uuid
from collections.abc import Callable
from datetime import datetime, timedelta, timezone
from typing import TYPE_CHECKING, Any

from astrbot.api import logger

from .maintenance.links import MemoryLinkManager
from .memory_protocol import (
    MemoryMetadata,
    MemoryScope,
    MemoryType,
    MemoryURI,
    MemoryVisibility,
    UMOInfo,
    build_session_id,
    build_user_id,
    format_memory_content,
    normalize_memory_scope,
)

if TYPE_CHECKING:
    from astrbot.core.knowledge_base.kb_helper import KBHelper
    from astrbot.core.knowledge_base.kb_mgr import KnowledgeBaseManager
    from astrbot.core.platform import AstrMessageEvent

# KV 存储回调类型
KVPutFn = Callable[[str, Any], Any]
KVGetFn = Callable[[str, Any], Any]
KVDeleteFn = Callable[[str], Any]

SIMILARITY_THRESHOLD = 0.85  # 相似度阈值，用于记忆合并

# 允许的记忆域
_ALLOWED_DOMAINS = frozenset(
    [
        "user_profile",
        "preferences",
        "facts",
        "events",
        "context",
        "fact",
        "preference",
        "event",  # 别名支持
    ]
)

# 域别名映射
_DOMAIN_ALIASES = {
    "fact": "facts",
    "preference": "preferences",
    "event": "events",
}

# 允许的记忆类型
_ALLOWED_MEMORY_TYPES = frozenset(
    [
        MemoryType.NORMAL,
        MemoryType.PERMANENT,
        "normal",
        "permanent",
    ]
)

# 记忆类型别名映射
_MEMORY_TYPE_ALIASES = {
    "normal": MemoryType.NORMAL,
    "permanent": MemoryType.PERMANENT,
}


def _safe_parse_metadata(metadata: Any) -> dict[str, Any]:
    """安全解析 metadata，确保返回字典"""
    if isinstance(metadata, dict):
        return metadata
    if isinstance(metadata, str):
        try:
            parsed = json.loads(metadata)
            return parsed if isinstance(parsed, dict) else {}
        except (json.JSONDecodeError, TypeError):
            return {}
    return {}


def normalize_domain(domain: str) -> str:
    """标准化记忆域名称"""
    domain = (domain or "").lower().strip()
    if domain in _DOMAIN_ALIASES:
        return _DOMAIN_ALIASES[domain]
    if domain in _ALLOWED_DOMAINS:
        return domain
    return "facts"  # 默认域


def normalize_memory_type(memory_type: str) -> str:
    """标准化记忆类型"""
    memory_type = (memory_type or "").lower().strip()
    if memory_type in _MEMORY_TYPE_ALIASES:
        return _MEMORY_TYPE_ALIASES[memory_type]
    if memory_type in _ALLOWED_MEMORY_TYPES:
        return memory_type
    return MemoryType.NORMAL


def normalize_visibility(visibility: str, memory_scope: str) -> str:
    """标准化记忆可见性"""
    visibility = (visibility or "").lower().strip()
    if visibility in (MemoryVisibility.PRIVATE, MemoryVisibility.GROUP):
        return visibility
    return (
        MemoryVisibility.GROUP
        if memory_scope in (MemoryScope.GLOBAL, MemoryScope.GROUP)
        else MemoryVisibility.PRIVATE
    )


def _normalize_sender_ids(sender_ids: list[str] | None, fallback: str) -> list[str]:
    values = sender_ids or [fallback]
    result = []
    for sender_id in values:
        text = str(sender_id).strip()
        if text:
            result.append(text)
    return list(dict.fromkeys(result))


def _clamp_importance(importance: int) -> int:
    """限制重要性范围在 1-5"""
    try:
        return max(1, min(5, int(importance)))
    except (TypeError, ValueError):
        return 3


def _debug_content_summary(value: Any, preview_limit: int = 80) -> dict[str, Any]:
    """返回用于 DEBUG 的脱敏内容摘要。"""
    text = str(value or "").replace("\x00", " ")
    compact = " ".join(text.split())
    preview = compact[: max(0, preview_limit)]
    if preview_limit > 0 and len(compact) > preview_limit:
        preview += "..."
    return {
        "content_len": len(text),
        "content_sha256": hashlib.sha256(text.encode("utf-8")).hexdigest()[:12],
        "preview": preview,
    }


def _debug_filter_summary(filters: dict[str, Any]) -> dict[str, Any]:
    """返回不含用户和会话标识的过滤条件摘要。"""
    return {
        "keys": sorted(filters),
        "domain": filters.get("domain", ""),
        "scope": filters.get("memory_scope", ""),
        "visibility": filters.get("visibility", ""),
        "has_owner_filter": bool(
            filters.get("owner_user_id") or filters.get("owner_session_id")
        ),
        "deprecated": filters.get("deprecated"),
    }


def _debug_memory_summary(memory: dict[str, Any]) -> dict[str, Any]:
    """返回实际召回或注入记忆的 URI 与正文预览。"""
    metadata = _safe_parse_metadata(memory.get("metadata", {}))
    body = (
        metadata.get("memory_content")
        or memory.get("text")
        or memory.get("content", "")
    )
    return {
        "uri": metadata.get("uri", ""),
        "scope": metadata.get("memory_scope", ""),
        "linked": bool(metadata.get("_is_linked", False)),
        "created_at": metadata.get("created_at", ""),
        "updated_at": metadata.get("updated_at", ""),
        "curated_at": metadata.get("curated_at", ""),
        "merged_from_count": len(metadata.get("merged_from", []))
        if isinstance(metadata.get("merged_from"), list)
        else 0,
        "owner_user_ids": metadata.get("owner_user_ids", []),
        "relation_types": metadata.get("_linked_relation_types", []),
        **_debug_content_summary(body),
    }


class MemoryManager:
    """记忆管理器 - 封装单知识库操作，通过 metadata 实现用户隔离"""

    def __init__(
        self,
        kb_mgr: KnowledgeBaseManager,
        config: dict,
        kv_put: KVPutFn | None = None,
        kv_get: KVGetFn | None = None,
        kv_delete: KVDeleteFn | None = None,
    ):
        self.kb_mgr = kb_mgr
        self.config = config
        self._kb_helper: KBHelper | None = None
        self._kb_name: str = ""
        self._rebuilding = False  # 重建/迁移锁
        self._pending_writes: list[dict[str, Any]] = []  # 重建期间暂存的写入
        # KV 持久化回调（由 Star 插件注入）
        self._kv_put = kv_put
        self._kv_get = kv_get
        self._kv_delete = kv_delete
        # 关联表管理器（connect_kb 后初始化）
        self._link_manager: MemoryLinkManager | None = None
        self._last_active_memory_stats: dict[str, Any] = {}

    # ---------- public state accessors ----------
    @property
    def is_kb_connected(self) -> bool:
        """KB 是否已连接"""
        return self._kb_helper is not None

    @property
    def current_kb_name(self) -> str:
        """当前绑定的 KB 名称"""
        return self._kb_name

    @property
    def is_rebuilding(self) -> bool:
        """当前是否正在执行重建/迁移。"""
        return self._rebuilding

    @property
    def link_manager(self) -> MemoryLinkManager | None:
        """关联表管理器（KB 连接后可用）"""
        return self._link_manager

    @property
    def last_active_memory_stats(self) -> dict[str, Any]:
        """返回最近一次活跃记忆分页的向量加载统计。"""
        return dict(self._last_active_memory_stats)

    async def purge_deprecated(
        self, after_days: int = 7, *, dry_run: bool = False
    ) -> dict[str, Any]:
        """物理清理废弃记忆，或返回不写入的候选统计。"""
        if not self._kb_helper:
            return {
                "purged": 0,
                "links_cleaned": 0,
                "candidates": 0,
                "dry_run": dry_run,
            }
        from .maintenance.purge import purge_deprecated_memories

        return await purge_deprecated_memories(
            vec_db=self.vec_db,
            kb_helper=self._kb_helper,
            link_manager=self._link_manager,
            after_days=after_days,
            dry_run=dry_run,
        )

    def load_pending_writes(self, records: list[dict[str, Any]]) -> None:
        """从外部恢复重建期间未落盘的写入缓冲（启动恢复用）"""
        self._pending_writes = list(records)

    def initialize(self) -> None:
        """初始化记忆管理器（仅校验配置，不连接 KB）

        Raises:
            ValueError: 知识库未配置
        """
        kb_name_raw = self.config.get("kb_name", [])
        kb_name = (
            kb_name_raw[0]
            if isinstance(kb_name_raw, list) and kb_name_raw
            else kb_name_raw
        )
        if not kb_name:
            raise ValueError("记忆知识库未配置，请在插件设置中选择一个知识库")

        self._kb_name = kb_name

    async def connect_kb(self) -> None:
        """连接知识库（需在 KB 模块就绪后调用）

        Raises:
            ValueError: 知识库不存在
        """
        kb = await self.kb_mgr.get_kb_by_name(self._kb_name)
        if not kb:
            raise ValueError(f"知识库 '{self._kb_name}' 不存在，请先在知识库管理中创建")

        self._kb_helper = kb
        logger.info(f"[简单长期记忆] 已连接知识库: {self._kb_name}")
        await self._migrate_patch_chunk_index()
        await self._migrate_patch_deprecated_at()
        # 初始化关联表
        self._link_manager = MemoryLinkManager(self.vec_db)
        await self._link_manager.ensure_table()

    async def _migrate_patch_chunk_index(self) -> None:
        """迁移补丁：为缺少 chunk_index 字段的旧记忆条目写入默认值 0。

        旧版插件直接写入 vec_db 时未设置 chunk_index，导致 AstrBot 知识库检索
        界面调用稀疏检索时抛出 KeyError: 'chunk_index'。
        通过 SQLite json_set 原地修改 metadata，无需重新嵌入向量。
        覆盖范围：有 is_memory_record 标记的新版记录 + 有 uri 但无标记的更早记录。
        """
        try:
            doc_storage = self.vec_db.document_storage
            async with doc_storage.get_session() as session, session.begin():
                from sqlalchemy import text as sa_text

                result = await session.execute(
                    sa_text(
                        "UPDATE documents "
                        "SET metadata = json_set(metadata, '$.chunk_index', 0) "
                        "WHERE json_extract(metadata, '$.chunk_index') IS NULL "
                        "  AND (json_extract(metadata, '$.is_memory_record') = 1 "
                        "       OR json_extract(metadata, '$.uri') IS NOT NULL)"
                    )
                )
                patched = result.rowcount
                if patched:
                    logger.info(
                        f"[简单长期记忆] 迁移补丁：已为 {patched} 条旧记忆补写 chunk_index=0"
                    )
        except Exception as e:
            logger.warning(f"[简单长期记忆] 迁移补丁执行失败（不影响功能）: {e}")

    async def _migrate_patch_deprecated_at(self) -> None:
        """迁移补丁：为已废弃但缺少 deprecated_at 的记忆回填时间戳。

        旧版废弃操作只写 deprecated=1，没有记录废弃时间。
        回填迁移执行时间（而非 created_at），确保宽限期从迁移时刻开始计算，
        不会导致历史废弃记忆被立即清理。
        """
        try:
            doc_storage = self.vec_db.document_storage
            async with doc_storage.get_session() as session, session.begin():
                from sqlalchemy import text as sa_text

                # 用 Python UTC ISO 格式，与 purge cutoff 的 isoformat() 保持一致
                now_iso = datetime.now(timezone.utc).isoformat()
                result = await session.execute(
                    sa_text(
                        "UPDATE documents "
                        "SET metadata = json_set(metadata, '$.deprecated_at', "
                        "    :now_iso) "
                        "WHERE json_extract(metadata, '$.deprecated') = 1 "
                        "  AND json_extract(metadata, '$.deprecated_at') IS NULL "
                        "  AND json_extract(metadata, '$.is_memory_record') = 1"
                    ),
                    {"now_iso": now_iso},
                )
                patched = result.rowcount
                if patched:
                    logger.info(
                        f"[简单长期记忆] 迁移补丁：已为 {patched} 条废弃记忆回填 deprecated_at"
                    )
        except Exception as e:
            logger.warning(
                f"[简单长期记忆] deprecated_at 迁移补丁失败（不影响功能）: {e}"
            )

    @property
    def vec_db(self):
        """获取向量数据库实例"""
        if not self._kb_helper:
            raise RuntimeError("记忆管理器未初始化")
        return self._kb_helper.vec_db

    async def _exec_metadata_update(
        self,
        set_clause: str,
        where_clause: str,
        params: dict[str, Any],
    ) -> int:
        """原地更新 documents.metadata（json_set），返回受影响行数。

        绕过 FaissVecDB 无 update_metadata API 的限制：只改 metadata JSON 列，
        不动 FAISS 向量、不动 FTS5 索引，廉价且安全。失败仅记录日志，不阻断检索。
        """
        if not self._kb_helper:
            return 0
        try:
            doc_storage = self.vec_db.document_storage
            async with doc_storage.get_session() as session, session.begin():
                from sqlalchemy import text as sa_text

                result = await session.execute(
                    sa_text(
                        f"UPDATE documents SET metadata = {set_clause} WHERE {where_clause}"
                    ),
                    params,
                )
                return int(result.rowcount or 0)
        except Exception as e:
            logger.warning(f"[简单长期记忆] metadata 原地更新失败（不影响检索）: {e}")
            return 0

    async def _bump_recall_stats(self, uris: list[str], trace_id: str = "") -> int:
        """递增给定 uri 记忆的 recall_count 并刷新 last_recalled_at（P0.1 召回反馈）。"""
        uris = [u for u in uris if u]
        if not uris:
            return 0
        now = datetime.now(timezone.utc).isoformat()
        set_clause = (
            "json_set(metadata, '$.recall_count', "
            "CAST(COALESCE(json_extract(metadata,'$.recall_count'),0) AS INTEGER) + 1, "
            "'$.last_recalled_at', :now)"
        )
        placeholders = ",".join(f":u{i}" for i in range(len(uris)))
        where_clause = (
            f"json_extract(metadata,'$.uri') IN ({placeholders}) "
            "AND json_extract(metadata,'$.is_memory_record') = 1"
        )
        params: dict[str, Any] = {"now": now}
        params.update({f"u{i}": u for i, u in enumerate(uris)})
        updated = await self._exec_metadata_update(set_clause, where_clause, params)
        logger.debug(
            "[简单长期记忆] 召回反馈更新: trace_id=%s, requested=%s, updated=%s",
            trace_id or "-",
            len(uris),
            updated,
        )
        return updated

    async def expire_stale_memories(self, ttl_days: int) -> int:
        """TTL 过期：把超过 ttl_days 且未废弃的记忆标记 deprecated=True（P1.1）。

        召回 filter 已排除 deprecated，标记后即从召回移除。返回标记条数。
        """
        if ttl_days <= 0 or not self._kb_helper:
            return 0
        kb_id = self._kb_helper.kb.kb_id
        cutoff_iso = (datetime.now(timezone.utc) - timedelta(days=ttl_days)).isoformat()
        set_clause = (
            "json_set(metadata, '$.deprecated', 1, '$.deprecated_at', :now_iso)"
        )
        where_clause = (
            "json_extract(metadata,'$.is_memory_record') = 1 "
            "AND json_extract(metadata,'$.deprecated') IS NOT 1 "
            "AND json_extract(metadata,'$.kb_id') = :kb_id "
            "AND json_extract(metadata,'$.created_at') < :cutoff "
            "AND json_extract(metadata,'$.memory_type') != 'permanent' "
            "AND json_extract(metadata,'$.memory_scope') != 'global'"
        )
        params: dict[str, Any] = {
            "kb_id": kb_id,
            "cutoff": cutoff_iso,
            "now_iso": datetime.now(timezone.utc).isoformat(),
        }
        return await self._exec_metadata_update(set_clause, where_clause, params)

    async def fetch_consolidation_candidates(
        self,
        event: AstrMessageEvent,
        min_age_days: int,
        max_recall: int = 1,
        limit: int = 30,
    ) -> list[dict[str, Any]]:
        """取出当前用户低频老旧、未废弃未压缩的个人记忆，作为巩固候选（P1.2）。

        owner 与 memory_scope 由 event 推导并锁定为当前用户 personal，防止跨用户/跨作用域泄露。
        """
        if not self._kb_helper or limit <= 0:
            return []
        owner_user_id = self._current_owner_user_id(event)
        memory_scope = MemoryScope.PERSONAL
        kb_id = self._kb_helper.kb.kb_id
        cutoff_iso = (
            datetime.now(timezone.utc) - timedelta(days=min_age_days)
        ).isoformat()
        try:
            doc_storage = self.vec_db.document_storage
            async with doc_storage.get_session() as session:
                from sqlalchemy import text as sa_text

                rows = (
                    await session.execute(
                        sa_text(
                            "SELECT text, metadata FROM documents "
                            "WHERE json_extract(metadata,'$.is_memory_record') = 1 "
                            "AND json_extract(metadata,'$.deprecated') IS NOT 1 "
                            "AND json_extract(metadata,'$.compressed') IS NOT 1 "
                            "AND json_extract(metadata,'$.kb_id') = :kb_id "
                            "AND json_extract(metadata,'$.created_at') < :cutoff "
                            "AND CAST(COALESCE(json_extract(metadata,'$.recall_count'),0) AS INTEGER) <= :max_recall "
                            "AND json_extract(metadata,'$.owner_user_id') = :owner_user_id "
                            "AND json_extract(metadata,'$.memory_scope') = :memory_scope "
                            "ORDER BY json_extract(metadata,'$.created_at') ASC "
                            "LIMIT :limit"
                        ),
                        {
                            "kb_id": kb_id,
                            "cutoff": cutoff_iso,
                            "max_recall": max_recall,
                            "owner_user_id": owner_user_id,
                            "memory_scope": memory_scope,
                            "limit": limit,
                        },
                    )
                ).all()
        except Exception as e:
            logger.warning(f"[简单长期记忆] 读取巩固候选失败: {e}")
            return []
        candidates: list[dict[str, Any]] = []
        for row in rows:
            text_val = getattr(row, "text", "") or ""
            meta = _safe_parse_metadata(getattr(row, "metadata", {}) or {})
            candidates.append({"text": text_val, "metadata": meta})
        return candidates

    async def mark_consolidated(self, uris: list[str]) -> int:
        """把已巩固的原记忆标记为 deprecated+compressed（P1.2）。"""
        uris = [u for u in uris if u]
        if not uris:
            return 0
        now_iso = datetime.now(timezone.utc).isoformat()
        set_clause = (
            "json_set(metadata, '$.deprecated', 1, '$.compressed', 1, "
            "'$.deprecated_at', :now_iso)"
        )
        placeholders = ",".join(f":u{i}" for i in range(len(uris)))
        where_clause = (
            f"json_extract(metadata,'$.uri') IN ({placeholders}) "
            "AND json_extract(metadata,'$.is_memory_record') = 1"
        )
        params: dict[str, Any] = {f"u{i}": u for i, u in enumerate(uris)}
        params["now_iso"] = now_iso
        return await self._exec_metadata_update(set_clause, where_clause, params)

    def _rerank_by_signal(
        self, memories: list[dict[str, Any]], query: str = ""
    ) -> list[dict[str, Any]]:
        """根据 importance/recall_count/recency/disclosure 对召回结果二次加权排序。

        纯本地计算，不依赖 AstrBot。权重可经配置调整。
        """
        if len(memories) <= 1:
            return memories
        now = time.time()
        half_life = max(self.config.get("recall_recency_halflife_days", 14), 1) * 86400
        w_importance = self.config.get("recall_weight_importance", 0.4)
        w_frequency = self.config.get("recall_weight_frequency", 0.3)
        w_recency = self.config.get("recall_weight_recency", 0.3)
        disclosure_bonus = self.config.get("recall_disclosure_bonus", 0.25)

        # 预计算 query tokens 用于 disclosure 匹配
        query_tokens: set[str] = set()
        if query and disclosure_bonus > 0:
            tokens = self._tokenize_query(query)
            if not tokens:
                tokens = [t for t in query.lower().split() if len(t) >= 2]
            query_tokens = set(tokens)

        def _score(mem: dict[str, Any]) -> float:
            meta = mem.get("metadata", {})
            try:
                _imp = meta.get("importance")
                _imp = 3 if _imp is None or _imp == "" else _imp
                importance = (int(float(_imp)) - 1) / 4.0
            except (ValueError, TypeError):
                importance = 0.5
            try:
                _rc = meta.get("recall_count")
                _rc = 0 if _rc is None or _rc == "" else _rc
                frequency = min(math.log1p(max(0, int(float(_rc)))) / 3.0, 1.0)
            except (ValueError, TypeError):
                frequency = 0.0
            ts_str = meta.get("last_recalled_at") or meta.get("created_at")
            recency = 0.5
            if ts_str:
                try:
                    ts = datetime.fromisoformat(
                        str(ts_str).replace("Z", "+00:00")
                    ).timestamp()
                    recency = math.exp(-max(0.0, now - ts) / half_life)
                except (ValueError, TypeError):
                    recency = 0.5
            score = (
                w_importance * importance
                + w_frequency * frequency
                + w_recency * recency
            )
            # disclosure 匹配加成：query 关键词与 disclosure 文本有交集时加分
            if query_tokens:
                disclosure = str(meta.get("disclosure", "")).lower()
                if disclosure:
                    disc_tokens = self._tokenize_query(disclosure)
                    if not disc_tokens:
                        disc_tokens = [t for t in disclosure.split() if len(t) >= 2]
                    if query_tokens & set(disc_tokens):
                        score += disclosure_bonus
            return score

        return sorted(memories, key=_score, reverse=True)

    # ==================== KB 文档注册 ====================

    async def _register_kb_document(
        self,
        doc_id: str,
        doc_name: str,
        content_size: int,
        kb_helper: KBHelper | None = None,
    ) -> None:
        """将记忆注册为 KB 文档，使其在知识库界面可见"""
        from astrbot.core.knowledge_base.models import KBDocument

        kb = kb_helper or self._kb_helper
        doc = KBDocument(
            doc_id=doc_id,
            kb_id=kb.kb.kb_id,
            doc_name=doc_name,
            file_type="memory",
            file_size=content_size,
            file_path="",
            chunk_count=1,
            media_count=0,
        )
        async with kb.kb_db.get_db() as session:
            async with session.begin():
                session.add(doc)
                await session.commit()

    async def _ensure_kb_document(
        self,
        doc_id: str,
        doc_name: str,
        content_size: int,
        kb_helper: KBHelper | None = None,
    ) -> bool:
        """确保向量文档对应的 KB 文档记录存在。"""
        if not doc_id:
            return False

        kb = kb_helper or self._kb_helper
        if not kb:
            return False

        try:
            existing = await kb.get_document(doc_id)
            if existing:
                return True
            await self._register_kb_document(
                doc_id,
                doc_name,
                content_size,
                kb_helper=kb,
            )
            return True
        except Exception as e:
            logger.warning(f"[简单长期记忆] KB 文档记录修复失败: {doc_id}, {e}")
            return False

    async def _find_memory_doc_by_uri(
        self,
        kb_helper: KBHelper,
        uri: str,
    ) -> dict[str, Any] | None:
        """按 URI 精确查找目标 KB 中已存在的记忆向量文档。"""
        if not uri:
            return None
        docs = await kb_helper.vec_db.document_storage.get_documents(
            metadata_filters={
                "uri": uri,
                "is_memory_record": True,
                "kb_id": kb_helper.kb.kb_id,
            },
            limit=1,
        )
        return docs[0] if docs else None

    async def _repair_kb_document_for_vector_doc(
        self,
        kb_helper: KBHelper,
        doc: dict[str, Any],
        fallback_name: str,
    ) -> bool:
        """向量文档已存在时，补齐缺失的 KB 文档记录。"""
        metadata = _safe_parse_metadata(doc.get("metadata", {}))
        doc_id = metadata.get("kb_doc_id")
        if not doc_id:
            return False
        doc_name = metadata.get("uri") or fallback_name or doc_id
        content_size = len(doc.get("text", "") or "")
        return await self._ensure_kb_document(
            doc_id,
            doc_name,
            content_size,
            kb_helper=kb_helper,
        )

    async def _unregister_kb_documents(
        self,
        doc_ids: list[str],
        kb_helper: KBHelper | None = None,
    ) -> None:
        """批量移除 KB 文档记录"""
        if not doc_ids:
            return
        from astrbot.core.knowledge_base.models import KBDocument
        from sqlmodel import col, delete

        kb = kb_helper or self._kb_helper
        async with kb.kb_db.get_db() as session:
            async with session.begin():
                stmt = delete(KBDocument).where(col(KBDocument.doc_id).in_(doc_ids))
                await session.execute(stmt)
                await session.commit()

    async def _sync_kb_stats(self, kb_helper: KBHelper | None = None) -> None:
        """同步知识库统计数据"""
        kb = kb_helper or self._kb_helper
        await kb.kb_db.update_kb_stats(
            kb_id=kb.kb.kb_id,
            vec_db=kb.vec_db,
        )
        await kb.refresh_kb()

    async def _delete_rebuild_source_records(
        self,
        kb_helper: KBHelper,
        memory_records: list[dict[str, Any]],
    ) -> None:
        kb_id = kb_helper.kb.kb_id
        await kb_helper.vec_db.delete_documents(
            metadata_filters={"is_memory_record": True, "kb_id": kb_id}
        )
        legacy_uris: set[str] = set()
        for record in memory_records:
            metadata = _safe_parse_metadata(record.get("metadata", {}))
            if (
                not metadata.get("is_memory_record")
                and metadata.get("kb_id") == kb_id
                and metadata.get("uri")
            ):
                legacy_uris.add(metadata["uri"])
        for uri in legacy_uris:
            await kb_helper.vec_db.delete_documents(
                metadata_filters={"uri": uri, "kb_id": kb_id}
            )

    def _build_user_filter(self, event: AstrMessageEvent) -> dict[str, Any]:
        """构建用户隔离的 metadata 过滤器

        Args:
            event: 消息事件

        Returns:
            metadata 过滤器字典
        """
        return {
            "user_id": build_user_id(event.get_platform_id(), event.get_sender_id()),
        }

    def _event_scope_ids(
        self, event: AstrMessageEvent, owner_sender_id: str | None = None
    ) -> tuple[UMOInfo, str, str]:
        parsed = UMOInfo.parse(event.unified_msg_origin)
        sender_id = owner_sender_id or event.get_sender_id()
        owner_user_id = build_user_id(parsed.platform_id, sender_id)
        owner_session_id = build_session_id(parsed.platform_id, parsed.session_id)
        return parsed, owner_user_id, owner_session_id

    def _build_owner_user_ids(
        self, platform_id: str, owner_sender_ids: list[str]
    ) -> list[str]:
        return [build_user_id(platform_id, sender_id) for sender_id in owner_sender_ids]

    def _current_owner_user_id(self, event: AstrMessageEvent) -> str:
        parsed = UMOInfo.parse(event.unified_msg_origin)
        return build_user_id(parsed.platform_id, event.get_sender_id())

    def _is_visible_shared_personal(
        self, event: AstrMessageEvent, metadata: dict[str, Any]
    ) -> bool:
        """多 owner personal 记忆只对 owner_user_ids 内的用户可见。"""
        if metadata.get("memory_scope") != MemoryScope.PERSONAL:
            return True
        if metadata.get("visibility") != MemoryVisibility.GROUP:
            return True
        owner_user_ids = metadata.get("owner_user_ids", [])
        if not isinstance(owner_user_ids, list):
            owner_user_ids = []
        return self._current_owner_user_id(event) in owner_user_ids

    def _filter_visible_shared_personal(
        self, event: AstrMessageEvent, memories: list[dict[str, Any]]
    ) -> list[dict[str, Any]]:
        visible = []
        for memory in memories:
            metadata = _safe_parse_metadata(memory.get("metadata", {}))
            if self._is_visible_shared_personal(event, metadata):
                visible.append(memory)
        return visible

    def _scope_filter(
        self,
        event: AstrMessageEvent,
        memory_scope: str,
        global_memory: bool = True,
    ) -> dict[str, Any]:
        _, owner_user_id, owner_session_id = self._event_scope_ids(event)
        scope = normalize_memory_scope(memory_scope)

        if scope == MemoryScope.GLOBAL:
            filters = {
                "memory_scope": MemoryScope.GLOBAL,
            }
        elif scope == MemoryScope.GROUP:
            filters = {
                "memory_scope": MemoryScope.GROUP,
                "owner_session_id": owner_session_id,
            }
        elif scope == MemoryScope.CONVERSATION:
            filters = {
                "memory_scope": MemoryScope.CONVERSATION,
                "umo": event.unified_msg_origin,
            }
        else:
            filters = {
                "memory_scope": MemoryScope.PERSONAL,
                "owner_user_id": owner_user_id,
            }
            if not global_memory:
                filters["umo"] = event.unified_msg_origin

        filters["is_memory_record"] = True
        filters["deprecated"] = False
        return filters

    def _build_query_filter(
        self,
        event: AstrMessageEvent | None,
        *,
        all_users: bool,
        domain: str | None = None,
        include_deprecated: bool = False,
        respect_global: bool = False,
    ) -> dict[str, Any]:
        """统一构建查询/列表/清空使用的 metadata 过滤器。

        Args:
            event: 消息事件（all_users 为 True 时可为 None）
            all_users: True 时跳过用户隔离，使用 is_memory_record 标记
            domain: 可选记忆域过滤
            include_deprecated: 为 False 时排除 deprecated=True 的记忆
            respect_global: True 时按 self.config['global_memory'] 决定是否限定 umo
        """
        if all_users:
            filters: dict[str, Any] = {"is_memory_record": True}
        else:
            if event is None:
                raise ValueError("非 all_users 模式需要传入 event")
            if respect_global:
                global_memory = self.config.get("global_memory", True)
                filters = self._scope_filter(event, MemoryScope.PERSONAL, global_memory)
            else:
                filters = self._build_user_filter(event)
        if not include_deprecated:
            filters["deprecated"] = False
        if domain:
            filters["domain"] = domain
        return filters

    def _build_memory_metadata(
        self,
        event: AstrMessageEvent,
        **extra: Any,
    ) -> dict[str, Any]:
        """构建完整的记忆元数据

        Args:
            event: 消息事件
            **extra: 额外的元数据字段

        Returns:
            完整的元数据字典
        """
        umo = event.unified_msg_origin
        memory_scope = normalize_memory_scope(extra.pop("memory_scope", ""))
        visibility = normalize_visibility(extra.pop("visibility", ""), memory_scope)
        speaker_id = extra.pop("speaker_id", event.get_sender_id())
        owner_sender_id = extra.pop("owner_sender_id", None)
        owner_sender_ids = _normalize_sender_ids(
            extra.pop("owner_sender_ids", None),
            owner_sender_id or event.get_sender_id(),
        )
        parsed, owner_user_id, owner_session_id = self._event_scope_ids(
            event, owner_sender_ids[0]
        )
        owner_user_ids = self._build_owner_user_ids(
            parsed.platform_id, owner_sender_ids
        )

        return {
            "user_id": owner_user_id,
            "platform_id": parsed.platform_id,
            "sender_id": owner_sender_ids[0],
            "umo": umo,
            "session_type": parsed.session_type,
            "session_id": parsed.session_id,
            "memory_scope": memory_scope,
            "owner_user_id": owner_user_id,
            "owner_user_ids": owner_user_ids,
            "owner_session_id": owner_session_id,
            "visibility": visibility,
            "speaker_id": speaker_id,
            "created_at": datetime.now(timezone.utc).isoformat(),
            "last_recalled_at": datetime.now(timezone.utc).isoformat(),
            "recall_count": 0,
            "compressed": False,
            **extra,
        }

    async def store_memory(
        self,
        event: AstrMessageEvent,
        content: str,
        domain: str,
        uri: str | None = None,
        memory_type: str = MemoryType.NORMAL,
        disclosure: str = "",
        importance: int = 3,
        memory_scope: str = MemoryScope.PERSONAL,
        visibility: str = "",
        subject: str = "",
        entities: list[str] | None = None,
        topics: list[str] | None = None,
        owner_sender_id: str | None = None,
        owner_sender_ids: list[str] | None = None,
    ) -> str:
        """存储记忆到知识库

        Args:
            event: 消息事件
            content: 记忆内容
            domain: 记忆域
            uri: 记忆 URI（可选，自动生成）
            memory_type: 记忆类型
            disclosure: 触发召回条件描述
            importance: 重要性等级 (1-5)

        Returns:
            存储的记忆 ID
        """
        # 标准化参数
        domain = normalize_domain(domain)
        memory_type = normalize_memory_type(memory_type)
        importance = _clamp_importance(importance)
        memory_scope = normalize_memory_scope(memory_scope)
        visibility = normalize_visibility(visibility, memory_scope)
        entities = entities or []
        topics = topics or []
        owner_sender_ids = _normalize_sender_ids(
            owner_sender_ids, owner_sender_id or event.get_sender_id()
        )
        if memory_scope == MemoryScope.PERSONAL and len(owner_sender_ids) > 1:
            visibility = MemoryVisibility.GROUP

        if uri is None:
            uri = str(MemoryURI.generate(domain))

        logger.debug(
            "[简单长期记忆] 存储开始: uri=%s, domain=%s, memory_type=%s, "
            "scope=%s, visibility=%s, importance=%s, content=%s",
            uri,
            domain,
            memory_type,
            memory_scope,
            visibility,
            importance,
            _debug_content_summary(content),
        )

        # 重建/迁移期间：暂存到本地缓冲区并持久化到 KV，完成后批量处理
        if self._rebuilding:
            umo = event.unified_msg_origin
            parsed, owner_user_id, owner_session_id = self._event_scope_ids(
                event, owner_sender_ids[0]
            )
            owner_user_ids = self._build_owner_user_ids(
                parsed.platform_id, owner_sender_ids
            )
            item = {
                "content": content,
                "domain": domain,
                "uri": uri,
                "memory_type": memory_type,
                "disclosure": disclosure,
                "importance": importance,
                "user_id": owner_user_id,
                "owner_user_ids": owner_user_ids,
                "platform_id": parsed.platform_id,
                "sender_id": owner_sender_ids[0],
                "umo": umo,
                "session_type": parsed.session_type,
                "session_id": parsed.session_id,
                "memory_scope": memory_scope,
                "owner_user_id": owner_user_id,
                "owner_session_id": owner_session_id,
                "visibility": visibility,
                "speaker_id": owner_sender_ids[0],
                "subject": subject,
                "entities": entities,
                "topics": topics,
            }
            self._pending_writes.append(item)
            # 持久化缓冲区到 KV，防进程重启丢失
            if self._kv_put:
                await self._kv_put("rebuild_pending_writes", self._pending_writes)
            logger.debug(f"[简单长期记忆] 重建进行中，已缓冲记忆: {uri}")
            return uri

        # URI 去重：同名 URI 已存在时，内容相同则跳过，内容不同则换新 URI
        existing = await self.vec_db.document_storage.get_documents(
            metadata_filters={"uri": uri}, limit=1
        )
        if existing:
            old_text = existing[0].get("text", "")
            if old_text.strip() == content.strip():
                logger.debug(f"[简单长期记忆] 内容重复，跳过写入: {uri}")
                return uri
            uri = str(MemoryURI.generate(domain))
            logger.debug(f"[简单长期记忆] URI 冲突且内容不同，已重新生成: {uri}")

        metadata = self._build_memory_metadata(
            event,
            domain=domain,
            uri=uri,
            version=1,
            deprecated=False,
            memory_type=memory_type,
            disclosure=disclosure,
            importance=importance,
            memory_scope=memory_scope,
            visibility=visibility,
            subject=subject,
            entities=entities,
            topics=topics,
            memory_content=content,
            owner_sender_ids=owner_sender_ids,
        )

        # 生成 KB 文档 ID 并关联到向量条目
        doc_id = str(uuid.uuid4())
        metadata["kb_doc_id"] = doc_id
        metadata["kb_id"] = self._kb_helper.kb.kb_id
        metadata["chunk_index"] = 0
        metadata["is_memory_record"] = True

        # 格式化内容
        formatted_content = format_memory_content(content, metadata)

        # 存储到向量数据库
        await self.vec_db.insert(
            content=formatted_content,
            metadata=metadata,
            id=doc_id,
        )

        logger.debug(
            "[简单长期记忆] 存储向量完成: uri=%s, doc_id=%s, kb_id=%s, "
            "formatted_content_len=%s",
            uri,
            doc_id,
            metadata["kb_id"],
            len(formatted_content),
        )

        # 注册为 KB 文档（界面可见）
        kb_registered = False
        try:
            await self._register_kb_document(doc_id, uri, len(formatted_content))
            await self._sync_kb_stats()
            kb_registered = True
            logger.debug(
                "[简单长期记忆] 存储 KB 注册完成: uri=%s, doc_id=%s",
                uri,
                doc_id,
            )
        except Exception as e:
            logger.warning(f"[简单长期记忆] KB 文档注册失败（不影响记忆功能）: {e}")

        logger.debug(
            "[简单长期记忆] 存储完成: uri=%s, doc_id=%s, kb_registered=%s",
            uri,
            doc_id,
            kb_registered,
        )
        return uri

    async def replace_memory(
        self,
        old_metadata: dict[str, Any],
        new_content: str,
        new_disclosure: str | None = None,
        updated_by: str = "user",
    ) -> str:
        """替换记忆内容，完整保留原始租户/作用域/归属元数据。

        与 store_memory 不同，此方法不从 event 重建 owner/session/UMO，
        而是直接复制旧 metadata，仅更新 content/disclosure/version/URI。
        先写入新记录，成功后再删除旧记录。

        Args:
            old_metadata: 旧记忆的完整 metadata dict
            new_content: 新的记忆内容
            new_disclosure: 新的触发条件（None 表示保留旧值）

        Returns:
            新的记忆 URI

        Raises:
            RuntimeError: 写入新记录失败
        """
        old_domain = old_metadata.get("domain", "facts")
        old_uri = old_metadata.get("uri", "")
        new_uri = str(MemoryURI.generate(old_domain))
        new_doc_id = str(uuid.uuid4())

        logger.debug(
            "[简单长期记忆] 替换开始: old_uri=%s, old_doc_id=%s, new_uri=%s, "
            "new_doc_id=%s, content=%s, disclosure_changed=%s",
            old_uri,
            old_metadata.get("kb_doc_id", ""),
            new_uri,
            new_doc_id,
            _debug_content_summary(new_content),
            new_disclosure is not None,
        )
        # 构建新 metadata：复制全部旧字段，只更新内容相关字段
        new_metadata = {**old_metadata}
        new_metadata["uri"] = new_uri
        new_metadata["kb_doc_id"] = new_doc_id
        new_metadata["memory_content"] = new_content
        new_metadata["version"] = old_metadata.get("version", 1) + 1
        new_metadata["updated_at"] = datetime.now(timezone.utc).isoformat()
        new_metadata["updated_by"] = updated_by
        new_metadata["deprecated"] = False
        if new_disclosure is not None:
            new_metadata["disclosure"] = new_disclosure
        # kb_id 保持当前连接的 KB
        if self._kb_helper:
            new_metadata["kb_id"] = self._kb_helper.kb.kb_id

        formatted_content = format_memory_content(new_content, new_metadata)

        # 先写入新记录
        await self.vec_db.insert(
            content=formatted_content,
            metadata=new_metadata,
            id=new_doc_id,
        )
        logger.debug(
            "[简单长期记忆] 替换新向量写入完成: new_uri=%s, new_doc_id=%s",
            new_uri,
            new_doc_id,
        )

        # 注册 KB 文档
        try:
            await self._register_kb_document(
                new_doc_id, new_uri, len(formatted_content)
            )
            await self._sync_kb_stats()
            logger.debug(
                "[简单长期记忆] 替换新 KB 注册完成: new_uri=%s, new_doc_id=%s",
                new_uri,
                new_doc_id,
            )
        except Exception as e:
            logger.warning(f"[简单长期记忆] replace KB 文档注册失败: {e}")

        # 删除旧记录（向量 + KB 文档）
        # 删除旧向量失败时中止，避免数据不一致
        old_doc_id = old_metadata.get("kb_doc_id", "")
        if old_doc_id:
            vector_deleted = False
            try:
                await self.vec_db.delete_documents(
                    metadata_filters={"kb_doc_id": old_doc_id}
                )
                vector_deleted = True
                logger.debug(
                    "[简单长期记忆] 替换旧向量删除完成: old_uri=%s, old_doc_id=%s",
                    old_uri,
                    old_doc_id,
                )
            except Exception as e:
                logger.error(f"[简单长期记忆] replace 删除旧向量失败，中止操作: {e}")
                # 尝试回滚：删除刚写入的新记录（向量 + KB 文档）
                try:
                    await self.vec_db.delete_documents(
                        metadata_filters={"kb_doc_id": new_doc_id}
                    )
                    logger.info(
                        f"[简单长期记忆] replace 回滚成功，已删除新向量记录: {new_doc_id}"
                    )
                except Exception as rollback_err:
                    logger.error(f"[简单长期记忆] replace 回滚向量失败: {rollback_err}")

                # 同时注销 KB 文档，避免 dangling entry
                try:
                    await self._unregister_kb_documents([new_doc_id])
                    await self._sync_kb_stats()
                    logger.info(
                        f"[简单长期记忆] replace 回滚成功，已注销 KB 文档: {new_doc_id}"
                    )
                except Exception as kb_err:
                    logger.error(f"[简单长期记忆] replace 回滚 KB 文档失败: {kb_err}")

                raise RuntimeError(f"replace 失败：无法删除旧向量 {old_doc_id}: {e}")
            # 只有向量删除成功后才继续清理 KB 文档
            if vector_deleted:
                try:
                    await self._unregister_kb_documents([old_doc_id])
                    await self._sync_kb_stats()
                    logger.debug(
                        "[简单长期记忆] 替换旧 KB 注销完成: old_uri=%s, old_doc_id=%s",
                        old_uri,
                        old_doc_id,
                    )
                except Exception as e:
                    logger.warning(f"[简单长期记忆] replace 删除旧 KB 文档失败: {e}")
        # 迁移关联边
        if self._link_manager and old_uri:
            try:
                out_links = await self._link_manager.get_links_for_uri(
                    old_uri, injectable_only=False, limit=0
                )
                in_links = await self._link_manager.get_links_to_uri(
                    old_uri, injectable_only=False, limit=0
                )
                for link in out_links:
                    await self._link_manager.add_link(
                        new_uri,
                        link["target_uri"],
                        link["relation_type"],
                        reason=link.get("reason", ""),
                        confidence=link.get("confidence", 1.0),
                        created_by="replace",
                    )
                for link in in_links:
                    await self._link_manager.add_link(
                        link["source_uri"],
                        new_uri,
                        link["relation_type"],
                        reason=link.get("reason", ""),
                        confidence=link.get("confidence", 1.0),
                        created_by="replace",
                    )
                await self._link_manager.delete_links_for_uris([old_uri])
                logger.debug(
                    "[简单长期记忆] 替换关联迁移完成: old_uri=%s, new_uri=%s, "
                    "outbound=%s, inbound=%s",
                    old_uri,
                    new_uri,
                    len(out_links),
                    len(in_links),
                )
            except Exception as e:
                logger.warning(f"[简单长期记忆] replace 关联迁移失败: {e}")

        logger.debug(
            "[简单长期记忆] 替换完成: old_uri=%s, old_doc_id=%s, new_uri=%s, "
            "new_doc_id=%s",
            old_uri,
            old_metadata.get("kb_doc_id", ""),
            new_uri,
            new_doc_id,
        )
        return new_uri

    async def recall_memories(
        self,
        event: AstrMessageEvent,
        query: str,
        domain: str | None = None,
        top_k: int | None = None,
        all_users: bool = False,
        memory_scope: str | None = None,
        bump: bool = False,
        source: str = "unspecified",
    ) -> list[dict[str, Any]]:
        """召回相关记忆（自动按用户隔离）

        Args:
            event: 消息事件
            query: 查询文本
            domain: 记忆域过滤（可选）
            top_k: 返回数量（可选，使用配置值）
            all_users: 为 True 时跳过用户过滤

        Returns:
            记忆列表，每项包含 'text' 和 'metadata'
        """
        trace_id = uuid.uuid4().hex[:8]
        started_at = time.monotonic()
        logger.debug(
            "[简单长期记忆] 召回开始: trace_id=%s, source=%s, query=%s, "
            "domain=%s, requested_top_k=%s, all_users=%s, scope=%s, bump=%s",
            trace_id,
            source,
            _debug_content_summary(query, preview_limit=0),
            domain or "",
            top_k,
            all_users,
            memory_scope or "",
            bump,
        )
        if top_k is None:
            top_k = self.config.get("max_memories_per_inject", 5)
        fetch_k = max(top_k, min(top_k * 3, 20))

        if all_users:
            filters = {"is_memory_record": True, "deprecated": False}
            if domain:
                filters["domain"] = domain
            if memory_scope:
                filters["memory_scope"] = memory_scope
            raw = await self._retrieve_with_filter(
                query, fetch_k, filters, trace_id=trace_id
            )
            deduped = self._dedupe_memories(raw)
            memories = deduped[:top_k]
            logger.debug(
                "[简单长期记忆] 召回全局过滤完成: trace_id=%s, filter=%s, "
                "raw=%s, deduped=%s, selected=%s",
                trace_id,
                _debug_filter_summary(filters),
                len(raw),
                len(deduped),
                len(memories),
            )
        else:
            global_memory = self.config.get("global_memory", True)
            filters_list = self._build_recall_filters(
                event,
                global_memory=global_memory,
                domain=domain,
                memory_scope=memory_scope,
            )

            logger.debug(
                "[简单长期记忆] 召回过滤器已构建: trace_id=%s, count=%s, filters=%s",
                trace_id,
                len(filters_list),
                [_debug_filter_summary(filters) for filters in filters_list],
            )

            tasks = [
                self._retrieve_with_filter(query, fetch_k, filters, trace_id=trace_id)
                for filters in filters_list
            ]
            results_list = await asyncio.gather(*tasks)
            channel_counts = [len(items) for items in results_list]
            combined = [item for sublist in results_list for item in sublist]
            visible = self._filter_visible_shared_personal(event, combined)
            deduped = self._dedupe_memories(visible)
            memories = deduped[:top_k]
            logger.debug(
                "[简单长期记忆] 召回过滤完成: trace_id=%s, per_filter=%s, "
                "combined=%s, visible=%s, visibility_dropped=%s, deduped=%s, selected=%s",
                trace_id,
                channel_counts,
                len(combined),
                len(visible),
                len(combined) - len(visible),
                len(deduped),
                len(memories),
            )

        # P0.2 信号加权重排（importance / recall_count / recency / disclosure 匹配）
        before_rerank = [
            memory.get("metadata", {}).get("uri", "") for memory in memories
        ]
        memories = self._rerank_by_signal(memories, query=query)
        after_rerank = [
            memory.get("metadata", {}).get("uri", "") for memory in memories
        ]
        logger.debug(
            "[简单长期记忆] 召回信号重排完成: trace_id=%s, count=%s, order_changed=%s",
            trace_id,
            len(memories),
            before_rerank != after_rerank,
        )
        # P0.1 召回反馈：仅注入路径递增 recall_count（避免 search/selftest/工具调用污染频次信号）
        if bump:
            recalled_uris = [
                m.get("metadata", {}).get("uri")
                for m in memories
                if m.get("metadata", {}).get("uri")
            ]
            await self._bump_recall_stats(recalled_uris, trace_id=trace_id)

        # 召回时注入关联记忆（单跳，最多 3 条，排除 contradicts/supersedes）
        if self._link_manager and not (all_users and memory_scope):
            linked_memories = await self._inject_linked_memories(
                memories, event, all_users=all_users, trace_id=trace_id
            )
            if linked_memories:
                memories.extend(linked_memories)
                # 关联注入后截断到 top_k，避免超出调用方预期
                if top_k and len(memories) > top_k:
                    memories = memories[:top_k]
        logger.debug(
            "[简单长期记忆] 召回完成: trace_id=%s, source=%s, count=%s, "
            "elapsed_ms=%s, memories=%s",
            trace_id,
            source,
            len(memories),
            round((time.monotonic() - started_at) * 1000),
            [_debug_memory_summary(memory) for memory in memories],
        )
        return memories

    async def _inject_linked_memories(
        self,
        memories: list[dict[str, Any]],
        event: AstrMessageEvent | None = None,
        all_users: bool = False,
        trace_id: str = "",
    ) -> list[dict[str, Any]]:
        """注入关联记忆（单跳，最多 3 条，排除 contradicts/supersedes）。

        规则：
        - 只召回未 deprecated 的关联记忆
        - 单跳（不递归），最多 3 条关联记忆
        - 总注入 token 预算上限（关联记忆 + 主记忆合计不超过 max_length）
        - contradicts/supersedes 关系不注入召回，仅用于后台分析
        """
        if not self._link_manager:
            return []

        # 收集所有主记忆的 URI
        main_uris = [
            m.get("metadata", {}).get("uri")
            for m in memories
            if m.get("metadata", {}).get("uri")
        ]
        if not main_uris:
            return []

        # 查询每条主记忆的关联（单跳，出边 + 入边，只取可注入类型）
        linked_uris: set[str] = set()
        linked_relations: dict[str, set[str]] = {}
        for uri in main_uris[:5]:  # 限制查询数量，避免过多 DB 查询
            try:
                # 出边：uri → target
                out_links = await self._link_manager.get_links_for_uri(
                    uri=uri,
                    injectable_only=True,
                    min_confidence=0.5,
                    limit=3,
                )
                for link in out_links:
                    target_uri = link.get("target_uri", "")
                    if target_uri and target_uri not in main_uris:
                        linked_uris.add(target_uri)
                        linked_relations.setdefault(target_uri, set()).add(
                            link.get("relation_type", "related")
                        )
                # 入边：source → uri（对称关系反向查询）
                in_links = await self._link_manager.get_links_to_uri(
                    uri=uri,
                    injectable_only=True,
                    min_confidence=0.5,
                    limit=3,
                )
                for link in in_links:
                    source_uri = link.get("source_uri", "")
                    if source_uri and source_uri not in main_uris:
                        linked_relations.setdefault(source_uri, set()).add(
                            link.get("relation_type", "related")
                        )
                        linked_uris.add(source_uri)
            except Exception as e:
                logger.debug(f"[简单长期记忆] 查询关联失败: {uri}, {e}")
        if not linked_uris:
            logger.debug(
                "[简单长期记忆] 关联注入无候选: trace_id=%s, primary_count=%s",
                trace_id or "-",
                len(main_uris),
            )
            return []

        # 拉取关联记忆内容（限制数量）
        linked_memories: list[dict[str, Any]] = []
        max_linked = 3  # 最多 3 条关联记忆
        for uri in list(linked_uris)[:max_linked]:
            try:
                # 通过 URI 查询记忆
                mem = await self._get_memory_by_uri(uri)
                if mem and not mem.get("metadata", {}).get("deprecated", False):
                    # 验证可见性，确保关联记忆对当前用户可见（all_users 模式下跳过）
                    if (
                        event
                        and not all_users
                        and not self._is_memory_visible(event, mem)
                    ):
                        logger.debug(f"[简单长期记忆] 关联记忆不可见，跳过: {uri}")
                        continue
                    # 标记为关联记忆
                    # 标记关联来源和关系，供注入格式明确区分身份
                    linked_meta = mem.setdefault("metadata", {})
                    linked_meta["_is_linked"] = True
                    linked_meta["_linked_relation_types"] = sorted(
                        linked_relations.get(uri, {"related"})
                    )
                    linked_meta["_linked_from_uris"] = main_uris[:5]
                    linked_memories.append(mem)
            except Exception as e:
                logger.debug(f"[简单长期记忆] 拉取关联记忆失败: {uri}, {e}")

        logger.debug(
            "[简单长期记忆] 关联注入完成: trace_id=%s, primary_count=%s, "
            "candidate_count=%s, injected=%s, memories=%s",
            trace_id or "-",
            len(main_uris),
            len(linked_uris),
            len(linked_memories),
            [_debug_memory_summary(memory) for memory in linked_memories],
        )
        return linked_memories

    async def _get_memory_by_uri(self, uri: str) -> dict[str, Any] | None:
        """通过 URI 查询单条记忆。

        使用 document_storage.get_documents() 直接查询，
        避免语义搜索的不确定性。
        """
        try:
            if not self.vec_db or not hasattr(self.vec_db, "document_storage"):
                logger.warning("[简单长期记忆] document_storage 不可用")
                return None

            doc_storage = self.vec_db.document_storage
            async with doc_storage.get_session() as session:
                from sqlalchemy import text as sa_text

                result = await session.execute(
                    sa_text(
                        "SELECT doc_id, text, metadata FROM documents "
                        "WHERE json_extract(metadata, '$.uri') = :uri "
                        "AND json_extract(metadata, '$.is_memory_record') = 1 "
                        "AND json_extract(metadata, '$.deprecated') = 0 "
                        "LIMIT 1"
                    ),
                    {"uri": uri},
                )
                row = result.first()
                if row:
                    import json

                    meta = (
                        json.loads(row.metadata)
                        if isinstance(row.metadata, str)
                        else row.metadata
                    )
                    return {
                        "doc_id": row.doc_id,
                        "content": row.text,
                        "metadata": meta,
                    }
                return None
        except Exception as e:
            logger.debug(f"[简单长期记忆] 查询记忆失败: {uri}, {e}")
            return None

    def _is_memory_visible(
        self, event: AstrMessageEvent, memory: dict[str, Any]
    ) -> bool:
        """检查记忆对当前事件是否可见。

        确保关联记忆注入时验证可见性，避免跨 scope 泄露。
        """
        meta = memory.get("metadata", {})
        scope = meta.get("memory_scope", MemoryScope.PERSONAL)
        owner = meta.get("owner_user_id", "")

        # global 记忆所有人可见
        if scope == MemoryScope.GLOBAL:
            return True

        # 获取当前用户信息（build_user_id 统一命名空间）
        current_user = self._current_owner_user_id(event)
        parsed = UMOInfo.parse(event.unified_msg_origin)
        # personal 记忆：仅 owner 可见（包括多 owner 共享场景）
        if scope == MemoryScope.PERSONAL:
            # 单 owner：直接比较
            if owner == current_user:
                return True
            # 多 owner 共享：检查 owner_user_ids
            return self._is_visible_shared_personal(event, meta)

        # group 记忆：同群可见（session_id 含群聊标识）
        if scope == MemoryScope.GROUP:
            mem_session = meta.get("owner_session_id", "")
            current_session = build_session_id(parsed.platform_id, parsed.session_id)
            return mem_session == current_session

        # conversation 记忆：精确匹配完整 UMO，避免私聊/群聊 ID 碰撞
        if scope == MemoryScope.CONVERSATION:
            mem_umo = meta.get("umo", "")
            return bool(mem_umo) and mem_umo == event.unified_msg_origin

        # 默认不可见
        return False

    def _build_recall_filters(
        self,
        event: AstrMessageEvent,
        global_memory: bool,
        domain: str | None = None,
        memory_scope: str | None = None,
    ) -> list[dict[str, Any]]:
        parsed = UMOInfo.parse(event.unified_msg_origin)
        scopes = (
            [normalize_memory_scope(memory_scope)]
            if memory_scope
            else [MemoryScope.GLOBAL, MemoryScope.PERSONAL]
        )
        if not memory_scope:
            if parsed.session_type == "group":
                scopes.extend([MemoryScope.GROUP, MemoryScope.CONVERSATION])
            else:
                scopes.append(MemoryScope.CONVERSATION)

        filters_list = []
        for scope in scopes:
            filters = self._scope_filter(event, scope, global_memory)
            if domain:
                filters["domain"] = domain
            filters_list.append(filters)
            if scope == MemoryScope.PERSONAL:
                if parsed.session_type == "group":
                    group_personal_filters = {
                        "memory_scope": MemoryScope.PERSONAL,
                        "owner_session_id": build_session_id(
                            parsed.platform_id, parsed.session_id
                        ),
                        "visibility": MemoryVisibility.GROUP,
                        "is_memory_record": True,
                        "deprecated": False,
                    }
                    if domain:
                        group_personal_filters["domain"] = domain
                    filters_list.append(group_personal_filters)
        return filters_list

    async def _retrieve_with_filter(
        self,
        query: str,
        top_k: int,
        filters: dict[str, Any],
        trace_id: str = "",
    ) -> list[dict[str, Any]]:
        started_at = time.monotonic()
        logger.debug(
            "[简单长期记忆] 检索通道开始: trace_id=%s, top_k=%s, filter=%s",
            trace_id or "-",
            top_k,
            _debug_filter_summary(filters),
        )
        use_rerank = self.config.get("use_reranker", True)
        # 稠密检索（本阶段不 rerank，融合后再统一 rerank）
        dense_results = await self.vec_db.retrieve(
            query=query,
            k=top_k,
            rerank=False,
            metadata_filters=filters,
        )
        dense_memories: list[dict[str, Any]] = []
        for result in dense_results:
            data = result.data
            dense_memories.append(
                {
                    "text": data.get("text", ""),
                    "metadata": _safe_parse_metadata(data.get("metadata", {})),
                    "similarity": result.similarity,
                }
            )

        # P0.3 稀疏检索（FTS5）
        sparse_memories: list[dict[str, Any]] = []
        if self.config.get("recall_sparse_fusion", True):
            sparse_memories = await self._sparse_retrieve(
                query, top_k, filters, trace_id=trace_id
            )

        # Disclosure 精确匹配通道
        disclosure_memories = await self._disclosure_retrieve(
            query, top_k, filters, trace_id=trace_id
        )

        # RRF 多通道融合（空列表自动跳过）
        channels = [dense_memories]
        if sparse_memories:
            channels.append(sparse_memories)
        if disclosure_memories:
            channels.append(disclosure_memories)
        memories = (
            self._rrf_fuse(*channels, limit=top_k)
            if len(channels) > 1
            else dense_memories
        )
        # 融合后统一 rerank（复用知识库配置的 rerank provider）
        rerank_applied = False
        if use_rerank and memories:
            rerank_provider = getattr(self.vec_db, "rerank_provider", None)
            if rerank_provider:
                try:
                    docs = [m.get("text", "") for m in memories]
                    reranked = await rerank_provider.rerank(query, docs)
                    reranked = sorted(
                        reranked, key=lambda x: x.relevance_score, reverse=True
                    )
                    memories = [
                        memories[r.index]
                        for r in reranked
                        if 0 <= r.index < len(memories)
                    ]
                    rerank_applied = True
                except Exception as e:
                    logger.debug(f"[简单长期记忆] rerank 失败，使用融合结果: {e}")
        logger.debug(
            "[简单长期记忆] 检索通道完成: trace_id=%s, dense=%s, sparse=%s, "
            "disclosure=%s, fused=%s, rerank_applied=%s, output=%s, elapsed_ms=%s",
            trace_id or "-",
            len(dense_memories),
            len(sparse_memories),
            len(disclosure_memories),
            len(memories),
            rerank_applied,
            len(memories),
            round((time.monotonic() - started_at) * 1000),
        )
        return memories

    async def _sparse_retrieve(
        self,
        query: str,
        top_k: int,
        filters: dict[str, Any],
        trace_id: str = "",
    ) -> list[dict[str, Any]]:
        """FTS5 稀疏检索（P0.3）。不可用或失败时返回空列表，调用方回退纯稠密。"""
        tokens = self._tokenize_query(query)
        if not tokens:
            logger.debug(
                "[简单长期记忆] 稀疏检索跳过: trace_id=%s, reason=no_tokens",
                trace_id or "-",
            )
            return []
        try:
            docs = await self.vec_db.document_storage.search_sparse(
                query_tokens=tokens, limit=max(top_k * 3, 20)
            )
        except Exception as e:
            logger.debug(f"[简单长期记忆] 稀疏检索失败，回退纯稠密: {e}")
            return []
        if not docs:
            return []
        out: list[dict[str, Any]] = []
        for doc in docs:
            meta = _safe_parse_metadata(doc.get("metadata", {}))
            if not self._matches_filters(meta, filters):
                continue
            out.append(
                {
                    "text": doc.get("text", ""),
                    "metadata": meta,
                    "similarity": -float(doc.get("score", 0) or 0),
                }
            )
        logger.debug(
            "[简单长期记忆] 稀疏检索完成: trace_id=%s, tokens=%s, scanned=%s, "
            "filtered_out=%s, matched=%s",
            trace_id or "-",
            len(tokens),
            len(docs),
            len(docs) - len(out),
            len(out),
        )
        return out

    def _tokenize_query(self, query: str) -> list[str]:
        """分词（依赖缓存）。AstrBot 检索依赖不可用时返回空列表。"""
        cache = getattr(self, "_sparse_tokenize", None)
        if cache is None:
            try:
                import os

                from astrbot.core.knowledge_base.retrieval import (
                    sparse_retriever as _sr,
                )
                from astrbot.core.knowledge_base.retrieval.tokenizer import (
                    load_stopwords,
                    tokenize_text,
                )

                stopwords = load_stopwords(
                    os.path.join(os.path.dirname(_sr.__file__), "hit_stopwords.txt")
                )
                self._sparse_tokenize = (tokenize_text, stopwords)
                cache = self._sparse_tokenize
            except Exception as e:
                logger.debug(f"[简单长期记忆] 稀疏检索分词依赖不可用: {e}")
                self._sparse_tokenize = False
                return []
        if cache is False:
            return []
        tokenize_text, stopwords = cache
        return tokenize_text(query, stopwords)

    @staticmethod
    def _matches_filters(metadata: dict[str, Any], filters: dict[str, Any]) -> bool:
        """本地 metadata 过滤（稀疏检索结果不带 filter，需本地过滤）。"""
        for key, expected in filters.items():
            if key == "is_memory_record":
                if bool(metadata.get(key)) != bool(expected):
                    return False
            elif key == "deprecated":
                if bool(metadata.get(key, False)) != bool(expected):
                    return False
            elif metadata.get(key) != expected:
                return False
        return True

    @staticmethod
    def _filters_to_sql(filters: dict[str, Any]) -> tuple[str, dict[str, Any]]:
        """将 metadata_filters dict 转为参数化 SQL WHERE 子句。

        返回 (where_clause, params)，where_clause 不含 WHERE 关键字。
        """
        clauses: list[str] = []
        params: dict[str, Any] = {}
        for key, value in filters.items():
            if key == "is_memory_record":
                clauses.append("json_extract(metadata,'$.is_memory_record') = 1")
            elif key == "deprecated":
                # deprecated=False 意为“未废弃”，NULL 也视为未废弃
                clauses.append("json_extract(metadata,'$.deprecated') IS NOT 1")
            else:
                param_name = f"f_{key}"
                clauses.append(f"json_extract(metadata,'$.{key}') = :{param_name}")
                params[param_name] = value
        return " AND ".join(clauses) if clauses else "1=1", params

    async def _disclosure_retrieve(
        self,
        query: str,
        top_k: int,
        filters: dict[str, Any],
        trace_id: str = "",
    ) -> list[dict[str, Any]]:
        """Disclosure 精确匹配检索通道。

        查找 disclosure 字段与查询关键词有交集的记忆，
        确保“当用户问 X 时”类型的触发条件不会被向量相似度漏掉。
        """
        if not self._kb_helper:
            return []

        # 分词：优先用 AstrBot 分词器，不可用时 fallback 到空格分词
        query_tokens = self._tokenize_query(query)
        if not query_tokens:
            query_tokens = [t for t in query.lower().split() if len(t) >= 2]
        if not query_tokens:
            return []

        # 构建 SQL：在现有 filters 基础上追加 disclosure 非空条件
        where_clause, params = self._filters_to_sql(filters)
        where_clause += (
            " AND json_extract(metadata,'$.disclosure') IS NOT NULL"
            " AND json_extract(metadata,'$.disclosure') != ''"
        )
        # 分页扫描直到找到足够匹配或遍历完所有候选
        try:
            scan_limit = int(self.config.get("disclosure_scan_limit", 200))
        except (TypeError, ValueError):
            scan_limit = 200
        scan_limit = max(10, min(scan_limit, 5000))

        query_token_set = set(query_tokens)
        matched: list[dict[str, Any]] = []
        target_matches = min(top_k * 2, 10)
        page_size = min(scan_limit, 200)
        offset = 0

        try:
            doc_storage = self.vec_db.document_storage
            async with doc_storage.get_session() as session:
                from sqlalchemy import text as sa_text

                while offset < scan_limit and len(matched) < target_matches:
                    batch_limit = min(page_size, scan_limit - offset)
                    rows = (
                        await session.execute(
                            sa_text(
                                f"SELECT text, metadata FROM documents "
                                f"WHERE {where_clause} "
                                "ORDER BY json_extract(metadata,'$.created_at') DESC "
                                "LIMIT :batch_lim OFFSET :batch_off"
                            ),
                            {**params, "batch_lim": batch_limit, "batch_off": offset},
                        )
                    ).all()
                    if not rows:
                        break
                    offset += len(rows)

                    for row in rows:
                        meta = _safe_parse_metadata(getattr(row, "metadata", {}) or {})
                        disclosure = str(meta.get("disclosure", "")).lower()
                        if not disclosure:
                            continue
                        disc_tokens = self._tokenize_query(disclosure)
                        if not disc_tokens:
                            disc_tokens = [t for t in disclosure.split() if len(t) >= 2]
                        if query_token_set & set(disc_tokens):
                            matched.append(
                                {
                                    "text": getattr(row, "text", "") or "",
                                    "metadata": meta,
                                    "similarity": 0.0,
                                }
                            )
                            if len(matched) >= target_matches:
                                break
        except Exception as e:
            logger.debug(f"[简单长期记忆] disclosure 检索失败: {e}")
            return []

        logger.debug(
            "[简单长期记忆] disclosure 检索完成: trace_id=%s, tokens=%s, "
            "scanned=%s, scan_limit=%s, target=%s, matched=%s",
            trace_id or "-",
            len(query_tokens),
            offset,
            scan_limit,
            target_matches,
            len(matched),
        )
        return matched

    def _rrf_fuse(
        self,
        *channels: list[dict[str, Any]],
        k: int = 60,
        limit: int = 20,
    ) -> list[dict[str, Any]]:
        """Reciprocal Rank Fusion 融合多路检索结果。

        score(doc) = sum(1 / (k + rank))，同 AstrBot rank_fusion.py 公式；uri 作去重主键。
        接受任意数量的排序列表（稠密/稀疏/disclosure 等），空列表自动跳过。
        """
        scores: dict[str, float] = {}
        mem_by_uri: dict[str, dict[str, Any]] = {}

        for channel in channels:
            for rank, mem in enumerate(channel):
                uri = mem.get("metadata", {}).get("uri") or mem.get("text", "")
                if not uri:
                    continue
                scores[uri] = scores.get(uri, 0.0) + 1.0 / (k + rank + 1)
                mem_by_uri.setdefault(uri, mem)

        ordered = sorted(scores, key=lambda u: scores[u], reverse=True)[:limit]
        fused: list[dict[str, Any]] = []
        for uri in ordered:
            mem = mem_by_uri.get(uri)
            if mem:
                # 保留原始 similarity（dense cosine / sparse 负分），RRF 分独立存放；
                # 勿覆盖 similarity，否则 smart_update_memory 的 0.85 阈值与 _dedupe 排序失效
                fused.append({**mem, "rrf_score": scores[uri]})
        return fused

    def _dedupe_memories(self, memories: list[dict[str, Any]]) -> list[dict[str, Any]]:
        seen = set()
        deduped = []
        for mem in sorted(memories, key=lambda m: m.get("similarity", 0), reverse=True):
            metadata = mem.get("metadata", {})
            key = metadata.get("uri") or mem.get("text", "")
            if key in seen:
                continue
            seen.add(key)
            deduped.append(mem)
        return deduped

    async def forget_memory(self, event: AstrMessageEvent, uri: str) -> int:
        """删除当前用户指定 URI 的记忆（LLM 工具用，按 user_id 隔离）

        Args:
            event: 消息事件
            uri: 记忆 URI

        Returns:
            删除数。无权限和不存在都返回 0，避免泄漏跨用户 URI 存在性。
        """
        filters = self._build_user_filter(event)
        filters["uri"] = uri
        deleted = await self._delete_by_filters(filters, uri)
        if deleted > 0:
            return deleted

        shared_docs = await self.vec_db.document_storage.get_documents(
            metadata_filters={
                "uri": uri,
                "memory_scope": MemoryScope.PERSONAL,
                "visibility": MemoryVisibility.GROUP,
                "is_memory_record": True,
                "deprecated": False,
            },
            limit=20,
        )
        for doc in shared_docs:
            metadata = _safe_parse_metadata(doc.get("metadata", {}))
            if self._is_visible_shared_personal(event, metadata):
                return await self._delete_by_filters(
                    {"uri": uri, "kb_doc_id": metadata.get("kb_doc_id", "")},
                    uri,
                )
        return 0

    async def forget_memory_by_uri(self, uri: str) -> int:
        """管理员按 URI 删除所有匹配的记忆（不限用户）

        Args:
            uri: 记忆 URI

        Returns:
            实际删除的记录数
        """
        return await self._delete_by_filters({"uri": uri}, uri)

    async def preview_admin_memory_removal(
        self, uri: str, memory_scope: str, owner_id: str
    ) -> dict[str, Any]:
        """返回管理员精确删除目标的预览及当前记录指纹。"""
        filters: dict[str, Any] = {
            "uri": uri,
            "memory_scope": memory_scope,
            "is_memory_record": True,
            "deprecated": False,
        }
        owner_field = {
            MemoryScope.PERSONAL: "owner_user_id",
            MemoryScope.GROUP: "owner_session_id",
            MemoryScope.CONVERSATION: "umo",
        }.get(memory_scope, "")
        if owner_field:
            filters[owner_field] = owner_id
        try:
            page_size = int(self.config.get("memory_delete_scan_page_size", 1000))
        except (TypeError, ValueError):
            page_size = 1000
        page_size = max(1, page_size)
        offset = 0
        records: list[dict[str, Any]] = []
        fingerprint_items: list[dict[str, str]] = []
        while True:
            docs = await self.vec_db.document_storage.get_documents(
                metadata_filters=filters, offset=offset, limit=page_size
            )
            if not docs:
                break
            offset += len(docs)
            for doc in docs:
                metadata = _safe_parse_metadata(doc.get("metadata", {}))
                summary = _debug_memory_summary(
                    {"text": doc.get("text", ""), "metadata": metadata}
                )
                records.append(summary)
                fingerprint_items.append(
                    {
                        "doc_id": str(metadata.get("kb_doc_id") or doc.get("id", "")),
                        "uri": str(metadata.get("uri", "")),
                        "scope": str(metadata.get("memory_scope", "")),
                        "owner": str(metadata.get(owner_field, "")),
                        "content_sha256": str(summary["content_sha256"]),
                    }
                )
            if len(docs) < page_size:
                break
        fingerprint_payload = json.dumps(
            sorted(
                fingerprint_items,
                key=lambda item: (
                    item["doc_id"],
                    item["uri"],
                    item["scope"],
                    item["owner"],
                    item["content_sha256"],
                ),
            ),
            ensure_ascii=True,
            separators=(",", ":"),
        )
        fingerprint = hashlib.sha256(fingerprint_payload.encode("utf-8")).hexdigest()
        logger.debug(
            "[简单长期记忆] 管理员记忆删除预览: uri=%s, scope=%s, owner=%s, "
            "records=%s, fingerprint=%s",
            uri,
            memory_scope,
            owner_id or "global",
            len(records),
            fingerprint[:12],
        )
        return {
            "count": len(records),
            "records": records,
            "fingerprint": fingerprint,
        }

    async def forget_admin_memory_by_target(
        self, uri: str, memory_scope: str, owner_id: str, expected_fingerprint: str
    ) -> dict[str, Any]:
        """按 URI、scope 和归属精确删除管理员选定的记忆。"""
        preview = await self.preview_admin_memory_removal(uri, memory_scope, owner_id)
        if preview["count"] == 0:
            return {"status": "not_found", "deleted": 0, "preview": preview}
        if preview["fingerprint"] != expected_fingerprint:
            logger.warning(
                "[简单长期记忆] 管理员记忆删除确认失效: uri=%s, scope=%s, "
                "owner=%s, expected=%s, actual=%s",
                uri,
                memory_scope,
                owner_id or "global",
                expected_fingerprint[:12],
                preview["fingerprint"][:12],
            )
            return {"status": "changed", "deleted": 0, "preview": preview}
        filters: dict[str, Any] = {
            "uri": uri,
            "memory_scope": memory_scope,
            "is_memory_record": True,
            "deprecated": False,
        }
        owner_field = {
            MemoryScope.PERSONAL: "owner_user_id",
            MemoryScope.GROUP: "owner_session_id",
            MemoryScope.CONVERSATION: "umo",
        }.get(memory_scope, "")
        if owner_field:
            filters[owner_field] = owner_id
        deleted = await self._delete_by_filters(filters, uri)
        return {"status": "deleted", "deleted": deleted, "preview": preview}

    async def _delete_by_filters(self, filters: dict[str, Any], uri: str) -> int:
        """按 filters 删除记忆并同步清理 KB 文档记录

        Args:
            filters: metadata 过滤条件
            uri: 用于日志的记忆 URI

        Returns:
            实际删除的记录数
        """
        doc_ids: list[str] = []
        uris: list[str] = []
        deleted = 0
        collection_ok = False
        logger.debug(
            "[简单长期记忆] 删除开始: uri=%s, filters=%s",
            uri,
            _debug_filter_summary(filters),
        )
        try:
            doc_ids, uris, deleted = await self._collect_kb_doc_ids_for_filters(filters)
            collection_ok = True
            logger.debug(
                "[简单长期记忆] 删除候选收集完成: uri=%s, records=%s, "
                "kb_doc_ids=%s, uris=%s",
                uri,
                deleted,
                len(doc_ids),
                len(uris),
            )
        except Exception as e:
            logger.warning(f"[简单长期记忆] 查询待删除文档失败: {e}")

        if not collection_ok:
            # 收集失败时禁止进入删除，防止无法跟踪删除结果
            logger.warning("[简单长期记忆] 候选收集失败，跳过删除")
            return 0

        await self.vec_db.delete_documents(metadata_filters=filters)
        logger.debug(
            "[简单长期记忆] 删除向量完成: uri=%s, records=%s",
            uri,
            deleted,
        )
        # 同步删除 KB 文档记录
        try:
            await self._unregister_kb_documents(doc_ids)
            await self._sync_kb_stats()
            logger.debug(
                "[简单长期记忆] 删除 KB 注销完成: uri=%s, kb_doc_ids=%s",
                uri,
                len(doc_ids),
            )
        except Exception as e:
            logger.warning(f"[简单长期记忆] KB 文档删除失败: {e}")

        # 级联清理关联边
        if self._link_manager and uris:
            await self._link_manager.delete_links_for_uris(uris)
            logger.debug(
                "[简单长期记忆] 删除关联清理完成: uri=%s, affected_uris=%s",
                uri,
                len(uris),
            )
        logger.info(f"[简单长期记忆] 删除记忆: {uri}, 实际删除 {deleted} 条")
        return deleted

    async def _collect_kb_doc_ids_for_filters(
        self,
        filters: dict[str, Any],
    ) -> tuple[list[str], list[str], int]:
        """分页收集匹配向量文档的 KB 文档 ID 和 URI，避免固定上限截断。"""
        try:
            page_size = int(self.config.get("memory_delete_scan_page_size", 1000))
        except (TypeError, ValueError):
            page_size = 1000
        page_size = max(1, page_size)

        doc_ids: list[str] = []
        uris: list[str] = []
        count = 0
        offset = 0
        while True:
            docs = await self.vec_db.document_storage.get_documents(
                metadata_filters=filters,
                offset=offset,
                limit=page_size,
            )
            if not docs:
                break
            count += len(docs)
            offset += len(docs)
            for doc in docs:
                md = _safe_parse_metadata(doc.get("metadata", {}))
                if md.get("kb_doc_id"):
                    doc_ids.append(md["kb_doc_id"])
                if md.get("uri"):
                    uris.append(md["uri"])
            if len(docs) < page_size:
                break

        return list(dict.fromkeys(doc_ids)), list(dict.fromkeys(uris)), count

    async def clear_memories(
        self,
        event: AstrMessageEvent,
        domain: str | None = None,
        all_users: bool = False,
    ) -> int:
        """清空记忆

        Args:
            event: 消息事件
            domain: 仅清空指定域的记忆（可选）
            all_users: 为 True 时清空所有用户记忆

        Returns:
            删除的记忆数量
        """
        filters = self._build_query_filter(
            event,
            all_users=all_users,
            domain=domain,
            include_deprecated=True,
        )
        scope = "全部" if all_users else filters.get("user_id", "unknown")
        return await self._clear_by_filters(filters, scope_label=scope)

    async def count_clear_memories(
        self,
        event: AstrMessageEvent,
        domain: str | None = None,
        all_users: bool = False,
    ) -> int:
        """统计 clear_memories 将影响的记录数。"""
        filters = self._build_query_filter(
            event,
            all_users=all_users,
            domain=domain,
            include_deprecated=True,
        )
        return await self.vec_db.count_documents(metadata_filter=filters)

    async def _clear_by_filters(
        self, filters: dict[str, Any], *, scope_label: str
    ) -> int:
        """底层清空逻辑：查询 doc_ids → 删除 → 反注册 → 同步统计"""
        doc_ids: list[str] = []
        uris: list[str] = []
        count = 0
        try:
            doc_ids, uris, count = await self._collect_kb_doc_ids_for_filters(filters)
        except Exception as e:
            logger.warning(f"[简单长期记忆] 查询待清空文档失败: {e}")
            try:
                count = await self.vec_db.count_documents(metadata_filter=filters)
            except Exception as ce:
                logger.warning(f"[简单长期记忆] 统计待清空文档失败: {ce}")

        await self.vec_db.delete_documents(metadata_filters=filters)

        try:
            await self._unregister_kb_documents(doc_ids)
            await self._sync_kb_stats()
        except Exception as e:
            logger.warning(f"[简单长期记忆] KB 文档批量删除失败: {e}")

        # 级联清理关联边
        if self._link_manager and uris:
            await self._link_manager.delete_links_for_uris(uris)
        logger.info(f"[简单长期记忆] 清空 {count} 条记忆, 范围: {scope_label}")
        return count

    async def list_memories(
        self,
        event: AstrMessageEvent,
        domain: str | None = None,
        page: int = 1,
        page_size: int = 10,
        all_users: bool = False,
    ) -> tuple[list[dict[str, Any]], int, bool]:
        """列出用户的记忆（分页）

        Args:
            event: 消息事件
            domain: 记忆域过滤（可选）
            page: 页码（从 1 开始）
            page_size: 每页数量
            all_users: 为 True 时跳过用户过滤

        Returns:
            (记忆列表, 总数, 总数是否被扫描上限截断)
        """
        truncated = False
        if all_users:
            filters: dict[str, Any] = {
                "is_memory_record": True,
                "deprecated": False,
            }
            if domain:
                filters["domain"] = domain
            total = await self.vec_db.count_documents(metadata_filter=filters)
            offset = (page - 1) * page_size
            docs = await self.vec_db.document_storage.get_documents(
                metadata_filters=filters,
                offset=offset,
                limit=page_size,
            )
        else:
            parsed = UMOInfo.parse(event.unified_msg_origin)
            offset = (page - 1) * page_size
            if parsed.session_type == "group":
                docs, total, truncated = await self._list_visible_user_documents(
                    event, domain, page=page, page_size=page_size
                )
            else:
                filters = self._build_user_filter(event)
                filters["deprecated"] = False
                if domain:
                    filters["domain"] = domain
                total = await self.vec_db.count_documents(metadata_filter=filters)
                docs = await self.vec_db.document_storage.get_documents(
                    metadata_filters=filters,
                    offset=offset,
                    limit=page_size,
                )

        memories = []
        for doc in docs:
            metadata = _safe_parse_metadata(doc.get("metadata", {}))

            memories.append(
                {
                    "text": doc.get("text", ""),
                    "metadata": metadata,
                }
            )

        return memories, total, truncated

    async def _list_visible_user_documents(
        self,
        event: AstrMessageEvent,
        domain: str | None = None,
        *,
        page: int = 1,
        page_size: int = 10,
    ) -> tuple[list[dict[str, Any]], int, bool]:
        parsed = UMOInfo.parse(event.unified_msg_origin)
        scan_limit = self._memory_list_scan_limit(page, page_size)
        source_filters = [
            self._scope_filter(event, MemoryScope.PERSONAL),
            {
                "memory_scope": MemoryScope.PERSONAL,
                "owner_session_id": build_session_id(
                    parsed.platform_id, parsed.session_id
                ),
                "visibility": MemoryVisibility.GROUP,
                "is_memory_record": True,
                "deprecated": False,
            },
            self._scope_filter(event, MemoryScope.GROUP),
            self._scope_filter(event, MemoryScope.CONVERSATION),
        ]
        if domain:
            for filters in source_filters:
                filters["domain"] = domain

        docs_by_source = await asyncio.gather(
            *(
                self.vec_db.document_storage.get_documents(
                    metadata_filters=filters,
                    limit=scan_limit + 1,
                )
                for filters in source_filters
            )
        )

        visible = []
        seen = set()
        scanned_all_sources = True
        for docs in docs_by_source:
            if len(docs) > scan_limit:
                scanned_all_sources = False
            for doc in docs[:scan_limit]:
                metadata = _safe_parse_metadata(doc.get("metadata", {}))
                if not self._is_visible_shared_personal(event, metadata):
                    continue
                uri = metadata.get("uri") or doc.get("text", "")
                if uri in seen:
                    continue
                visible.append(doc)
                seen.add(uri)

        truncated = not scanned_all_sources
        total = len(visible) if not truncated else max(len(visible), scan_limit + 1)
        offset = (page - 1) * page_size
        return visible[offset : offset + page_size], total, truncated

    def _memory_list_scan_limit(self, page: int, page_size: int) -> int:
        try:
            configured = int(self.config.get("max_memory_list_scan", 200))
        except (TypeError, ValueError):
            configured = 200
        configured = max(1, configured)
        needed = max(1, page) * max(1, page_size)
        return min(configured, needed)

    async def get_memory_by_uri(
        self,
        event: AstrMessageEvent,
        uri: str,
    ) -> dict[str, Any] | None:
        """通过 URI 获取记忆

        Args:
            event: 消息事件
            uri: 记忆 URI

        Returns:
            记忆数据或 None
        """
        filters = self._build_user_filter(event)
        filters["uri"] = uri

        docs = await self.vec_db.document_storage.get_documents(
            metadata_filters=filters,
            limit=1,
        )

        if not docs:
            return None

        doc = docs[0]
        metadata = _safe_parse_metadata(doc.get("metadata", {}))

        return {
            "text": doc.get("text", ""),
            "metadata": metadata,
        }

    async def smart_update_memory(
        self,
        event: AstrMessageEvent,
        content: str,
        domain: str,
        similarity_threshold: float = SIMILARITY_THRESHOLD,
    ) -> str:
        """智能更新记忆：相似则合并，否则新建

        Args:
            event: 消息事件
            content: 新的记忆内容
            domain: 记忆域
            similarity_threshold: 相似度阈值

        Returns:
            操作结果描述
        """
        # 检索相似记忆
        candidates = await self.recall_memories(
            event=event,
            query=content,
            domain=domain,
            top_k=5,
        )

        # 找到最相似的候选
        best_match = None
        best_score = 0.0

        for result in candidates:
            score = result.get("similarity", 0)
            if score > best_score and score >= similarity_threshold:
                best_score = score
                best_match = result

        if best_match:
            # 高相似度：创建新版本（简化处理，不自动合并）
            old_uri = best_match["metadata"].get("uri", "")
            logger.info(
                f"[简单长期记忆] 发现相似记忆: {old_uri}, 相似度: {best_score:.2f}"
            )
            # 返回提示，让调用方决定是否合并
            return f"found_similar:{old_uri}:{best_score:.2f}"
        else:
            # 低相似度：创建新记忆
            uri = await self.store_memory(
                event=event,
                content=content,
                domain=domain,
                uri=str(MemoryURI.generate(domain)),
                memory_type=MemoryType.NORMAL,
            )
            return f"created:{uri}"

    async def get_memory_stats(
        self,
        event: AstrMessageEvent,
        all_users: bool = False,
    ) -> dict[str, int]:
        """获取记忆统计

        Args:
            event: 消息事件
            all_users: 为 True 时统计全局数据

        Returns:
            统计信息字典
        """
        query_event = None if all_users else event
        filters = self._build_query_filter(query_event, all_users=all_users)

        # 总数
        total = await self.vec_db.count_documents(metadata_filter=filters)

        # 永久记忆数
        permanent_filters = {**filters, "memory_type": MemoryType.PERMANENT}
        permanent = await self.vec_db.count_documents(metadata_filter=permanent_filters)

        # 普通记忆数
        normal_filters = {**filters, "memory_type": MemoryType.NORMAL}
        normal = await self.vec_db.count_documents(metadata_filter=normal_filters)

        # 已压缩数
        compressed_filters = self._build_query_filter(
            query_event,
            all_users=all_users,
            include_deprecated=True,
        )
        compressed_filters["compressed"] = True
        compressed = await self.vec_db.count_documents(
            metadata_filter=compressed_filters
        )

        return {
            "total": total,
            "permanent": permanent,
            "normal": normal,
            "compressed": compressed,
        }

    @staticmethod
    def _prepare_faiss_index(faiss_index: Any) -> dict[str, Any]:
        """准备 FAISS 索引及外部 doc ID 到内部 position 的映射。"""
        if faiss_index is None:
            return {"mode": "index_unavailable", "error": "index is None"}
        try:
            import faiss
        except Exception as exc:
            return {
                "mode": "faiss_unavailable",
                "error": f"{type(exc).__name__}: {exc}",
            }

        index_type = type(faiss_index).__name__
        common = {
            "faiss": faiss,
            "index_type": index_type,
            "ntotal": int(getattr(faiss_index, "ntotal", 0) or 0),
        }
        if not (hasattr(faiss_index, "id_map") and hasattr(faiss_index, "index")):
            return {**common, "mode": "direct"}

        try:
            external_ids = faiss.vector_to_array(faiss_index.id_map)
            positions: dict[int, int] = {}
            duplicate_ids = 0
            for position, external_id in enumerate(external_ids.tolist()):
                key = int(external_id)
                if key in positions:
                    duplicate_ids += 1
                    continue
                positions[key] = position
            return {
                **common,
                "mode": "id_map",
                "positions": positions,
                "duplicate_ids": duplicate_ids,
            }
        except Exception as exc:
            return {
                **common,
                "mode": "index_unavailable",
                "error": f"{type(exc).__name__}: {exc}",
            }

    @staticmethod
    def _reconstruct_faiss_vector(
        faiss_index: Any,
        doc_int_id: int,
        prepared: dict[str, Any],
    ) -> tuple[list[float] | None, str, str]:
        """按 FAISS 索引类型重建单条向量并返回诊断状态。"""
        mode = prepared.get("mode")
        if mode in {"index_unavailable", "faiss_unavailable"}:
            return None, mode, str(prepared.get("error", ""))
        try:
            if mode == "id_map":
                position = prepared.get("positions", {}).get(int(doc_int_id))
                if position is None:
                    return None, "id_not_found", f"doc_id={doc_int_id}"
                base_index = prepared["faiss"].downcast_index(faiss_index.index)
                vector = base_index.reconstruct(int(position))
            else:
                vector = faiss_index.reconstruct(int(doc_int_id))
            import numpy as np

            values = np.asarray(vector, dtype=np.float32).reshape(-1)
            if values.size == 0:
                return None, "invalid_vector", "empty vector"
            if not np.all(np.isfinite(values)):
                return None, "invalid_vector", "non-finite value"
            return values.tolist(), "ok", ""
        except Exception as exc:
            return None, "reconstruct_error", f"{type(exc).__name__}: {exc}"

    async def get_all_active_memories(
        self,
        owner_filter: dict[str, Any] | None = None,
        limit: int = 100,
        offset: int = 0,
    ) -> list[dict[str, Any]]:
        """拉取活跃记忆（后台整理用，无 event 场景，支持分页）

        Args:
            owner_filter: 限定用户范围（如 {"user_id": "xxx"}），None 表示全部
            limit: 每页条数
            offset: 分页偏移

        Returns:
            记忆列表，每条包含 uri / content / metadata
        """
        filters: dict[str, Any] = {
            "is_memory_record": True,
            "deprecated": False,
        }
        if owner_filter:
            filters.update(owner_filter)

        stats: dict[str, Any] = {
            "offset": offset,
            "limit": limit,
            "documents": 0,
            "vectors_loaded": 0,
            "vectors_missing": 0,
            "missing_by_reason": {},
        }
        try:
            docs = await self.vec_db.document_storage.get_documents(
                metadata_filters=filters,
                limit=limit,
                offset=offset,
            )
            stats["documents"] = len(docs)
            embedding_storage = getattr(self.vec_db, "embedding_storage", None)
            faiss_index = (
                getattr(embedding_storage, "index", None)
                if embedding_storage is not None
                else None
            )
            prepared = self._prepare_faiss_index(faiss_index)
            stats.update(
                {
                    "index_type": prepared.get("index_type", ""),
                    "index_mode": prepared.get("mode", "index_unavailable"),
                    "index_ntotal": prepared.get("ntotal", 0),
                    "index_duplicate_ids": prepared.get("duplicate_ids", 0),
                }
            )

            memories = []
            for doc in docs:
                metadata = _safe_parse_metadata(doc.get("metadata", {}))
                vector = None
                doc_int_id = doc.get("id")
                if doc_int_id is None:
                    status, detail = "document_id_missing", "document id is None"
                else:
                    vector, status, detail = self._reconstruct_faiss_vector(
                        faiss_index, int(doc_int_id), prepared
                    )
                if vector is None:
                    stats["vectors_missing"] += 1
                    missing = stats["missing_by_reason"]
                    missing[status] = missing.get(status, 0) + 1
                    logger.debug(
                        "[简单长期记忆] 活跃记忆向量不可用: doc_id=%s, status=%s, detail=%s",
                        doc_int_id,
                        status,
                        detail,
                    )
                else:
                    stats["vectors_loaded"] += 1
                memories.append(
                    {
                        "uri": metadata.get("uri", ""),
                        "content": doc.get("text", ""),
                        "metadata": metadata,
                        "vector": vector,
                    }
                )
            self._last_active_memory_stats = stats
            logger.debug(
                "[简单长期记忆] 活跃记忆分页完成: filters=%s, stats=%s",
                filters,
                stats,
            )
            return memories
        except Exception as exc:
            stats["error"] = f"{type(exc).__name__}: {exc}"
            self._last_active_memory_stats = stats
            logger.exception(
                "[简单长期记忆] 拉取活跃记忆失败: filters=%s, stats=%s",
                filters,
                stats,
            )
            return []

    async def deprecate_memory(self, uri: str, reason: str = "") -> bool:
        """标记单条记忆为 deprecated（delete + re-insert 保持 doc/vector 一致）。

        Args:
            uri: 记忆 URI
            reason: 废弃原因

        Returns:
            是否成功
        """
        try:
            old_docs = await self.vec_db.document_storage.get_documents(
                metadata_filters={"uri": uri, "is_memory_record": True},
                limit=1,
            )
            if not old_docs:
                logger.warning(f"[简单长期记忆] deprecate 找不到记忆: {uri}")
                return False

            doc = old_docs[0]
            old_meta = _safe_parse_metadata(doc.get("metadata", {}))
            old_meta["deprecated"] = True
            old_meta["deprecated_at"] = datetime.now(timezone.utc).isoformat()
            old_meta["deprecated_reason"] = reason or "deprecated"

            # 先插入新记录（原子性：插入失败时原记录不受影响）
            new_doc_id = str(uuid.uuid4())
            old_meta["kb_doc_id"] = new_doc_id
            text = doc.get("text", old_meta.get("memory_content", ""))
            await self.vec_db.insert(
                content=text,
                metadata=old_meta,
                id=new_doc_id,
            )

            # 插入成功后同步 KBDocument（注册新 ID，注销旧 ID）
            old_doc_id = doc.get("doc_id", "")
            try:
                await self._register_kb_document(new_doc_id, uri, len(text))
            except Exception:
                pass
            if old_doc_id:
                try:
                    await self._unregister_kb_documents([old_doc_id])
                except Exception:
                    pass
                try:
                    await self.vec_db.delete_documents(
                        metadata_filters={"kb_doc_id": old_doc_id}
                    )
                except Exception as del_err:
                    # 删除失败 → 原记录仍活跃，返回失败让调用方处理
                    logger.warning(
                        f"[简单长期记忆] deprecate 删除旧记录失败: {del_err}"
                    )
                    return False

            logger.debug(f"[简单长期记忆] 已标记 deprecated: {uri}")
            return True
        except Exception as e:
            logger.warning(f"[简单长期记忆] deprecate 失败: {uri}, {e}")
            return False

    async def merge_memories(
        self,
        source_uris: list[str],
        merged_content: str,
        reason: str = "",
        created_by: str = "organizer",
    ) -> dict[str, Any]:
        """合并多条记忆（supersede 语义，不物理删除）

        1. 新建一条合并后的记忆
        2. 旧记忆标 deprecated + deprecated_at
        3. memory_links 写入 supersedes 边
        4. 旧记忆由 purge 统一物理清除

        Args:
            source_uris: 待合并的旧记忆 URI 列表
            merged_content: 合并后的内容
            reason: 合并原因
            created_by: 创建者（默认 organizer）

        Returns:
            {"merged_uri": str, "deprecated_count": int, "success": bool}
        """
        if not source_uris or not merged_content:
            return {"merged_uri": "", "deprecated_count": 0, "success": False}

        try:
            # 0. 预验证：所有源必须存在且未 deprecated，否则不做任何修改
            for pre_uri in source_uris:
                pre_docs = await self.vec_db.document_storage.get_documents(
                    metadata_filters={
                        "uri": pre_uri,
                        "is_memory_record": True,
                        "deprecated": False,
                    },
                    limit=1,
                )
                if not pre_docs:
                    logger.warning(
                        f"[简单长期记忆] 合并预验证失败: {pre_uri} 不存在或已 deprecated"
                    )
                    return {"merged_uri": "", "deprecated_count": 0, "success": False}

            # 1. 新建合并后的记忆（使用第一条旧记忆的 metadata 作为模板）
            first_doc = await self.vec_db.document_storage.get_documents(
                metadata_filters={"uri": source_uris[0], "is_memory_record": True},
                limit=1,
            )
            if not first_doc:
                logger.warning(
                    f"[简单长期记忆] 合并失败：找不到源记忆 {source_uris[0]}"
                )
                return {"merged_uri": "", "deprecated_count": 0, "success": False}

            first_meta = _safe_parse_metadata(first_doc[0].get("metadata", {}))
            new_uri = f"{first_meta.get('uri', '').split('://')[0]}://merged_{uuid.uuid4().hex[:12]}"

            # 构建新记忆的 metadata（继承旧记忆的域、作用域等）
            new_metadata = {
                **first_meta,
                "uri": new_uri,
                "memory_content": merged_content,
                "created_at": datetime.now(timezone.utc).isoformat(),
                "curated_at": datetime.now(timezone.utc).isoformat(),
                "curated_by": created_by,
                "created_by": created_by,
                "deprecated": False,
                "merged_from": source_uris,
                "merged_reason": reason,
            }

            # 写入新记忆（通过 vec_db.insert 自动生成 embedding）
            new_doc_id = str(uuid.uuid4())
            new_metadata["kb_doc_id"] = new_doc_id
            new_metadata["is_memory_record"] = True
            new_metadata["chunk_index"] = 0
            formatted = format_memory_content(merged_content, new_metadata)
            await self.vec_db.insert(
                content=formatted,
                metadata=new_metadata,
                id=new_doc_id,
            )
            # 注册 KB 文档（与其他写入路径一致）
            try:
                await self._register_kb_document(new_doc_id, new_uri, len(formatted))
                await self._sync_kb_stats()
            except Exception:
                pass

            # 2. 旧记忆标 deprecated（全部成功才报成功，否则回滚新记录）
            deprecated_count = 0
            failed_uris: list[str] = []
            for old_uri in source_uris:
                try:
                    ok = await self.deprecate_memory(old_uri, reason="merged")
                    if ok:
                        deprecated_count += 1
                        # 3. 写 supersedes 边
                        if self._link_manager:
                            await self._link_manager.add_link(
                                source_uri=new_uri,
                                target_uri=old_uri,
                                relation_type="supersedes",
                                reason=reason,
                                confidence=1.0,
                                created_by=created_by,
                            )
                    else:
                        failed_uris.append(old_uri)
                except Exception as e:
                    logger.warning(
                        f"[简单长期记忆] 标记旧记忆 deprecated 失败: {old_uri}, {e}"
                    )
                    failed_uris.append(old_uri)

            if failed_uris:
                # 回滚：删除合并记录 + 恢复已 deprecate 的源
                logger.warning(
                    f"[简单长期记忆] 合并回滚: {len(failed_uris)} 条源未能 deprecated"
                )
                try:
                    await self.vec_db.delete_documents(
                        metadata_filters={"kb_doc_id": new_doc_id}
                    )
                except Exception:
                    pass
                # 恢复已成功 deprecate 的源（去掉 deprecated 标记）
                for ok_uri in source_uris:
                    if ok_uri not in failed_uris:
                        try:
                            dep_docs = await self.vec_db.document_storage.get_documents(
                                metadata_filters={
                                    "uri": ok_uri,
                                    "is_memory_record": True,
                                    "deprecated": True,
                                },
                                limit=1,
                            )
                            if dep_docs:
                                dep_meta = _safe_parse_metadata(
                                    dep_docs[0].get("metadata", {})
                                )
                                dep_meta["deprecated"] = False
                                dep_meta.pop("deprecated_at", None)
                                dep_meta.pop("deprecated_reason", None)
                                restore_id = str(uuid.uuid4())
                                dep_meta["kb_doc_id"] = restore_id
                                await self.vec_db.insert(
                                    content=dep_docs[0].get("text", ""),
                                    metadata=dep_meta,
                                    id=restore_id,
                                )
                                # 同步 KB：注册恢复的 ID，注销 deprecated 的 ID
                                try:
                                    await self._register_kb_document(
                                        restore_id,
                                        ok_uri,
                                        len(dep_docs[0].get("text", "")),
                                    )
                                except Exception:
                                    pass
                                old_id = dep_docs[0].get("doc_id", "")
                                if old_id:
                                    try:
                                        await self._unregister_kb_documents([old_id])
                                    except Exception:
                                        pass
                                    await self.vec_db.delete_documents(
                                        metadata_filters={"kb_doc_id": old_id}
                                    )
                        except Exception as restore_err:
                            logger.warning(
                                f"[简单长期记忆] 恢复源失败: {ok_uri}, {restore_err}"
                            )
                # 清理回滚产生的 supersedes 边
                if self._link_manager:
                    try:
                        await self._link_manager.delete_links_for_uris([new_uri])
                    except Exception:
                        pass
                return {
                    "merged_uri": "",
                    "deprecated_count": 0,
                    "success": False,
                }

            logger.info(
                f"[简单长期记忆] 合并完成: {len(source_uris)} 条 → {new_uri}, "
                f"deprecated {deprecated_count} 条"
            )
            return {
                "merged_uri": new_uri,
                "deprecated_count": deprecated_count,
                "success": True,
            }

        except Exception as e:
            logger.error(f"[简单长期记忆] 合并记忆失败: {e}")
            return {"merged_uri": "", "deprecated_count": 0, "success": False}

    async def forget_memory_by_user(
        self,
        event: AstrMessageEvent,
        uri: str,
        target_user_id: str,
    ) -> int:
        """按 user_id + uri 删除指定用户的记忆

        Args:
            event: 消息事件
            uri: 记忆 URI
            target_user_id: 目标用户 ID

        Returns:
            实际删除的记录数
        """
        filters: dict[str, Any] = {
            "user_id": target_user_id,
            "uri": uri,
            "is_memory_record": True,
        }
        return await self._delete_by_filters(filters, uri)

    async def clear_memories_by_user(
        self,
        event: AstrMessageEvent,
        target_user_id: str,
        domain: str | None = None,
    ) -> int:
        """按 user_id 清空指定用户全部记忆

        Args:
            event: 消息事件
            target_user_id: 目标用户 ID
            domain: 仅清空指定域的记忆（可选）

        Returns:
            删除的记忆数量
        """
        filters: dict[str, Any] = {
            "user_id": target_user_id,
            "is_memory_record": True,
        }
        if domain:
            filters["domain"] = domain
        return await self._clear_by_filters(
            filters, scope_label=f"管理员清空用户 {target_user_id}"
        )

    async def count_clear_memories_by_user(
        self,
        event: AstrMessageEvent,
        target_user_id: str,
        domain: str | None = None,
    ) -> int:
        """统计 clear_memories_by_user 将影响的记录数。"""
        filters: dict[str, Any] = {
            "user_id": target_user_id,
            "is_memory_record": True,
        }
        if domain:
            filters["domain"] = domain
        return await self.vec_db.count_documents(metadata_filter=filters)

    async def count_kb_memory_records_by_name(self, kb_name: str) -> int:
        """统计指定知识库中的插件记忆记录数。"""
        kb = await self.kb_mgr.get_kb_by_name(kb_name)
        if not kb:
            return 0
        return await kb.vec_db.count_documents(
            metadata_filter={"is_memory_record": True, "kb_id": kb.kb.kb_id}
        )

    async def kb_exists(self, kb_name: str) -> bool:
        """检查知识库是否存在。"""
        return bool(await self.kb_mgr.get_kb_by_name(kb_name))

    async def get_kb_id_by_name(self, kb_name: str | None) -> str:
        """按知识库名称获取稳定 KB ID。"""
        if not kb_name:
            return ""
        kb = await self.kb_mgr.get_kb_by_name(kb_name)
        return kb.kb.kb_id if kb else ""

    async def _resolve_rebuild_target_kb(
        self,
        rebuild_context: dict[str, Any] | None,
    ) -> KBHelper | None:
        """从重建上下文解析恢复目标，优先使用稳定 KB ID。"""
        if not rebuild_context:
            return self._kb_helper

        target_kb_id = rebuild_context.get("target_kb_id")
        if target_kb_id:
            resolved_by_id = await self.kb_mgr.get_kb(target_kb_id)
            if resolved_by_id:
                return resolved_by_id

        target_kb_name = rebuild_context.get("target_kb_name")
        if target_kb_name:
            return await self.kb_mgr.get_kb_by_name(target_kb_name)

        return self._kb_helper

    async def _resume_rebuild_from_snapshot(
        self,
        memory_records: list[dict[str, Any]],
        rebuild_context: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        """从 KV 快照恢复中断的重建

        将快照中的记忆重新写入当前 KB（原地重建清空后或迁移目标已写入的场景）。
        写入失败的记录会保留，供下次启动继续恢复。

        Args:
            memory_records: 从 KV 恢复的记忆记录列表

        Returns:
            {"success": int, "failed": int, "remaining_records": list}
        """
        logger.debug(
            "[简单长期记忆] 重建恢复开始: records=%s, context_present=%s, "
            "current_kb=%s",
            len(memory_records),
            bool(rebuild_context),
            self._kb_name or "(unknown)",
        )
        if not memory_records:
            return {
                "success": 0,
                "failed": 0,
                "remaining_records": list(memory_records or []),
            }

        target_kb = await self._resolve_rebuild_target_kb(rebuild_context)
        if not target_kb:
            target_label = (
                (rebuild_context or {}).get("target_kb_id")
                or (rebuild_context or {}).get("target_kb_name")
                or self._kb_name
                or "unknown"
            )
            logger.warning(f"[简单长期记忆] 快照恢复目标知识库不存在: {target_label}")
            return {
                "success": 0,
                "failed": len(memory_records),
                "remaining_records": list(memory_records),
                "error": f"目标知识库不存在: {target_label}",
            }

        expected_target_kb_id = (rebuild_context or {}).get("target_kb_id")
        if expected_target_kb_id and target_kb.kb.kb_id != expected_target_kb_id:
            logger.warning(
                "[简单长期记忆] 快照恢复目标知识库 ID 不匹配: "
                f"expected={expected_target_kb_id}, actual={target_kb.kb.kb_id}"
            )
            return {
                "success": 0,
                "failed": len(memory_records),
                "remaining_records": list(memory_records),
                "error": f"目标知识库 ID 不匹配: {target_kb.kb.kb_name}",
            }

        logger.debug(
            "[简单长期记忆] 重建恢复目标已确定: target_kb=%s, target_kb_id=%s, "
            "records=%s",
            target_kb.kb.kb_name,
            target_kb.kb.kb_id,
            len(memory_records),
        )
        success = 0
        failed = 0
        remaining_records: list[dict[str, Any]] = []

        for index, record in enumerate(memory_records, start=1):
            text = record.get("text", "")
            metadata = record.get("metadata", {})
            uri = metadata.get("uri", "")

            if not text:
                logger.warning(
                    "[简单长期记忆] 快照恢复跳过空内容记录: index=%s/%s",
                    index,
                    len(memory_records),
                )
                failed += 1
                continue

            try:
                if uri:
                    existing_doc = await self._find_memory_doc_by_uri(target_kb, uri)
                    if existing_doc:
                        if not await self._repair_kb_document_for_vector_doc(
                            target_kb,
                            existing_doc,
                            uri,
                        ):
                            failed += 1
                            remaining_records.append(record)
                            continue
                        success += 1
                        logger.debug(
                            "[简单长期记忆] 快照恢复复用已有记录: index=%s/%s, "
                            "success=%s, failed=%s",
                            index,
                            len(memory_records),
                            success,
                            failed,
                        )
                        continue

                new_doc_id = str(uuid.uuid4())
                updated_metadata = self._normalize_rebuild_record_metadata(
                    text,
                    metadata,
                    kb_id=target_kb.kb.kb_id,
                    doc_id=new_doc_id,
                )
                content = updated_metadata.get("memory_content", "")
                uri = updated_metadata.get("uri", uri)
                formatted_content = format_memory_content(content, updated_metadata)

                await target_kb.vec_db.insert(
                    content=formatted_content,
                    metadata=updated_metadata,
                    id=new_doc_id,
                )
                await self._ensure_kb_document(
                    new_doc_id,
                    uri,
                    len(formatted_content),
                    kb_helper=target_kb,
                )
                success += 1
                logger.debug(
                    "[简单长期记忆] 快照恢复写入成功: index=%s/%s, success=%s, failed=%s",
                    index,
                    len(memory_records),
                    success,
                    failed,
                )
            except Exception as e:
                logger.warning(
                    "[简单长期记忆] 快照恢复写入失败: index=%s/%s, "
                    "error_type=%s, error=%s",
                    index,
                    len(memory_records),
                    type(e).__name__,
                    e,
                )
                failed += 1
                remaining_records.append(record)

        if success > 0:
            try:
                await self._sync_kb_stats(kb_helper=target_kb)
            except Exception as e:
                logger.warning(f"[简单长期记忆] 快照恢复后统计同步失败: {e}")

        logger.info(f"[简单长期记忆] 快照恢复完成: 成功 {success}, 失败 {failed}")
        logger.debug(
            "[简单长期记忆] 重建恢复结束: total=%s, success=%s, failed=%s, "
            "remaining=%s",
            len(memory_records),
            success,
            failed,
            len(remaining_records),
        )
        return {
            "success": success,
            "failed": failed,
            "remaining_records": remaining_records,
        }

    def _normalize_rebuild_record_metadata(
        self,
        text: str,
        metadata: dict[str, Any],
        *,
        kb_id: str,
        doc_id: str,
    ) -> dict[str, Any]:
        """将重建缓存中的旧记录规范化为当前 metadata 结构。"""
        meta = MemoryMetadata.from_dict(metadata)
        content = meta.memory_content or self._memory_content_from_text(text)
        now = datetime.now(timezone.utc).isoformat()
        created_at = meta.created_at or now
        owner_user_id = meta.owner_user_id or meta.user_id
        owner_user_ids = meta.owner_user_ids or (
            [owner_user_id] if owner_user_id else []
        )

        normalized = {
            **metadata,
            **meta.to_dict(),
            "memory_content": content,
            "created_at": created_at,
            "last_recalled_at": meta.last_recalled_at or created_at,
            "memory_scope": normalize_memory_scope(meta.memory_scope),
            "owner_user_id": owner_user_id,
            "owner_user_ids": owner_user_ids,
            "visibility": normalize_visibility(meta.visibility, meta.memory_scope),
            "speaker_id": meta.speaker_id or meta.sender_id,
            "kb_doc_id": doc_id,
            "kb_id": kb_id,
            "chunk_index": 0,
            "is_memory_record": True,
            "deprecated": False,
        }
        if normalized["memory_scope"] in (MemoryScope.GLOBAL, MemoryScope.GROUP):
            normalized["visibility"] = MemoryVisibility.GROUP
        return normalized

    def _memory_content_from_text(self, text: str) -> str:
        for line in str(text).splitlines():
            if line.startswith("memory: "):
                return line.removeprefix("memory: ").strip()
        return str(text).strip()

    async def rebuild_memories(
        self,
        target_kb_name: str | None = None,
    ) -> dict[str, Any]:
        """重建或迁移所有记忆

        原地重建：拉取所有记忆到本地 → 持久化到 KV → 清空当前 KB → 从本地重新嵌入写入
        迁移模式：拉取所有记忆到本地 → 持久化到 KV → 写入目标 KB → 仅当全部成功时清空源 KB

        重建期间新的写入请求会被缓冲并持久化到 KV，完成后批量语义去重再写入。
        所有中间数据通过 KV 持久化，进程重启后可恢复。

        Args:
            target_kb_name: 目标知识库名称（为 None 时原地重建）

        Returns:
            {"total": int, "success": int, "failed": int,
             "target_kb": str, "is_migration": bool,
             "pending_flushed": int}

        Raises:
            ValueError: 目标知识库不存在
        """
        if self._rebuilding:
            raise RuntimeError("重建/迁移正在进行中，请等待完成")

        # 立即加锁，防止并发竞态；finally 会兜底释放，避免异常路径遗留锁
        self._rebuilding = True

        logger.debug(
            "[简单长期记忆] 重建开始: target_kb=%s, source_kb=%s, mode=%s, "
            "source_connected=%s, pending_before=%s",
            target_kb_name or "(current)",
            self._kb_name or "(unknown)",
            "migration"
            if target_kb_name and target_kb_name != self._kb_name
            else "rebuild",
            self.is_kb_connected,
            len(self._pending_writes),
        )
        try:
            if self._kv_delete:
                for key in (
                    "rebuild_memory_records",
                    "rebuild_context",
                    "rebuild_status",
                ):
                    try:
                        await self._kv_delete(key)
                    except Exception as e:
                        logger.warning(f"[简单长期记忆] 清理旧重建缓存失败: {key}, {e}")

            source_kb = self._kb_helper
            source_kb_name = self._kb_name
            is_migration = (
                target_kb_name is not None and target_kb_name != source_kb_name
            )
            logger.debug(
                "[简单长期记忆] 重建目标解析: source_kb=%s, source_kb_id=%s, "
                "target_kb=%s, mode=%s",
                source_kb_name or "(unknown)",
                getattr(getattr(source_kb, "kb", None), "kb_id", "(unknown)"),
                target_kb_name or "(current)",
                "migration" if is_migration else "rebuild",
            )

            # 解析目标 KB
            if is_migration:
                target_kb = await self.kb_mgr.get_kb_by_name(target_kb_name)
                if not target_kb:
                    all_kbs = await self._list_all_kb_names()
                    available = ", ".join(all_kbs) if all_kbs else "(无)"
                    logger.debug(
                        "[简单长期记忆] 重建目标不存在: target_kb=%s, available_count=%s",
                        target_kb_name,
                        len(all_kbs),
                    )
                    raise ValueError(
                        f"目标知识库 '{target_kb_name}' 不存在。"
                        f"当前可用知识库: {available}"
                    )
                target_existing = await target_kb.vec_db.count_documents(
                    metadata_filter={
                        "is_memory_record": True,
                        "kb_id": target_kb.kb.kb_id,
                    }
                )
                logger.debug(
                    "[简单长期记忆] 迁移目标检查: target_kb=%s, target_kb_id=%s, "
                    "existing_records=%s",
                    target_kb_name,
                    target_kb.kb.kb_id,
                    target_existing,
                )
                if target_existing > 0:
                    return await self._finalize_rebuild(
                        total=0,
                        success=0,
                        failed=0,
                        target_kb_name=target_kb_name,
                        is_migration=True,
                        error=(
                            f"目标知识库 '{target_kb_name}' 已存在 "
                            f"{target_existing} 条记忆记录，请先清空目标或更换知识库"
                        ),
                    )
            else:
                target_kb = source_kb
                target_kb_name = source_kb_name

            logger.debug(
                "[简单长期记忆] 重建目标已确定: target_kb=%s, target_kb_id=%s, mode=%s",
                target_kb_name or "(unknown)",
                getattr(getattr(target_kb, "kb", None), "kb_id", "(unknown)"),
                "migration" if is_migration else "rebuild",
            )
            if self._kv_put:
                await self._kv_put(
                    "rebuild_context",
                    {
                        "source_kb_name": source_kb_name,
                        "source_kb_id": source_kb.kb.kb_id,
                        "target_kb_name": target_kb_name,
                        "target_kb_id": target_kb.kb.kb_id,
                        "is_migration": is_migration,
                    },
                )

            # ── 阶段 1: 拉取所有记忆到本地 + 持久化到 KV ──
            source_doc_ids: list[str] = []
            memory_records: list[dict[str, Any]] = []

            page_size = 5000
            seen_records: set[str] = set()

            async def collect_memory_records(metadata_filters: dict[str, Any]) -> None:
                offset = 0
                while True:
                    page_docs = await source_kb.vec_db.document_storage.get_documents(
                        offset=offset,
                        limit=page_size,
                        metadata_filters=metadata_filters,
                    )
                    if not page_docs:
                        break
                    offset += len(page_docs)
                    logger.debug(
                        "[简单长期记忆] 重建收集分页: filters=%s, page_count=%s, "
                        "offset=%s, unique_records=%s, unique_doc_ids=%s",
                        metadata_filters,
                        len(page_docs),
                        offset,
                        len(memory_records),
                        len(source_doc_ids),
                    )
                    for doc in page_docs:
                        metadata = _safe_parse_metadata(doc.get("metadata", {}))
                        if not metadata.get("uri"):
                            continue
                        record_key = metadata.get("kb_doc_id") or metadata.get("uri")
                        if record_key in seen_records:
                            continue
                        seen_records.add(record_key)
                        old_doc_id = metadata.get("kb_doc_id", "")
                        if old_doc_id:
                            source_doc_ids.append(old_doc_id)
                        memory_records.append(
                            {
                                "text": doc.get("text", ""),
                                "metadata": metadata,
                            }
                        )

            source_kb_id = source_kb.kb.kb_id
            try:
                await collect_memory_records(
                    {"is_memory_record": True, "kb_id": source_kb_id}
                )
                await collect_memory_records(
                    {"deprecated": False, "kb_id": source_kb_id}
                )
            except Exception as e:
                logger.error(f"[简单长期记忆] 读取源知识库文档失败: {e}")
                return await self._finalize_rebuild(
                    total=0,
                    success=0,
                    failed=0,
                    target_kb_name=target_kb_name,
                    is_migration=is_migration,
                    error=f"读取源知识库失败: {e}",
                )

            total = len(memory_records)
            logger.info(
                f"[简单长期记忆] 已拉取 {total} 条记忆到本地, "
                f"模式: {'迁移' if is_migration else '重建'}"
            )
            logger.debug(
                "[简单长期记忆] 重建收集完成: total=%s, source_doc_ids=%s, "
                "deduplicated=%s, source_kb_id=%s",
                total,
                len(source_doc_ids),
                len(seen_records) - total,
                source_kb_id,
            )

            # 安全检查：拉取 0 条但源 KB 有记忆记录时中止，防止误删
            if total == 0:
                source_count = await source_kb.vec_db.count_documents(
                    metadata_filter={"is_memory_record": True, "kb_id": source_kb_id}
                )
                if source_count > 0:
                    return await self._finalize_rebuild(
                        total=0,
                        success=0,
                        failed=0,
                        target_kb_name=target_kb_name,
                        is_migration=is_migration,
                        error=(
                            f"拉取到 0 条记忆，但源知识库中存在 {source_count} 条"
                            "记忆记录，中止重建以防止数据丢失。"
                            "请检查知识库配置后重试"
                        ),
                    )

            # 持久化拉取的数据到 KV（防进程崩溃丢失）
            if self._kv_put:
                await self._kv_put("rebuild_memory_records", memory_records)
                await self._kv_put("rebuild_status", "in_progress")
                logger.debug(
                    "[简单长期记忆] 重建快照已持久化: records=%s, status=%s, "
                    "context=%s",
                    total,
                    "in_progress",
                    bool(self._kv_put),
                )

            # ── 阶段 2: 清空源 KB（原地重建时）或 留待后续清理（迁移时） ──
            if not is_migration:
                logger.debug(
                    "[简单长期记忆] 重建阶段 2 开始: source_kb=%s, records=%s, "
                    "source_doc_ids=%s",
                    source_kb_name,
                    total,
                    len(source_doc_ids),
                )
                try:
                    await self._delete_rebuild_source_records(source_kb, memory_records)
                    if source_doc_ids:
                        await self._unregister_kb_documents(
                            source_doc_ids, kb_helper=source_kb
                        )
                    logger.info("[简单长期记忆] 已清空当前知识库旧记忆")
                    logger.debug(
                        "[简单长期记忆] 重建阶段 2 完成: deleted_memory_records=%s, "
                        "unregistered_doc_ids=%s",
                        total,
                        len(source_doc_ids),
                    )
                except Exception as e:
                    logger.error(f"[简单长期记忆] 清空当前知识库失败: {e}")
                    return await self._finalize_rebuild(
                        total=total,
                        success=0,
                        failed=0,
                        target_kb_name=target_kb_name,
                        is_migration=is_migration,
                        error=f"清空当前知识库失败: {e}",
                    )

            # ── 阶段 3: 从本地缓存写入目标 KB ──
            success = 0
            failed = 0
            logger.debug(
                "[简单长期记忆] 重建阶段 3 开始: target_kb=%s, target_kb_id=%s, "
                "records=%s",
                target_kb_name,
                target_kb.kb.kb_id,
                total,
            )

            for index, record in enumerate(memory_records, start=1):
                text = record["text"]
                metadata = record["metadata"]
                uri = metadata.get("uri", "")

                try:
                    new_doc_id = str(uuid.uuid4())
                    updated_metadata = self._normalize_rebuild_record_metadata(
                        text,
                        metadata,
                        kb_id=target_kb.kb.kb_id,
                        doc_id=new_doc_id,
                    )
                    content = updated_metadata.get("memory_content", "")
                    uri = updated_metadata.get("uri", uri)
                    formatted_content = format_memory_content(content, updated_metadata)

                    await target_kb.vec_db.insert(
                        content=formatted_content,
                        metadata=updated_metadata,
                        id=new_doc_id,
                    )
                    await self._ensure_kb_document(
                        new_doc_id,
                        uri,
                        len(formatted_content),
                        kb_helper=target_kb,
                    )

                    success += 1
                    logger.debug(
                        "[简单长期记忆] 重建写入成功: index=%s/%s, success=%s, failed=%s",
                        index,
                        total,
                        success,
                        failed,
                    )
                except Exception as e:
                    logger.warning(
                        "[简单长期记忆] 写入记忆失败: index=%s/%s, error_type=%s, error=%s",
                        index,
                        total,
                        type(e).__name__,
                        e,
                    )
                    failed += 1
                    logger.debug(
                        "[简单长期记忆] 重建写入失败统计: index=%s/%s, success=%s, failed=%s",
                        index,
                        total,
                        success,
                        failed,
                    )

            logger.debug(
                "[简单长期记忆] 重建阶段 3 完成: total=%s, success=%s, failed=%s",
                total,
                success,
                failed,
            )
            # ── 阶段 4: 迁移模式 — 仅当全部成功时清空源 KB 并切换 ──
            migration_committed = False
            migration_commit_error = ""
            if is_migration:
                logger.debug(
                    "[简单长期记忆] 重建阶段 4 开始: source_kb=%s, target_kb=%s, "
                    "success=%s, failed=%s, source_doc_ids=%s",
                    source_kb_name,
                    target_kb_name,
                    success,
                    failed,
                    len(source_doc_ids),
                )
                if failed == 0:
                    try:
                        # 先导出源关联表（在删除源记录之前）
                        exported_links: list[dict] = []
                        if self._link_manager:
                            exported_links = await self._link_manager.export_all()

                        logger.debug(
                            "[简单长期记忆] 迁移提交准备: exported_links=%s, "
                            "source_doc_ids=%s",
                            len(exported_links),
                            len(source_doc_ids),
                        )
                        await self._delete_rebuild_source_records(
                            source_kb, memory_records
                        )
                        if source_doc_ids:
                            await self._unregister_kb_documents(
                                source_doc_ids, kb_helper=source_kb
                            )
                        self._kb_helper = target_kb
                        self._kb_name = target_kb_name
                        # 目标建表 + 导入关联
                        self._link_manager = MemoryLinkManager(self.vec_db)
                        await self._link_manager.ensure_table()
                        if exported_links:
                            imported = await self._link_manager.import_all(
                                exported_links
                            )
                            if imported != len(exported_links):
                                logger.warning(
                                    f"[简单长期记忆] 关联迁移数量不一致: "
                                    f"导出 {len(exported_links)}, 导入 {imported}"
                                )
                            else:
                                logger.info(f"[简单长期记忆] 迁移关联表: {imported} 条")
                        migration_committed = True
                        logger.info(f"[简单长期记忆] 已迁移到知识库: {target_kb_name}")
                        logger.debug(
                            "[简单长期记忆] 重建阶段 4 完成: committed=%s, "
                            "imported_links=%s",
                            migration_committed,
                            len(exported_links),
                        )
                    except Exception as e:
                        migration_commit_error = f"清理源知识库失败: {e}"
                        logger.error(f"[简单长期记忆] {migration_commit_error}")
                elif failed > 0:
                    logger.warning(
                        f"[简单长期记忆] 存在 {failed} 条写入失败，"
                        "跳过源知识库清理以防止数据丢失"
                    )
                    logger.debug(
                        "[简单长期记忆] 重建阶段 4 跳过源清理: failed=%s, "
                        "reason=target_write_failures",
                        failed,
                    )

            # ── 阶段 5: 同步统计 ──
            logger.debug(
                "[简单长期记忆] 重建阶段 5 开始: source_sync=%s, target_sync=%s",
                is_migration,
                True,
            )
            try:
                if is_migration:
                    await self._sync_kb_stats(kb_helper=source_kb)
                await self._sync_kb_stats(kb_helper=target_kb)
                logger.debug(
                    "[简单长期记忆] 重建阶段 5 完成: source_sync=%s, target_sync=%s",
                    is_migration,
                    True,
                )
            except Exception as e:
                logger.warning(f"[简单长期记忆] 同步统计数据失败: {e}")
                logger.debug(
                    "[简单长期记忆] 重建阶段 5 异常: error_type=%s, error=%s",
                    type(e).__name__,
                    e,
                )

            # ── 阶段 6: 解锁 + 处理缓冲写入 ──
            self._rebuilding = False
            # 迁移未提交时插件仍使用 source KB，缓冲写入必须落到当前活跃 KB
            flush_target = (
                source_kb if is_migration and not migration_committed else None
            )
            pending_flushed = await self._flush_pending_writes(target_kb=flush_target)
            logger.debug(
                "[简单长期记忆] 重建阶段 6 完成: flush_target=%s, pending_flushed=%s, "
                "pending_remaining=%s",
                getattr(getattr(flush_target, "kb", None), "kb_name", None)
                if flush_target
                else "current",
                pending_flushed,
                len(self._pending_writes),
            )

            final_status = "completed" if failed == 0 else "partial"
            if migration_commit_error:
                final_status = "interrupted"
            if self._kv_put:
                await self._kv_put("rebuild_status", final_status)

            # ── 阶段 7: 完整性校验 ──
            verification_expected = success
            if not is_migration or migration_committed:
                verification_expected += pending_flushed
            logger.debug(
                "[简单长期记忆] 重建阶段 7 开始: expected_success=%s, "
                "expected_pending=%s, expected_total=%s",
                success,
                pending_flushed,
                verification_expected,
            )
            verification = await self._verify_rebuild_integrity(
                target_kb,
                total,
                verification_expected,
            )
            logger.debug(
                "[简单长期记忆] 重建阶段 7 完成: expected=%s, actual=%s, diff=%s, "
                "passed=%s",
                verification.get("expected"),
                verification.get("actual"),
                verification.get("diff"),
                verification.get("passed"),
            )

            logger.info(
                f"[简单长期记忆] 重建完成: 总计 {total}, 成功 {success}, "
                f"失败 {failed}, 目标: {target_kb_name}, "
                f"迁移: {is_migration}, 缓冲写入: {pending_flushed}, "
                f"校验: {'通过' if verification['passed'] else '不一致'}"
            )

            return {
                "status": final_status,
                "total": total,
                "success": success,
                "failed": failed,
                "target_kb": target_kb_name,
                "is_migration": is_migration,
                "pending_flushed": pending_flushed,
                "verification": verification,
                "migration_committed": migration_committed,
                "error": migration_commit_error,
            }
        finally:
            self._rebuilding = False

    async def _verify_rebuild_integrity(
        self,
        target_kb: KBHelper,
        expected_total: int,
        expected_count: int,
    ) -> dict[str, Any]:
        """校验目标 KB 中的记忆记录数是否与预期一致

        Args:
            target_kb: 目标 KBHelper
            expected_total: 拉取的记忆总数
            expected_count: 目标 KB 中预期存在的记忆数量

        Returns:
            {"passed": bool, "expected": int, "actual": int, "diff": int}
        """
        logger.debug(
            "[简单长期记忆] 完整性校验开始: target_kb=%s, target_kb_id=%s, "
            "source_total=%s, expected_count=%s",
            getattr(target_kb.kb, "kb_name", "(unknown)"),
            target_kb.kb.kb_id,
            expected_total,
            expected_count,
        )
        try:
            actual = await target_kb.vec_db.count_documents(
                metadata_filter={"is_memory_record": True, "kb_id": target_kb.kb.kb_id}
            )
        except Exception as e:
            logger.warning(f"[简单长期记忆] 完整性校验失败: {e}")
            logger.debug(
                "[简单长期记忆] 完整性校验异常: error_type=%s, error=%s",
                type(e).__name__,
                e,
            )
            return {
                "passed": False,
                "expected": expected_count,
                "actual": -1,
                "diff": -1,
                "error": str(e),
            }

        passed = actual == expected_count
        logger.debug(
            "[简单长期记忆] 完整性校验结果: expected=%s, actual=%s, diff=%s, passed=%s",
            expected_count,
            actual,
            actual - expected_count,
            passed,
        )
        return {
            "passed": passed,
            "expected": expected_count,
            "actual": actual,
            "diff": actual - expected_count,
        }

    async def _finalize_rebuild(
        self,
        total: int,
        success: int,
        failed: int,
        target_kb_name: str,
        is_migration: bool,
        error: str = "",
    ) -> dict[str, Any]:
        """重建失败时安全收尾：解锁 + flush 缓冲写入

        KV 缓存数据保留不清理，等待用户确认后手动清理。

        Returns:
            包含 "status": "interrupted" 的结果字典
        """
        logger.debug(
            "[简单长期记忆] 重建异常收尾开始: total=%s, success=%s, failed=%s, "
            "target_kb=%s, mode=%s, error_type=%s",
            total,
            success,
            failed,
            target_kb_name,
            "migration" if is_migration else "rebuild",
            type(error).__name__ if error else "unknown",
        )
        self._rebuilding = False
        pending_flushed = await self._flush_pending_writes()

        if self._kv_put:
            await self._kv_put("rebuild_status", "interrupted")

        logger.warning(
            f"[简单长期记忆] 重建异常终止: 总计 {total}, 成功 {success}, "
            f"失败 {failed}, 已 flush 缓冲 {pending_flushed} 条"
        )
        logger.debug(
            "[简单长期记忆] 重建异常收尾完成: status=interrupted, "
            "pending_flushed=%s, cache_retained=%s",
            pending_flushed,
            True,
        )

        return {
            "status": "interrupted",
            "total": total,
            "success": success,
            "failed": failed,
            "target_kb": target_kb_name,
            "is_migration": is_migration,
            "pending_flushed": pending_flushed,
            "error": error,
        }

    async def _flush_pending_writes(self, target_kb: KBHelper | None = None) -> int:
        """处理重建期间缓冲的写入请求，语义去重后写入 KB

        Args:
            target_kb: 指定写入目标 KB（迁移未提交时使用 target KB），
                       为 None 时使用当前 self._kb_helper

        Returns:
            成功写入的缓冲条数
        """
        logger.debug(
            "[简单长期记忆] 缓冲写入开始: pending=%s, target_kb=%s, target_kb_id=%s",
            len(self._pending_writes),
            getattr(getattr(target_kb or self._kb_helper, "kb", None), "kb_name", None),
            getattr(getattr(target_kb or self._kb_helper, "kb", None), "kb_id", None),
        )
        if not self._pending_writes:
            logger.debug("[简单长期记忆] 缓冲写入结束: pending=0, flushed=0")
            return 0

        write_kb = target_kb or self._kb_helper
        if not write_kb:
            logger.warning("[简单长期记忆] 缓冲写入结束: 当前没有可写入的知识库")
            return 0

        pending = list(self._pending_writes)
        self._pending_writes.clear()
        flushed = 0
        retry_pending: list[dict[str, Any]] = []

        for item in pending:
            content = item["content"]
            uri = item.get("uri", "")
            try:
                existing_doc = await self._find_memory_doc_by_uri(write_kb, uri)
                if existing_doc:
                    if await self._repair_kb_document_for_vector_doc(
                        write_kb,
                        existing_doc,
                        uri,
                    ):
                        flushed += 1
                        continue
                    retry_pending.append(item)
                    continue

                # 语义去重：召回相似记忆，高相似度则跳过
                memory_scope = normalize_memory_scope(
                    item.get("memory_scope", MemoryScope.PERSONAL)
                )
                filters: dict[str, Any] = {
                    "memory_scope": memory_scope,
                    "domain": item["domain"],
                    "is_memory_record": True,
                    "deprecated": False,
                }
                if memory_scope == MemoryScope.GLOBAL:
                    pass
                elif memory_scope == MemoryScope.GROUP:
                    filters["owner_session_id"] = item.get("owner_session_id", "")
                elif memory_scope == MemoryScope.CONVERSATION:
                    filters["umo"] = item["umo"]
                else:
                    filters["owner_user_id"] = item.get(
                        "owner_user_id", item["user_id"]
                    )
                candidates = await write_kb.vec_db.retrieve(
                    query=content,
                    k=1,
                    metadata_filters=filters,
                )
                if candidates and candidates[0].similarity >= SIMILARITY_THRESHOLD:
                    logger.debug(
                        f"[简单长期记忆] 缓冲写入去重跳过: {item['uri']}, "
                        f"相似度: {candidates[0].similarity:.2f}"
                    )
                    continue

                # 构建完整 metadata 并写入
                now = datetime.now(timezone.utc).isoformat()
                metadata = {
                    "user_id": item["user_id"],
                    "platform_id": item["platform_id"],
                    "sender_id": item["sender_id"],
                    "umo": item["umo"],
                    "session_type": item["session_type"],
                    "session_id": item["session_id"],
                    "created_at": now,
                    "last_recalled_at": now,
                    "recall_count": 0,
                    "compressed": False,
                    "memory_scope": item.get("memory_scope", MemoryScope.PERSONAL),
                    "owner_user_id": item.get("owner_user_id", item["user_id"]),
                    "owner_user_ids": item.get("owner_user_ids", [item["user_id"]]),
                    "owner_session_id": item.get("owner_session_id", ""),
                    "visibility": item.get("visibility", MemoryVisibility.PRIVATE),
                    "speaker_id": item.get("speaker_id", item["sender_id"]),
                    "domain": item["domain"],
                    "uri": item["uri"],
                    "version": 1,
                    "deprecated": False,
                    "memory_type": item["memory_type"],
                    "disclosure": item["disclosure"],
                    "importance": item["importance"],
                    "subject": item.get("subject", ""),
                    "entities": item.get("entities", []),
                    "topics": item.get("topics", []),
                    "memory_content": content,
                }

                doc_id = str(uuid.uuid4())
                metadata["kb_doc_id"] = doc_id
                metadata["kb_id"] = write_kb.kb.kb_id
                metadata["chunk_index"] = 0
                metadata["is_memory_record"] = True

                formatted_content = format_memory_content(content, metadata)
                await write_kb.vec_db.insert(
                    content=formatted_content,
                    metadata=metadata,
                    id=doc_id,
                )

                if not await self._ensure_kb_document(
                    doc_id,
                    item["uri"],
                    len(formatted_content),
                    kb_helper=write_kb,
                ):
                    retry_pending.append(item)
                    continue

                flushed += 1
                logger.debug(f"[简单长期记忆] 缓冲写入已处理: {item['uri']}")
            except Exception as e:
                logger.warning(f"[简单长期记忆] 缓冲写入处理失败: {item['uri']}, {e}")
                retry_pending.append(item)

        if flushed:
            try:
                await self._sync_kb_stats(kb_helper=write_kb)
            except Exception as e:
                logger.warning(f"[简单长期记忆] 缓冲写入后统计同步失败: {e}")

        self._pending_writes = retry_pending
        if retry_pending:
            if self._kv_put:
                await self._kv_put("rebuild_pending_writes", retry_pending)
        elif self._kv_delete:
            try:
                await self._kv_delete("rebuild_pending_writes")
            except Exception:
                pass

        logger.info(
            f"[简单长期记忆] 缓冲写入处理完成: "
            f"共 {len(pending)} 条, 写入 {flushed} 条, "
            f"待重试 {len(retry_pending)} 条"
        )
        logger.debug(
            "[简单长期记忆] 缓冲写入结束: pending=%s, flushed=%s, retry=%s, "
            "kv_pending_persisted=%s",
            len(pending),
            flushed,
            len(retry_pending),
            bool(retry_pending and self._kv_put),
        )
        return flushed

    async def _list_all_kb_names(self) -> list[str]:
        """获取所有知识库名称列表"""
        try:
            kbs = await self.kb_mgr.list_kbs()
            return [kb.kb_name for kb in kbs] if kbs else []
        except Exception:
            return []

    async def clear_rebuild_cache(self) -> dict[str, bool]:
        """清理重建/迁移过程中缓存的 KV 数据

        在确认重建结果无误后调用。

        Returns:
            各 KV 键的清理结果
        """
        result: dict[str, bool] = {}
        keys = [
            "rebuild_memory_records",
            "rebuild_pending_writes",
            "rebuild_status",
            "rebuild_context",
        ]
        logger.debug(
            "[简单长期记忆] 重建缓存清理开始: keys=%s, kv_delete_available=%s",
            keys,
            bool(self._kv_delete),
        )
        for key in keys:
            try:
                if self._kv_delete:
                    await self._kv_delete(key)
                result[key] = True
            except Exception:
                result[key] = False
        logger.info(f"[简单长期记忆] 已清理重建缓存: {result}")
        logger.debug(
            "[简单长期记忆] 重建缓存清理结束: succeeded=%s, failed=%s, result=%s",
            sum(1 for ok in result.values() if ok),
            sum(1 for ok in result.values() if not ok),
            result,
        )
        return result

    def _hash_rebuild_cache_parts(self, *parts: Any) -> str:
        payload = json.dumps(
            parts,
            ensure_ascii=False,
            sort_keys=True,
            default=str,
        )
        return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:12]

    def rebuild_cache_fingerprint(self, status: dict[str, Any]) -> str:
        """生成重建缓存确认指纹。"""
        cached = status.get("fingerprint")
        if cached:
            return str(cached)
        return self._hash_rebuild_cache_parts(
            status.get("memory_records", 0),
            status.get("pending_writes", 0),
            status.get("status"),
            status.get("source_kb"),
            status.get("target_kb"),
            status.get("is_migration"),
        )

    async def get_rebuild_cache_status(self) -> dict[str, Any]:
        """查看重建缓存的当前状态

        Returns:
            缓存状态信息
        """
        status: dict[str, Any] = {}
        if self._kv_get:
            records = await self._kv_get("rebuild_memory_records", None)
            pending = await self._kv_get("rebuild_pending_writes", None)
            rebuild_status = await self._kv_get("rebuild_status", None)
            rebuild_context = await self._kv_get("rebuild_context", None)
            status["memory_records"] = len(records) if records else 0
            status["pending_writes"] = len(pending) if pending else 0
            status["status"] = rebuild_status
            if isinstance(rebuild_context, dict):
                status["source_kb"] = rebuild_context.get("source_kb_name", "")
                status["target_kb"] = rebuild_context.get("target_kb_name", "")
                status["is_migration"] = bool(rebuild_context.get("is_migration"))
            status["fingerprint"] = self._hash_rebuild_cache_parts(
                records or [],
                pending or [],
                rebuild_status,
                rebuild_context or {},
            )
        return status