release: publish Hy Memory 2.0.5.dev4

Co-authored-by: Cursor <cursoragent@cursor.com>

.env.example

@@ -86,6 +86,7 @@ MEMORY_VECTOR_STORE=chroma
 # MEMORY_CODING_DB_PATH=                        # coding_memory.db 路径（默认 {DATA_DIR}/data/coding_memory.db）
 # MEMORY_CODING_TOOL_RESULT_MAX_BYTES=2048      # 单条 tool_result 截断字节上限
 # MEMORY_CODING_VERBATIM_FALLBACK_ENABLED=false # 模型无结果时是否保存原文 fact；默认关，避免 tool trace 外泄
+# MEMORY_CODING_CHAIN_BLOCK_ENABLED=false       # search 是否注入版本演进链块（diff）；默认关，省 token
 # MEMORY_EXTRACT_HISTORY_TURNS=5                 # 除本次 write 外，额外注入 extractor 的历史 user-turn 数
 
 # ── 写入流程开关 ──

CHANGELOG.md

@@ -2,6 +2,40 @@
 
 Hy Memory and MemHub are versioned and released independently.
 
+## Hy Memory 2.0.5.dev4 — 2026-10-21
+
+- Development preview; not the final `2.0.5` release.
+- Adds idempotent `process_raw` / `async_process_raw` APIs that extract and
+  reconcile ordered existing L1 raw IDs without embedding or inserting raw
+  again.
+- Returns durable raw/memory ID receipts for MemHub queue recovery.
+
+## Hy Memory 2.0.5.dev3 — 2026-10-21
+
+- Development preview; not the final `2.0.5` release.
+- Published as
+  [`hy-memory==2.0.5.dev3`](https://pypi.org/project/hy-memory/2.0.5.dev3/).
+- Lets the existing `search` API explicitly query SDK raw with
+  `layers=["l1_raw"]`, including existing source filters.
+
+## Hy Memory 2.0.5.dev2 — 2026-10-21
+
+- Development preview; not the final `2.0.5` release.
+- Published as
+  [`hy-memory==2.0.5.dev2`](https://pypi.org/project/hy-memory/2.0.5.dev2/).
+- Adds exact, idempotent semantic-memory import for legacy MemHub cutover
+  without extraction, reconciliation or identity changes.
+
+## Hy Memory 2.0.5.dev1 — 2026-10-21
+
+- Development preview for the MemHub SDK-authority migration; this is not the
+  final `2.0.5` release.
+- Published as
+  [`hy-memory==2.0.5.dev1`](https://pypi.org/project/hy-memory/2.0.5.dev1/).
+- Adds lifecycle status filtering, durable idempotent write outcomes, exact raw
+  APIs, offline local data-plane snapshot/restore, and fail-closed physical
+  delete receipts.
+
 ## Hy Memory 2.0.3 — 2026-10-17
 
 - Added an optional runtime HTTP transport factory so embedded products can
@@ -14,6 +48,22 @@ Hy Memory and MemHub are versioned and released independently.
 
 ## Unreleased
 
+- `search(..., layers=["l1_raw"])` now explicitly searches SDK raw records in
+  pro/ultra mode and keeps existing user/agent/session/source filters; default
+  search continues to hide raw.
+- Added `import_memory_exact` for idempotent migration of an existing semantic
+  memory with its canonical ID, lifecycle and supersession links, without
+  extraction or reconciliation.
+- Added SDK-authoritative lifecycle listing with explicit status filters,
+  idempotent durable write outcomes, exact raw import/get/list APIs, and
+  quiesced local data-plane snapshot/restore contracts for the MemHub
+  authority migration.
+- Physical delete now reports incomplete VDB/Graph cleanup as a partial failure
+  instead of claiming success after only one store was changed.
+- Coding search no longer injects the version-evolution diff block
+  (`chain_block`) by default. Enable with
+  `MEMORY_CODING_CHAIN_BLOCK_ENABLED=true` or
+  `CodingConfig.chain_block_enabled=True` (write-side observe unchanged).
 - Fixed governed-memory revisions and repeated evidence links failing when the
   visibility dirty-marker trigger encounters the same memory twice. Startup
   now replaces the affected legacy trigger without changing schema versions

README.md

@@ -153,7 +153,7 @@ result = client.search(
 
 result["facts"]          # fact layer
 result["experiences"]    # experience layer
-result["chain_block"]    # coding only: version-evolution context
+result["chain_block"]    # coding only: version-evolution context (empty unless MEMORY_CODING_CHAIN_BLOCK_ENABLED)
 ```
 
 ### 5. Routing, Source Isolation & Advanced APIs

README.zh.md

@@ -153,7 +153,7 @@ result = client.search(
 
 result["facts"]          # 事实层
 result["experiences"]    # 经验层
-result["chain_block"]    # 仅 coding：版本演进上下文
+result["chain_block"]    # 仅 coding：版本演进上下文（默认空；需 MEMORY_CODING_CHAIN_BLOCK_ENABLED）
 ```
 
 ### 5. 自动路由、来源过滤与进阶 API

docs/env_reference.md

@@ -197,6 +197,7 @@ SDK 不设置 Rerank Search 并发上限。开启后模型在 `HyMemoryClient`
 | `MEMORY_HISTORY_DB_PATH` | `{DATA_DIR}/data/history.db` | History SQLite 路径 |
 | `MEMORY_HISTORY_RECORD_SEARCHES` | `false` | 是否记录 Search 操作 |
 | `MEMORY_CODING_CHAIN_REPO_ROOT` | — | coding 版本链读取本地 Git diff 的仓库根目录 |
+| `MEMORY_CODING_CHAIN_BLOCK_ENABLED` | `false` | search(scene=coding) 是否注入版本演进链块（diff / `chain_block`）；默认关 |
 
 ## HTTP Server
 

docs/hy_memory_search_skill.md

@@ -78,7 +78,8 @@ default 多-query 超过 `MEMORY_QUERY_REWRITE_MAX_CHARS` 时不改写：Coding
 Coding 关闭则直接使用最后一条 query。
 
 路由到 `working` / `coding` 后，Search 读取独立生产力记忆库，返回 `facts` /
-`experiences`，不使用 chat 三通道结构。`coding` 还返回 `chain_block` / `chain_used`。
+`experiences`，不使用 chat 三通道结构。`coding` 还返回 `chain_block` / `chain_used`
+（默认空；需 `MEMORY_CODING_CHAIN_BLOCK_ENABLED` 才注入 diff 块）。
 
 ## Chat 成功返回结构
 

docs/scene_coding_usage.md

@@ -168,9 +168,11 @@ working/coding 搜索返回：
 
 `limit` 分别限制 facts 与 experiences，因此两层最多各返回 `limit` 条。
 
-`scene="coding"` 额外返回 `chain_block` 和 `chain_used`：
+`scene="coding"` 额外返回 `chain_block` 和 `chain_used` 字段（键始终存在；默认内容为空）：
 
-- `chain_block` 是与已召回 fact 对齐的版本演进上下文；无法挂链时为 `""`。
+- 注入 diff / 版本演进块需显式开启 `MEMORY_CODING_CHAIN_BLOCK_ENABLED=true`
+  （或 `CodingConfig.chain_block_enabled=True`）。默认关闭，以节约 prompt token。
+- 开启后，`chain_block` 是与已召回 fact 对齐的版本演进上下文；无法挂链时为 `""`。
 - `chain_used` 是已经进入 chain block 的 `facts` 下标。
 - SDK 不会从 `facts` 删除这些条目。调用方若要拼成“chain block + 剩余 facts”，需自行按 `chain_used` 过滤。
 
@@ -222,10 +224,12 @@ working/coding 搜索返回：
 | `MEMORY_CODING_DB_PATH` | `{DATA_DIR}/data/coding_memory.db` | working 基础 SQLite 路径；coding/trace 路径由此派生 |
 | `MEMORY_CODING_TOOL_RESULT_MAX_BYTES` | `2048` | 单条 tool result 的字节上限 |
 | `MEMORY_CODING_VERBATIM_FALLBACK_ENABLED` | `false` | 模型无结果时是否保存原文 fact；默认关闭，避免 assistant/tool trace 直接展示 |
+| `MEMORY_CODING_CHAIN_BLOCK_ENABLED` | `false` | search 是否注入版本演进链块（`chain_block` / diff）；默认关闭（opt-in） |
 | `MEMORY_EXTRACT_HISTORY_TURNS` | `5` | 回读的历史 user-turn 数；当前 prompt 最终最多使用 5 |
 | `MEMORY_CODING_CHAIN_REPO_ROOT` | 空 | coding 版本链读取本地 Git diff 的仓库路径 |
 
 `HCM_*` 变量属于版本链的高级调优项，不是稳定的公共配置契约；普通接入不应依赖其默认值。
+其中 `HCM_BLOCK_OFF=1` 可在已开启 `MEMORY_CODING_CHAIN_BLOCK_ENABLED` 时做消融（完全不出块）。
 
 ## 8. 定向验证
 

docs/usage.md

@@ -210,7 +210,8 @@ result = client.search(
 ```
 
 `working` / `coding` 路径返回 `scene="productivity"`、`facts`、`experiences`，不使用
-chat 的三桶结构；`coding` 还会返回 `chain_block` 和 `chain_used`。
+chat 的三桶结构；`coding` 还会返回 `chain_block` 和 `chain_used` 字段（默认空；
+需 `MEMORY_CODING_CHAIN_BLOCK_ENABLED=true` 才注入 diff 块）。
 
 #### `search_mode`
 

hy_memory/__init__.py

@@ -55,6 +55,13 @@ except Exception:
 from .client import HyMemoryClient
 from .inspector import MemoryInspector
 from .runtime import SharedRuntime
+from .snapshot import (
+    DataPlaneRestore,
+    DataPlaneSnapshot,
+    DataPlaneSnapshotRejected,
+    restore_local_data_plane,
+    snapshot_local_data_plane,
+)
 
 # ====== 配置 ======
 from .config import MemoryConfig
@@ -111,6 +118,11 @@ __all__ = [
     "HyMemoryClient",
     "MemoryInspector",
     "SharedRuntime",
+    "DataPlaneRestore",
+    "DataPlaneSnapshot",
+    "DataPlaneSnapshotRejected",
+    "restore_local_data_plane",
+    "snapshot_local_data_plane",
     # 配置
     "MemoryConfig",
     # Embedding 端点覆盖

hy_memory/coding/chain_store.py

@@ -647,7 +647,9 @@ class HyCodeMem:
         top-k，但它们的 diff 对当前工作区不可能有用，因此挂链前必须做同仓过滤。
         """
         facts = facts or []
-        # 开关: HCM_BLOCK_OFF=1 → 完全不出块, 也不从 bullet 删任何 fact。
+        # 高级消融开关: HCM_BLOCK_OFF=1 → 完全不出块, 也不从 bullet 删任何 fact。
+        # 公共 opt-in 见 CodingConfig.chain_block_enabled /
+        # MEMORY_CODING_CHAIN_BLOCK_ENABLED（默认关；客户端在调用 decorate 前拦截）。
         if os.environ.get("HCM_BLOCK_OFF") == "1":
             return "", []
         picked, used = [], []

hy_memory/config.py

@@ -1136,6 +1136,9 @@ class CodingConfig:
     # LLM 失败或没有抽出 fact 时，是否把对话原文存成 working fact。
     # 默认关闭，避免 assistant/tool 轨迹以 statement 形式直接暴露给消费方。
     verbatim_fallback_enabled: bool = None
+    # search(scene="coding") 是否注入版本演进链块（diff / chain_block）。
+    # 默认关闭（opt-in）：中重度 coding 用户可省下大量 prompt token；写入侧仍可 observe。
+    chain_block_enabled: bool = None
 
     def __post_init__(self):
         if self.enable is None:
@@ -1153,6 +1156,10 @@ class CodingConfig:
             self.verbatim_fallback_enabled = _get_env_bool(
                 "MEMORY_CODING_VERBATIM_FALLBACK_ENABLED", False
             )
+        if self.chain_block_enabled is None:
+            self.chain_block_enabled = _get_env_bool(
+                "MEMORY_CODING_CHAIN_BLOCK_ENABLED", False
+            )
 
 
 @dataclass

hy_memory/data/cache_base.py

@@ -114,6 +114,37 @@ class CacheBase(ABC):
         """查询 pipeline 中间结果日志"""
         ...
 
+    # ================================================================
+    # Durable write outcomes
+    # ================================================================
+
+    @abstractmethod
+    async def claim_write_outcome(
+        self,
+        request_id: str,
+        user_id: str,
+        agent_id: str,
+        session_id: str,
+        operation: str,
+    ) -> Optional[Dict[str, Any]]:
+        """Claim one idempotent write; return the existing row if already claimed."""
+        ...
+
+    @abstractmethod
+    async def finish_write_outcome(
+        self,
+        request_id: str,
+        status: str,
+        result: Dict[str, Any],
+    ) -> bool:
+        """Persist the terminal/unknown result for a claimed write."""
+        ...
+
+    @abstractmethod
+    async def get_write_outcome(self, request_id: str) -> Optional[Dict[str, Any]]:
+        """Read one durable write outcome by exact request id."""
+        ...
+
     # ================================================================
     # Summary Buffer（rolling summary 的未消费 turn 暂存）
     #

hy_memory/data/cache_mysql.py

@@ -81,6 +81,21 @@ _CREATE_TABLES_SQL = [
     )
     """,
     """
+    CREATE TABLE IF NOT EXISTS write_outcomes (
+        request_id VARCHAR(256) NOT NULL,
+        user_id    VARCHAR(255) NOT NULL,
+        agent_id   VARCHAR(255) NOT NULL DEFAULT '',
+        session_id VARCHAR(255) NOT NULL DEFAULT '',
+        operation  VARCHAR(32) NOT NULL,
+        status     VARCHAR(32) NOT NULL,
+        result_json LONGTEXT NOT NULL,
+        created_at VARCHAR(64) NOT NULL,
+        updated_at VARCHAR(64) NOT NULL,
+        PRIMARY KEY (request_id),
+        KEY idx_write_outcomes_owner (user_id, agent_id, session_id)
+    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
+    """,
+    """
     CREATE TABLE IF NOT EXISTS system_metrics (
         minute_ts   VARCHAR(32) NOT NULL,
         data        LONGTEXT NOT NULL,
@@ -459,6 +474,62 @@ class MysqlCache(CacheBase):
             logger.warning(f"get_pipeline_logs failed: {e}")
             return []
 
+    async def claim_write_outcome(
+        self,
+        request_id: str,
+        user_id: str,
+        agent_id: str,
+        session_id: str,
+        operation: str,
+    ) -> Optional[Dict[str, Any]]:
+        now = datetime.now().isoformat()
+        inserted = await self._execute(
+            """
+            INSERT IGNORE INTO write_outcomes (
+                request_id, user_id, agent_id, session_id, operation,
+                status, result_json, created_at, updated_at
+            ) VALUES (%s, %s, %s, %s, %s, 'running', '{}', %s, %s)
+            """,
+            (request_id, user_id, agent_id, session_id, operation, now, now),
+        )
+        if inserted == 1:
+            return None
+        return await self.get_write_outcome(request_id)
+
+    async def finish_write_outcome(
+        self,
+        request_id: str,
+        status: str,
+        result: Dict[str, Any],
+    ) -> bool:
+        updated = await self._execute(
+            """
+            UPDATE write_outcomes
+            SET status = %s, result_json = %s, updated_at = %s
+            WHERE request_id = %s
+            """,
+            (
+                status,
+                json.dumps(result, ensure_ascii=False, default=str),
+                datetime.now().isoformat(),
+                request_id,
+            ),
+        )
+        return updated == 1
+
+    async def get_write_outcome(self, request_id: str) -> Optional[Dict[str, Any]]:
+        row = await self._fetchone(
+            "SELECT * FROM write_outcomes WHERE request_id = %s",
+            (request_id,),
+        )
+        if row is None:
+            return None
+        try:
+            row["result"] = json.loads(row.pop("result_json") or "{}")
+        except (TypeError, json.JSONDecodeError):
+            row["result"] = {}
+        return row
+
     # ================================================================
     # System Metrics（分钟粒度落盘）
     # ================================================================

hy_memory/data/cache_sqlite.py

@@ -87,6 +87,20 @@ CREATE TABLE IF NOT EXISTS pipeline_logs_v2 (
 CREATE INDEX IF NOT EXISTS idx_plog_request ON pipeline_logs_v2(request_id);
 CREATE INDEX IF NOT EXISTS idx_plog_user    ON pipeline_logs_v2(user_id);
 
+CREATE TABLE IF NOT EXISTS write_outcomes (
+    request_id   TEXT PRIMARY KEY,
+    user_id      TEXT NOT NULL,
+    agent_id     TEXT NOT NULL DEFAULT '',
+    session_id   TEXT NOT NULL DEFAULT '',
+    operation    TEXT NOT NULL,
+    status       TEXT NOT NULL,
+    result_json  TEXT NOT NULL DEFAULT '{}',
+    created_at   TEXT NOT NULL,
+    updated_at   TEXT NOT NULL
+);
+CREATE INDEX IF NOT EXISTS idx_write_outcomes_owner
+ON write_outcomes(user_id, agent_id, session_id);
+
 CREATE TABLE IF NOT EXISTS system_metrics (
     minute_ts  TEXT PRIMARY KEY,
     data       TEXT NOT NULL,
@@ -414,6 +428,88 @@ class SqliteCache(CacheBase):
             logger.warning(f"get_pipeline_logs failed: {e}")
             return []
 
+    # ================================================================
+    # Durable write outcomes
+    # ================================================================
+
+    async def claim_write_outcome(
+        self,
+        request_id: str,
+        user_id: str,
+        agent_id: str,
+        session_id: str,
+        operation: str,
+    ) -> Optional[Dict[str, Any]]:
+        def _claim():
+            with self._lock:
+                now = datetime.now().isoformat()
+                cursor = self._conn.execute(
+                    """
+                    INSERT OR IGNORE INTO write_outcomes (
+                        request_id, user_id, agent_id, session_id, operation,
+                        status, result_json, created_at, updated_at
+                    ) VALUES (?, ?, ?, ?, ?, 'running', '{}', ?, ?)
+                    """,
+                    (request_id, user_id, agent_id, session_id, operation, now, now),
+                )
+                self._conn.commit()
+                if cursor.rowcount == 1:
+                    return None
+                row = self._conn.execute(
+                    "SELECT * FROM write_outcomes WHERE request_id = ?",
+                    (request_id,),
+                ).fetchone()
+                return dict(row) if row else None
+
+        existing = await _run_in_sqlite_pool(_claim)
+        if existing is None:
+            return None
+        return await self.get_write_outcome(request_id)
+
+    async def finish_write_outcome(
+        self,
+        request_id: str,
+        status: str,
+        result: Dict[str, Any],
+    ) -> bool:
+        def _finish():
+            with self._lock:
+                cursor = self._conn.execute(
+                    """
+                    UPDATE write_outcomes
+                    SET status = ?, result_json = ?, updated_at = ?
+                    WHERE request_id = ?
+                    """,
+                    (
+                        status,
+                        json.dumps(result, ensure_ascii=False, default=str),
+                        datetime.now().isoformat(),
+                        request_id,
+                    ),
+                )
+                self._conn.commit()
+                return cursor.rowcount == 1
+
+        return await _run_in_sqlite_pool(_finish)
+
+    async def get_write_outcome(self, request_id: str) -> Optional[Dict[str, Any]]:
+        def _get():
+            with self._lock:
+                row = self._conn.execute(
+                    "SELECT * FROM write_outcomes WHERE request_id = ?",
+                    (request_id,),
+                ).fetchone()
+                return dict(row) if row else None
+
+        row = await _run_in_sqlite_pool(_get)
+        if row is None:
+            return None
+        try:
+            row["result"] = json.loads(row.pop("result_json") or "{}")
+        except (TypeError, json.JSONDecodeError):
+            row["result"] = {}
+        return row
+
     # ================================================================
     # System Metrics（分钟粒度落盘）
     # ================================================================

hy_memory/pipelines/base.py

@@ -231,6 +231,10 @@ class WriteRequest:
     # 已有记忆 (V1 合并检测用)
     existing_memories: Optional[List[Dict[str, Any]]] = None
 
+    # Public client process_raw() only: reuse these exact L1 nodes as input.
+    # Ordinary add() never populates this field.
+    existing_raw_ids: List[str] = field(default_factory=list)
+
     # 记忆时间戳（不传则用当前时间 datetime.now()）
     memory_at: Optional[datetime] = None
 

hy_memory/pipelines/writer.py

@@ -280,6 +280,27 @@ class MemoryWriter(WritePipeline):
             custom["source_raw_memory_ids"] = source_ids
         return custom
 
+    async def _attach_existing_raw_ids(
+        self,
+        vector_store,
+        memory_ids: List[str],
+        raw_ids: List[str],
+    ) -> None:
+        """Attach every reused raw record to canonical outputs."""
+        for memory_id in memory_ids:
+            node = await vector_store.get_by_id(memory_id)
+            if node is None:
+                continue
+            custom = dict(node.custom or {})
+            source_ids = list(custom.get("source_raw_memory_ids") or [])
+            legacy_source = str(getattr(node, "source_raw_memory_id", "") or "")
+            for raw_id in (legacy_source, *raw_ids):
+                if raw_id and raw_id not in source_ids:
+                    source_ids.append(raw_id)
+            if source_ids:
+                custom["source_raw_memory_ids"] = source_ids
+                await vector_store.update_payload(memory_id, {"custom": custom})
+
     @staticmethod
     def _attach_client_meta(custom: Dict[str, Any], request: Optional[WriteRequest]) -> None:
         """给新建节点挂上调用方 meta（单份 list）。
@@ -2596,13 +2617,23 @@ class MemoryWriter(WritePipeline):
             from ..metrics import MetricsCollector
             MetricsCollector.get().sys1_start()
 
-            # 2. 向量化 + 持久化（L1_RAW）
+            existing_raw_ids = request.existing_raw_ids
+            if (
+                not isinstance(existing_raw_ids, list)
+                or any(not isinstance(raw_id, str) or not raw_id for raw_id in existing_raw_ids)
+            ):
+                raise ValueError("_existing_raw_ids must be a list of non-empty ids")
+
+            # 2. 向量化（处理已有 raw 时复用已存节点，不再次 embed）
             _t_embed = datetime.now()
-            with tracer.span("embed") as s:
+            if existing_raw_ids:
-                embedding, embed_meta = await self.embed_service.embed_queued_with_meta(
+                embedding, embed_meta = [], {"reused_raw": True}
-                    request.content
+            else:
-                )
+                with tracer.span("embed") as s:
-                s.set_output({"dims": len(embedding), "content_len": len(request.content)})
+                    embedding, embed_meta = await self.embed_service.embed_queued_with_meta(
+                        request.content
+                    )
+                    s.set_output({"dims": len(embedding), "content_len": len(request.content)})
             _embed_ms = (datetime.now() - _t_embed).total_seconds() * 1000
             response.extra["embedding"] = embedding
             await self._emit_pipeline_step(
@@ -2625,27 +2656,43 @@ class MemoryWriter(WritePipeline):
             with tracer.span("qdrant_upsert") as s:
                 try:
                     vector_store = await self._get_vector_store()
-                    layer_enum = MemoryLayer.from_string(suggested_layer)
+                    if existing_raw_ids:
-                    _raw_custom: Dict[str, Any] = {}
+                        memory_id = existing_raw_ids[-1]
-                    self._attach_client_meta(_raw_custom, request)
+                        mem_node = await vector_store.get_by_id(memory_id)
-                    mem_node = MemoryNode(
+                        if (
-                        user_id=request.user_id,
+                            mem_node is None
-                        agent_id=request.agent_id or "default_agent",
+                            or mem_node.layer != MemoryLayer.L1_RAW
-                        session_id=request.session_id or "default_session",
+                            or mem_node.user_id != request.user_id
-                        source=request.source,
+                            or mem_node.agent_id != (request.agent_id or "default_agent")
-                        layer=layer_enum,
+                            or mem_node.session_id != (request.session_id or "default_session")
-                        content=request.content,
+                            or mem_node.source != request.source
-                        source_type=SourceType.EXPLICIT,
+                        ):
-                        status=MemoryStatus.ACTIVE,
+                            raise ValueError("existing raw record is outside the write scope")
-                        is_latest=True,
+                    else:
-                        embedding=embedding,
+                        layer_enum = MemoryLayer.from_string(suggested_layer)
-                        memory_at=request.memory_at,
+                        _raw_custom: Dict[str, Any] = {}
-                        custom=_raw_custom,
+                        self._attach_client_meta(_raw_custom, request)
-                    )
+                        mem_node = MemoryNode(
-                    memory_id = await vector_store.upsert(mem_node)
+                            user_id=request.user_id,
+                            agent_id=request.agent_id or "default_agent",
+                            session_id=request.session_id or "default_session",
+                            source=request.source,
+                            layer=layer_enum,
+                            content=request.content,
+                            source_type=SourceType.EXPLICIT,
+                            status=MemoryStatus.ACTIVE,
+                            is_latest=True,
+                            embedding=embedding,
+                            memory_at=request.memory_at,
+                            custom=_raw_custom,
+                        )
+                        memory_id = await vector_store.upsert(mem_node)
                     response.memory_id = memory_id
                     s.set_output({"memory_id": memory_id, "layer": suggested_layer})
-                    logger.debug(f"[write] persisted: id={memory_id} layer={suggested_layer}")
+                    logger.debug(
+                        f"[write] {'reused' if existing_raw_ids else 'persisted'}: "
+                        f"id={memory_id} layer={suggested_layer}"
+                    )
                 except Exception as persist_err:
                     _l1_error = str(persist_err)
                     s.set_error(_l1_error)
@@ -2709,6 +2756,12 @@ class MemoryWriter(WritePipeline):
                         recent_history_count=_recent_history_count,
                     )
                     total_tokens += response.extra.get("_agent_tokens", 0)
+                if existing_raw_ids and response.extra.get("agent_stored_ids"):
+                    await self._attach_existing_raw_ids(
+                        vector_store,
+                        list(response.extra["agent_stored_ids"]),
+                        existing_raw_ids,
+                    )
                 if response.extra.get("agent_status") == "failed":
                     await self._emit_pipeline_step(
                         request,
@@ -2718,12 +2771,13 @@ class MemoryWriter(WritePipeline):
                             "error": response.extra.get("agent_error", ""),
                         },
                     )
-                    await self._rollback_failed_l1(
+                    if not existing_raw_ids:
-                        request=request,
+                        await self._rollback_failed_l1(
-                        response=response,
+                            request=request,
-                        vector_store=vector_store,
+                            response=response,
-                        memory_id=memory_id,
+                            vector_store=vector_store,
-                    )
+                            memory_id=memory_id,
+                        )
                     response.tokens_used = total_tokens
                     _failed_workflow_ms = (
                         datetime.now() - _t_workflow

hy_memory/snapshot.py

@@ -0,0 +1,150 @@
+"""Public, offline snapshot support for the SDK-owned local data plane."""
+
+from __future__ import annotations
+
+import hashlib
+import os
+import shutil
+from dataclasses import dataclass
+from pathlib import Path
+
+
+class DataPlaneSnapshotRejected(RuntimeError):
+    def __init__(self, code: str) -> None:
+        super().__init__(code)
+        self.code = code
+
+
+@dataclass(frozen=True)
+class DataPlaneSnapshot:
+    source_root: Path
+    snapshot_root: Path
+    files: tuple[dict[str, object], ...]
+    total_bytes: int
+
+
+@dataclass(frozen=True)
+class DataPlaneRestore:
+    restored_root: Path
+    previous_root: Path | None
+    files_restored: int
+    total_bytes: int
+
+
+def _sha256(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as handle:
+        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def snapshot_local_data_plane(
+    *,
+    source_root: Path,
+    target_root: Path,
+    writers_stopped: bool,
+) -> DataPlaneSnapshot:
+    """Copy one quiesced local SDK root and return a verified inventory.
+
+    Remote providers require their own export implementation and must not call
+    this local-path contract.
+    """
+    if not writers_stopped:
+        raise DataPlaneSnapshotRejected("data_plane_snapshot_writers_running")
+    source = source_root.resolve()
+    target = target_root.resolve()
+    if source.is_symlink() or not source.is_dir():
+        raise DataPlaneSnapshotRejected("data_plane_snapshot_source_invalid")
+    if target.exists():
+        raise DataPlaneSnapshotRejected("data_plane_snapshot_target_exists")
+
+    try:
+        target.mkdir(parents=True)
+        for root, directories, files in os.walk(source, followlinks=False):
+            root_path = Path(root)
+            relative = root_path.relative_to(source)
+            target_directory = target / relative
+            target_directory.mkdir(parents=True, exist_ok=True)
+            for name in directories:
+                if (root_path / name).is_symlink():
+                    raise DataPlaneSnapshotRejected("data_plane_snapshot_symlink")
+            for name in files:
+                item = root_path / name
+                if item.is_symlink():
+                    raise DataPlaneSnapshotRejected("data_plane_snapshot_symlink")
+                shutil.copy2(item, target_directory / name)
+
+        inventory: list[dict[str, object]] = []
+        for path in sorted(item for item in target.rglob("*") if item.is_file()):
+            inventory.append(
+                {
+                    "path": path.relative_to(target).as_posix(),
+                    "bytes": path.stat().st_size,
+                    "sha256": _sha256(path),
+                }
+            )
+        for item in inventory:
+            copied = target / str(item["path"])
+            if copied.stat().st_size != item["bytes"] or _sha256(copied) != item["sha256"]:
+                raise DataPlaneSnapshotRejected("data_plane_snapshot_verify_failed")
+    except DataPlaneSnapshotRejected:
+        shutil.rmtree(target, ignore_errors=True)
+        raise
+    except OSError as error:
+        shutil.rmtree(target, ignore_errors=True)
+        raise DataPlaneSnapshotRejected("data_plane_snapshot_failed") from error
+
+    return DataPlaneSnapshot(
+        source_root=source,
+        snapshot_root=target,
+        files=tuple(inventory),
+        total_bytes=sum(int(item["bytes"]) for item in inventory),
+    )
+
+
+def restore_local_data_plane(
+    *,
+    snapshot_root: Path,
+    target_root: Path,
+    writers_stopped: bool,
+    protective_root: Path | None = None,
+) -> DataPlaneRestore:
+    """Replace one local SDK root from a verified snapshot with rollback."""
+    if not writers_stopped:
+        raise DataPlaneSnapshotRejected("data_plane_restore_writers_running")
+    source = snapshot_root.resolve()
+    target = target_root.resolve()
+    protective = (
+        protective_root.resolve()
+        if protective_root is not None
+        else target.parent / f".{target.name}.pre-restore"
+    )
+    staging = target.parent / f".{target.name}.restore.tmp"
+    if target.is_symlink() or protective.exists() or staging.exists():
+        raise DataPlaneSnapshotRejected("data_plane_restore_target_unsafe")
+
+    copied = snapshot_local_data_plane(
+        source_root=source,
+        target_root=staging,
+        writers_stopped=True,
+    )
+    previous: Path | None = None
+    try:
+        if target.exists():
+            target.replace(protective)
+            previous = protective
+        staging.replace(target)
+    except OSError as error:
+        if previous is not None and previous.exists() and not target.exists():
+            previous.replace(target)
+            previous = None
+        shutil.rmtree(staging, ignore_errors=True)
+        raise DataPlaneSnapshotRejected("data_plane_restore_replace_failed") from error
+
+    return DataPlaneRestore(
+        restored_root=target,
+        previous_root=previous,
+        files_restored=len(copied.files),
+        total_bytes=copied.total_bytes,
+    )

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "hy-memory"
-version = "2.0.4"
+version = "2.0.5.dev4"
 description = "HY Memory - Industrial-grade dual-system cognitive memory framework"
 readme = "README.md"
 requires-python = ">=3.9"

tests/coding/test_chain_block_config.py

@@ -0,0 +1,19 @@
+from hy_memory.config import MemoryConfig
+
+
+def test_chain_block_defaults_disabled(monkeypatch):
+    monkeypatch.delenv("MEMORY_CODING_CHAIN_BLOCK_ENABLED", raising=False)
+    assert MemoryConfig().coding.chain_block_enabled is False
+
+
+def test_chain_block_env_can_enable(monkeypatch):
+    monkeypatch.setenv("MEMORY_CODING_CHAIN_BLOCK_ENABLED", "true")
+    assert MemoryConfig().coding.chain_block_enabled is True
+
+
+def test_chain_block_from_dict_can_enable(monkeypatch):
+    monkeypatch.delenv("MEMORY_CODING_CHAIN_BLOCK_ENABLED", raising=False)
+    config = MemoryConfig.from_dict({
+        "coding": {"chain_block_enabled": True},
+    })
+    assert config.coding.chain_block_enabled is True

tests/coding/test_scene_coding_search.py

@@ -62,7 +62,21 @@ def _seed_chain(tmpdir, branch="feat/q4", path=_PATH):
     return H
 
 
-def _bind_chain(ctx, store):
+def _enable_chain_block(ctx):
+    """search 侧 chain_block 默认关；测注入路径时显式 opt-in。"""
+    from hy_memory.config import MemoryConfig
+
+    cfg = getattr(ctx, "_config", None)
+    if cfg is None:
+        cfg = MemoryConfig()
+        ctx._config = cfg
+    cfg.coding.chain_block_enabled = True
+    ctx._chain_block_enabled = HyMemoryClient._chain_block_enabled.__get__(
+        ctx, type(ctx))
+    return ctx
+
+
+def _bind_chain(ctx, store, *, enable_block=True):
     """把链层挂到 search_ctx 这个 SimpleNamespace 上。
 
     search_ctx 是裸 namespace，不是真 client，所以 _decorate_with_chain 这个
@@ -73,6 +87,8 @@ def _bind_chain(ctx, store):
     ctx._get_chain_store = lambda owner="": store
     ctx._decorate_with_chain = HyMemoryClient._decorate_with_chain.__get__(
         ctx, type(ctx))
+    if enable_block:
+        _enable_chain_block(ctx)
     return ctx
 
 
@@ -129,6 +145,17 @@ class TestSearchCodingVariantEndToEnd:
         assert "第1版" in blk and "第2版" in blk
         assert "255.0" in blk and "360.0" in blk
 
+    async def test_coding_variant_chain_block_default_off(self, search_ctx, chain_dir):
+        """默认不注入 chain_block（省 token）；库里有链也返回空块。"""
+        await self._seed_fact(search_ctx)
+        # 挂上真链层，但不开启 chain_block_enabled
+        _bind_chain(search_ctx, _seed_chain(chain_dir), enable_block=False)
+
+        resp = await self._search(search_ctx, _STATEMENT, variant="coding")
+        assert len(resp["facts"]) == 1
+        assert resp["chain_block"] == ""
+        assert resp["chain_used"] == []
+
     async def test_facts_not_deduped(self, search_ctx, chain_dir):
         """★ 挂上链的 fact 仍留在 facts 里 —— 刻意不去重。
 

tests/coding/test_scene_coding_variant.py

@@ -326,12 +326,34 @@ class TestChainDecorate:
             os.environ.pop("HCM_BLOCK_OFF", None)
 
 
+def _ns_with_chain_block(**kwargs):
+    """胶水测需要显式 opt-in：默认 chain_block_enabled=False 会在进 store 前短路。"""
+    import types
+    from hy_memory.config import MemoryConfig
+
+    cfg = MemoryConfig()
+    cfg.coding.chain_block_enabled = True
+    ns = types.SimpleNamespace(_config=cfg, **kwargs)
+    ns._chain_block_enabled = HyMemoryClient._chain_block_enabled.__get__(
+        ns, type(ns))
+    return ns
+
+
 class TestDecorateGlue:
     """client._decorate_with_chain —— 胶水层的 fail-safe。"""
 
-    def test_returns_empty_when_chain_unavailable(self):
+    def test_returns_empty_when_disabled_by_default(self):
         import types
         ns = types.SimpleNamespace(
+            _chain_store=None, _chain_store_failed=False,
+            _get_chain_store=lambda owner="": object(),
+        )
+        ns._chain_block_enabled = HyMemoryClient._chain_block_enabled.__get__(
+            ns, type(ns))
+        assert HyMemoryClient._decorate_with_chain(ns, [dict(_FACT)]) == ("", [])
+
+    def test_returns_empty_when_chain_unavailable(self):
+        ns = _ns_with_chain_block(
             _chain_store=None, _chain_store_failed=True,
             _get_chain_store=lambda owner="": None,
         )
@@ -339,13 +361,12 @@ class TestDecorateGlue:
 
     def test_store_exception_degrades_not_raises(self):
         # 这层是增量, 挂了要退回纯向量双层而不是把整次 search 拖失败。
-        import types
 
         class _Boom:
             def decorate(self, *a, **k):
                 raise RuntimeError("boom")
 
-        ns = types.SimpleNamespace(_get_chain_store=lambda owner="": _Boom())
+        ns = _ns_with_chain_block(_get_chain_store=lambda owner="": _Boom())
         assert HyMemoryClient._decorate_with_chain(ns, [dict(_FACT)]) == ("", [])
 
 

tests/test_cache_backend_parity.py

@@ -24,6 +24,9 @@ from hy_memory.data.cache_sqlite import SqliteCache
 STATEFUL_METHODS = [
     "get_episode_state",
     "upsert_episode_state",
+    "claim_write_outcome",
+    "finish_write_outcome",
+    "get_write_outcome",
 ]
 
 BACKENDS = [
@@ -70,6 +73,16 @@ def test_episode_state_table_declared(backend):
     )
 
 
+@pytest.mark.parametrize("backend", BACKENDS)
+def test_write_outcomes_table_declared(backend):
+    import hy_memory.data.cache_mysql as m
+    import hy_memory.data.cache_sqlite as s
+
+    mod = s if backend is SqliteCache else m
+    src = inspect.getsource(mod)
+    assert "CREATE TABLE IF NOT EXISTS write_outcomes" in src
+
+
 class TestRollingSummaryFullyRemoved:
     """rolling summary 已整体移除，残留引用会让应用在启动路径报错。"""
 

tests/test_data_plane_snapshot.py

@@ -0,0 +1,125 @@
+from __future__ import annotations
+
+import hashlib
+
+import pytest
+
+from hy_memory.snapshot import (
+    DataPlaneSnapshotRejected,
+    restore_local_data_plane,
+    snapshot_local_data_plane,
+)
+
+
+def _sha256(path):
+    return hashlib.sha256(path.read_bytes()).hexdigest()
+
+
+def test_snapshot_copies_and_verifies_local_sdk_tree(tmp_path):
+    source = tmp_path / "source"
+    (source / "vector").mkdir(parents=True)
+    (source / "vector" / "nodes.bin").write_bytes(b"vector")
+    (source / "cache.sqlite3").write_bytes(b"cache")
+    target = tmp_path / "snapshot"
+
+    result = snapshot_local_data_plane(
+        source_root=source,
+        target_root=target,
+        writers_stopped=True,
+    )
+
+    assert {item["path"] for item in result.files} == {
+        "cache.sqlite3",
+        "vector/nodes.bin",
+    }
+    for item in result.files:
+        copied = target / item["path"]
+        assert copied.stat().st_size == item["bytes"]
+        assert _sha256(copied) == item["sha256"]
+    assert result.total_bytes == len(b"vector") + len(b"cache")
+
+
+def test_snapshot_refuses_active_writers(tmp_path):
+    source = tmp_path / "source"
+    source.mkdir()
+    with pytest.raises(DataPlaneSnapshotRejected) as caught:
+        snapshot_local_data_plane(
+            source_root=source,
+            target_root=tmp_path / "snapshot",
+            writers_stopped=False,
+        )
+    assert caught.value.code == "data_plane_snapshot_writers_running"
+
+
+def test_snapshot_rejects_symlink_and_removes_partial_target(tmp_path):
+    source = tmp_path / "source"
+    source.mkdir()
+    outside = tmp_path / "outside"
+    outside.write_text("outside", encoding="utf-8")
+    try:
+        (source / "linked").symlink_to(outside)
+    except OSError:
+        pytest.skip("symlinks unavailable")
+    target = tmp_path / "snapshot"
+
+    with pytest.raises(DataPlaneSnapshotRejected) as caught:
+        snapshot_local_data_plane(
+            source_root=source,
+            target_root=target,
+            writers_stopped=True,
+        )
+    assert caught.value.code == "data_plane_snapshot_symlink"
+    assert not target.exists()
+
+
+def test_snapshot_target_must_not_exist(tmp_path):
+    source = tmp_path / "source"
+    target = tmp_path / "snapshot"
+    source.mkdir()
+    target.mkdir()
+    with pytest.raises(DataPlaneSnapshotRejected) as caught:
+        snapshot_local_data_plane(
+            source_root=source,
+            target_root=target,
+            writers_stopped=True,
+        )
+    assert caught.value.code == "data_plane_snapshot_target_exists"
+
+
+def test_restore_replaces_data_plane_and_keeps_protective_copy(tmp_path):
+    snapshot = tmp_path / "snapshot"
+    target = tmp_path / "live"
+    (snapshot / "vector").mkdir(parents=True)
+    (snapshot / "vector" / "nodes.bin").write_bytes(b"new")
+    (target / "vector").mkdir(parents=True)
+    (target / "vector" / "nodes.bin").write_bytes(b"old")
+    protective = tmp_path / "previous"
+
+    result = restore_local_data_plane(
+        snapshot_root=snapshot,
+        target_root=target,
+        protective_root=protective,
+        writers_stopped=True,
+    )
+
+    assert (target / "vector" / "nodes.bin").read_bytes() == b"new"
+    assert (protective / "vector" / "nodes.bin").read_bytes() == b"old"
+    assert result.previous_root == protective
+    assert result.files_restored == 1
+
+
+def test_restore_refuses_active_writers_without_touching_live_data(tmp_path):
+    snapshot = tmp_path / "snapshot"
+    target = tmp_path / "live"
+    snapshot.mkdir()
+    target.mkdir()
+    (target / "value").write_text("old", encoding="utf-8")
+
+    with pytest.raises(DataPlaneSnapshotRejected) as caught:
+        restore_local_data_plane(
+            snapshot_root=snapshot,
+            target_root=target,
+            writers_stopped=False,
+        )
+    assert caught.value.code == "data_plane_restore_writers_running"
+    assert (target / "value").read_text(encoding="utf-8") == "old"

tests/test_delete_complete.py

@@ -0,0 +1,120 @@
+from __future__ import annotations
+
+from unittest.mock import AsyncMock
+
+import pytest
+
+from hy_memory.client import HyMemoryClient
+from hy_memory.config import MemoryConfig
+from hy_memory.models.memory import MemoryLayer, MemoryNode
+
+
+class VectorStore:
+    _collection_name = "test"
+
+    def __init__(self, node=None, *, delete_result=True):
+        self.node = node
+        self.delete_result = delete_result
+
+    async def get_by_id(self, _memory_id):
+        return self.node
+
+    async def delete(self, _memory_id):
+        result = self.node is not None and self.delete_result
+        if result:
+            self.node = None
+        return result
+
+    async def delete_retrieval_clauses(self, _memory_id):
+        return None
+
+
+class GraphStore:
+    def __init__(self, node=None, *, delete_result=True, lookup_error=False):
+        self.node = node
+        self.delete_result = delete_result
+        self.lookup_error = lookup_error
+
+    async def get_node(self, _memory_id):
+        if self.lookup_error:
+            raise RuntimeError("graph unavailable")
+        return self.node
+
+    async def delete_node(self, _memory_id):
+        if self.delete_result:
+            self.node = None
+        return self.delete_result
+
+
+def _node():
+    return MemoryNode(
+        node_id="m1",
+        user_id="u1",
+        agent_id="a1",
+        session_id="s1",
+        content="body",
+        layer=MemoryLayer.L2_FACT,
+    )
+
+
+def _client(vector, graph):
+    client = HyMemoryClient.__new__(HyMemoryClient)
+    client._vector_store = vector
+    client._graph_store = graph
+    client._history_store = None
+    client._config = MemoryConfig.from_dict(
+        {"vector_store": {"provider": "faiss", "collection_name": "test"}}
+    )
+    client._maybe_store_op_log = AsyncMock()
+    client._cleanup_graph_evidence = AsyncMock(return_value=True)
+    return client
+
+
+@pytest.mark.asyncio
+async def test_delete_succeeds_only_when_every_existing_store_is_deleted():
+    client = _client(VectorStore(_node()), GraphStore(_node()))
+    result = await HyMemoryClient.async_delete.__wrapped__(client, "m1")
+
+    assert result["success"] is True
+    assert result["deleted_count"] == 1
+    assert result["vdb_deleted"] is True
+    assert result["graph_deleted"] is True
+    assert result["partial"] is False
+    assert result["error_code"] is None
+
+
+@pytest.mark.asyncio
+async def test_graph_failure_is_reported_as_partial_not_success():
+    client = _client(
+        VectorStore(_node()),
+        GraphStore(_node(), delete_result=False),
+    )
+    result = await HyMemoryClient.async_delete.__wrapped__(client, "m1")
+
+    assert result["success"] is False
+    assert result["deleted_count"] == 0
+    assert result["vdb_deleted"] is True
+    assert result["graph_deleted"] is False
+    assert result["partial"] is True
+    assert result["error_code"] == "delete_incomplete"
+
+
+@pytest.mark.asyncio
+async def test_graph_only_memory_can_be_physically_deleted():
+    client = _client(VectorStore(None), GraphStore(_node()))
+    result = await HyMemoryClient.async_delete.__wrapped__(client, "m1")
+
+    assert result["success"] is True
+    assert result["vdb_deleted"] is False
+    assert result["graph_deleted"] is True
+    assert result["deleted_count"] == 1
+
+
+@pytest.mark.asyncio
+async def test_graph_lookup_failure_never_collapses_to_not_found_success():
+    client = _client(VectorStore(None), GraphStore(lookup_error=True))
+    result = await HyMemoryClient.async_delete.__wrapped__(client, "m1")
+
+    assert result["success"] is False
+    assert result["partial"] is False
+    assert result["error_code"] == "delete_incomplete"

tests/test_list_layer_filter_client.py

@@ -31,7 +31,12 @@ def run(coro):
 DIMS = 32
 
 
-def _node(user_id="u1", content="x", layer=MemoryLayer.L2_FACT):
+def _node(
+    user_id="u1",
+    content="x",
+    layer=MemoryLayer.L2_FACT,
+    status=MemoryStatus.ACTIVE,
+):
     import random
     n = MemoryNode(
         node_id=str(uuid.uuid4()),
@@ -40,8 +45,8 @@ def _node(user_id="u1", content="x", layer=MemoryLayer.L2_FACT):
         session_id="s1",
         content=content,
         layer=layer,
-        status=MemoryStatus.ACTIVE,
+        status=status,
-        is_latest=True,
+        is_latest=status == MemoryStatus.ACTIVE,
     )
     vec = [random.gauss(0, 1) for _ in range(DIMS)]
     norm = sum(v * v for v in vec) ** 0.5
@@ -104,6 +109,22 @@ class TestNormalizeLayers:
             HyMemoryClient._normalize_layers(["l3_identity"])
 
 
+class TestNormalizeStatuses:
+    def test_none_and_empty_preserve_active_default(self):
+        assert HyMemoryClient._normalize_statuses(None) is None
+        assert HyMemoryClient._normalize_statuses([]) is None
+
+    def test_string_and_enum_values_are_deduplicated(self):
+        got = HyMemoryClient._normalize_statuses(
+            ["superseded", MemoryStatus.SHADOW, "superseded"]
+        )
+        assert got == [MemoryStatus.SUPERSEDED, MemoryStatus.SHADOW]
+
+    def test_invalid_status_raises(self):
+        with pytest.raises(ValueError):
+            HyMemoryClient._normalize_statuses(["deleted"])
+
+
 class TestVdbBucketLayerFilter:
     def _seed(self, store):
         run(store.upsert_batch([
@@ -163,6 +184,24 @@ class TestVdbBucketLayerFilter:
         ))
         assert [m["content"] for m in got["memories"]] == ["fact"]
 
+    def test_status_filter_defaults_to_active_and_can_read_superseded(self, faiss_store):
+        run(faiss_store.upsert_batch([
+            _node(content="active", status=MemoryStatus.ACTIVE),
+            _node(content="old", status=MemoryStatus.SUPERSEDED),
+            _node(content="hidden", status=MemoryStatus.SHADOW),
+        ]))
+
+        default_page = run(_client_with(faiss_store)._list_vdb_bucket(
+            user_id="u1", agent_id=None, limit=100, offset=0, order="desc",
+        ))
+        assert [m["content"] for m in default_page["memories"]] == ["active"]
+
+        lifecycle_page = run(_client_with(faiss_store)._list_vdb_bucket(
+            user_id="u1", agent_id=None, limit=100, offset=0, order="desc",
+            statuses=[MemoryStatus.SUPERSEDED, MemoryStatus.SHADOW],
+        ))
+        assert {m["content"] for m in lifecycle_page["memories"]} == {"old", "hidden"}
+
     def test_layer_filter_composes_with_multi_user(self, faiss_store):
         """R3 与 R1 叠加：多用户 + 层级筛选同时生效。"""
         run(faiss_store.upsert_batch([
@@ -215,3 +254,37 @@ class TestGraphBucketLayerSkip:
             user_id="u1", agent_id=None, limit=10, offset=0, order="desc",
         ))
         assert calls == [1], "不传 layers 时行为与改动前一致"
+
+    def test_graph_status_filter_matches_vdb_contract(self):
+        active = _node(content="active graph", layer=MemoryLayer.L6_SCHEMA)
+        old = _node(
+            content="old graph",
+            layer=MemoryLayer.L6_SCHEMA,
+            status=MemoryStatus.SUPERSEDED,
+        )
+
+        class GraphStore:
+            async def get_all_nodes(self, *_args, **_kwargs):
+                return [active, old]
+
+            async def close(self):
+                return None
+
+        c = HyMemoryClient.__new__(HyMemoryClient)
+
+        async def _acquire():
+            return GraphStore(), True
+
+        c._acquire_graph_store_for_purge = _acquire
+        default_page = run(c._list_graph_bucket(
+            user_id="u1", agent_id=None, limit=10, offset=0, order="desc",
+            layers=[MemoryLayer.L6_SCHEMA],
+        ))
+        assert [item["content"] for item in default_page["nodes"]] == ["active graph"]
+
+        old_page = run(c._list_graph_bucket(
+            user_id="u1", agent_id=None, limit=10, offset=0, order="desc",
+            layers=[MemoryLayer.L6_SCHEMA],
+            statuses=[MemoryStatus.SUPERSEDED],
+        ))
+        assert [item["content"] for item in old_page["nodes"]] == ["old graph"]

tests/test_raw_public_api.py

@@ -0,0 +1,405 @@
+from __future__ import annotations
+
+from unittest.mock import AsyncMock
+
+import pytest
+
+from hy_memory.client import HyMemoryClient
+from hy_memory.config import MemoryConfig
+from hy_memory.models.memory import (
+    MemoryLayer,
+    MemoryNode,
+    MemoryStatus,
+    SourceType,
+)
+from hy_memory.pipelines.base import WriteRequest
+from hy_memory.pipelines.writer import MemoryWriter
+
+
+class OutcomeCache:
+    def __init__(self):
+        self.rows = {}
+
+    async def claim_write_outcome(
+        self, request_id, user_id, agent_id, session_id, operation
+    ):
+        if request_id in self.rows:
+            return self.rows[request_id]
+        self.rows[request_id] = {
+            "request_id": request_id,
+            "user_id": user_id,
+            "agent_id": agent_id,
+            "session_id": session_id,
+            "operation": operation,
+            "status": "running",
+            "result": {},
+        }
+        return None
+
+    async def finish_write_outcome(self, request_id, status, result):
+        self.rows[request_id].update(status=status, result=dict(result))
+        return True
+
+    async def get_write_outcome(self, request_id):
+        return self.rows.get(request_id)
+
+
+class RawStore:
+    def __init__(self):
+        self.nodes = {}
+        self.upserts = 0
+        self.supports_fulltext = False
+
+    async def get_by_id(self, node_id):
+        return self.nodes.get(node_id)
+
+    async def initialize(self):
+        return None
+
+    async def upsert(self, node):
+        self.upserts += 1
+        self.nodes[node.node_id] = node
+        return node.node_id
+
+    async def update_payload(self, node_id, payload):
+        node = self.nodes[node_id]
+        for key, value in payload.items():
+            setattr(node, key, value)
+
+    async def list_by_user(
+        self,
+        *,
+        user_id,
+        agent_id=None,
+        include_sources=None,
+        **_kwargs,
+    ):
+        return [
+            node
+            for node in self.nodes.values()
+            if node.user_id == user_id
+            and (agent_id is None or node.agent_id == agent_id)
+            and (not include_sources or node.source in include_sources)
+        ]
+
+    async def search(
+        self,
+        *,
+        user_ids,
+        agent_ids,
+        include_sources,
+        exclude_sources,
+        layers,
+        limit,
+        **_kwargs,
+    ):
+        rows = []
+        for node in self.nodes.values():
+            if node.user_id not in user_ids:
+                continue
+            if agent_ids and node.agent_id not in agent_ids:
+                continue
+            if include_sources and node.source not in include_sources:
+                continue
+            if exclude_sources and node.source in exclude_sources:
+                continue
+            if node.layer not in layers:
+                continue
+            rows.append({"node": node, "score": 0.9})
+        return rows[:limit]
+
+
+def _client():
+    client = HyMemoryClient.__new__(HyMemoryClient)
+    client._cache = OutcomeCache()
+    client._vector_store = RawStore()
+    client._embed_service = AsyncMock()
+    client._embed_service.embed.return_value = [0.1, 0.2]
+    return client
+
+
+@pytest.mark.asyncio
+async def test_import_raw_preserves_identity_without_extraction():
+    client = _client()
+    result = await HyMemoryClient.async_import_raw.__wrapped__(
+        client,
+        "verbatim",
+        user_id="u1",
+        agent_id="a1",
+        session_id="s1",
+        source="cursor",
+        raw_id="raw-1",
+        request_id="request-1",
+        metadata={"role": "user", "source_order": 7},
+    )
+
+    assert result == {
+        "success": True,
+        "raw_id": "raw-1",
+        "request_id": "request-1",
+    }
+    node = client._vector_store.nodes["raw-1"]
+    assert node.layer == MemoryLayer.L1_RAW
+    assert node.content == "verbatim"
+    assert node.custom["role"] == "user"
+    assert node.custom["source_order"] == 7
+    assert node.custom["input_kind"] == "raw_exact_import"
+
+
+@pytest.mark.asyncio
+async def test_import_raw_request_replay_does_not_write_twice():
+    client = _client()
+    first = await HyMemoryClient.async_import_raw.__wrapped__(
+        client, "same", user_id="u1", raw_id="raw-1", request_id="request-1"
+    )
+    second = await HyMemoryClient.async_import_raw.__wrapped__(
+        client, "same", user_id="u1", raw_id="raw-1", request_id="request-1"
+    )
+
+    assert second == {**first, "replayed": True}
+    assert client._vector_store.upserts == 1
+
+
+@pytest.mark.asyncio
+async def test_process_raw_is_idempotent_and_does_not_insert_raw_again():
+    client = _client()
+    for index, role in enumerate(("user", "assistant"), start=1):
+        await HyMemoryClient.async_import_raw.__wrapped__(
+            client,
+            f"turn-{index}",
+            user_id="u1",
+            agent_id="a1",
+            session_id="s1",
+            source="cursor",
+            raw_id=f"raw-{index}",
+            request_id=f"import-{index}",
+            metadata={"role": role, "source_order": index},
+        )
+    client._async_add_impl = AsyncMock(
+        return_value={"success": True, "memory_ids": ["memory-1"]}
+    )
+
+    first = await HyMemoryClient.async_process_raw.__wrapped__(
+        client,
+        ["raw-1", "raw-2"],
+        user_id="u1",
+        agent_id="a1",
+        session_id="s1",
+        source="cursor",
+        request_id="process-1",
+    )
+    second = await HyMemoryClient.async_process_raw.__wrapped__(
+        client,
+        ["raw-1", "raw-2"],
+        user_id="u1",
+        agent_id="a1",
+        session_id="s1",
+        source="cursor",
+        request_id="process-1",
+    )
+
+    assert first["raw_ids"] == ["raw-1", "raw-2"]
+    assert first["memory_ids"] == ["memory-1"]
+    assert second == {**first, "replayed": True}
+    assert client._vector_store.upserts == 2
+    client._async_add_impl.assert_awaited_once()
+    assert client._async_add_impl.await_args.kwargs["_existing_raw_ids"] == [
+        "raw-1",
+        "raw-2",
+    ]
+
+
+@pytest.mark.asyncio
+async def test_writer_reuses_existing_raw_without_embedding_or_upsert():
+    store = RawStore()
+    raw = MemoryNode(
+        node_id="raw-1",
+        user_id="u1",
+        agent_id="a1",
+        session_id="s1",
+        source="cursor",
+        layer=MemoryLayer.L1_RAW,
+        content="verbatim",
+        source_type=SourceType.EXPLICIT,
+        status=MemoryStatus.ACTIVE,
+        is_latest=True,
+        embedding=[0.1, 0.2],
+    )
+    store.nodes[raw.node_id] = raw
+    embed = AsyncMock()
+    writer = MemoryWriter(
+        MemoryConfig(),
+        embed_service=embed,
+        vector_store=store,
+        cache=None,
+    )
+
+    response = await writer.write(
+        WriteRequest(
+            content="[user]: verbatim",
+            user_id="u1",
+            agent_id="a1",
+            session_id="s1",
+            source="cursor",
+            request_id="process-1",
+            existing_raw_ids=["raw-1"],
+            extra={
+                "agent_mode": "disabled",
+            },
+        )
+    )
+
+    assert response.success is True
+    assert response.memory_id == "raw-1"
+    assert store.upserts == 0
+    embed.embed_queued_with_meta.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_existing_raw_id_must_match_exact_content_and_owner():
+    client = _client()
+    await HyMemoryClient.async_import_raw.__wrapped__(
+        client, "first", user_id="u1", raw_id="raw-1", request_id="request-1"
+    )
+
+    with pytest.raises(ValueError, match="different source content"):
+        await HyMemoryClient.async_import_raw.__wrapped__(
+            client, "changed", user_id="u1", raw_id="raw-1", request_id="request-2"
+        )
+
+
+@pytest.mark.asyncio
+async def test_raw_get_and_list_use_explicit_api():
+    client = _client()
+    for raw_id, session_id, source in (
+        ("raw-1", "s1", "cursor"),
+        ("raw-2", "s2", "claude"),
+    ):
+        await HyMemoryClient.async_import_raw.__wrapped__(
+            client,
+            raw_id,
+            user_id="u1",
+            session_id=session_id,
+            source=source,
+            raw_id=raw_id,
+            request_id=f"request-{raw_id}",
+        )
+
+    item = await HyMemoryClient.async_get_raw.__wrapped__(client, "raw-1")
+    assert item["raw_id"] == "raw-1"
+    assert item["content"] == "raw-1"
+
+    page = await HyMemoryClient.async_list_raw.__wrapped__(
+        client,
+        user_id="u1",
+        source="claude",
+        limit=10,
+        offset=0,
+    )
+    assert page["total"] == 1
+    assert [item["raw_id"] for item in page["raw"]] == ["raw-2"]
+
+
+@pytest.mark.asyncio
+async def test_exact_memory_import_preserves_id_lifecycle_and_chain():
+    client = _client()
+    result = await HyMemoryClient.async_import_memory_exact.__wrapped__(
+        client,
+        "older preference",
+        memory_id="memory-1",
+        user_id="u1",
+        agent_id="a1",
+        session_id="s1",
+        source="cursor",
+        layer="l2_fact",
+        status="superseded",
+        owner="user",
+        tags=["preference"],
+        superseded_by=["memory-2"],
+        is_latest=False,
+        request_id="request-memory-1",
+    )
+
+    assert result["memory_ids"] == ["memory-1"]
+    node = client._vector_store.nodes["memory-1"]
+    assert node.content == "older preference"
+    assert node.layer == MemoryLayer.L2_FACT
+    assert node.status == MemoryStatus.SUPERSEDED
+    assert node.superseded_by == ["memory-2"]
+    assert node.is_latest is False
+    assert node.custom["input_kind"] == "memory_exact_import"
+
+
+@pytest.mark.asyncio
+async def test_exact_memory_import_refuses_same_id_with_different_body():
+    client = _client()
+    await HyMemoryClient.async_import_memory_exact.__wrapped__(
+        client,
+        "first",
+        memory_id="memory-1",
+        user_id="u1",
+        request_id="request-memory-1",
+    )
+
+    with pytest.raises(ValueError, match="different memory content"):
+        await HyMemoryClient.async_import_memory_exact.__wrapped__(
+            client,
+            "changed",
+            memory_id="memory-1",
+            user_id="u1",
+            request_id="request-memory-2",
+        )
+
+
+@pytest.mark.asyncio
+async def test_exact_memory_import_rejects_raw_layer():
+    client = _client()
+    with pytest.raises(ValueError, match="use import_raw"):
+        await HyMemoryClient.async_import_memory_exact.__wrapped__(
+            client,
+            "raw",
+            memory_id="memory-1",
+            user_id="u1",
+            layer="l1_raw",
+            request_id="request-memory-1",
+        )
+
+
+@pytest.mark.asyncio
+async def test_existing_search_contract_can_explicitly_search_raw_with_source_filter():
+    client = _client()
+    for raw_id, session_id, source in (
+        ("raw-1", "s1", "cursor"),
+        ("raw-2", "s2", "claude"),
+    ):
+        await HyMemoryClient.async_import_raw.__wrapped__(
+            client,
+            f"content {raw_id}",
+            user_id="u1",
+            agent_id="a1",
+            session_id=session_id,
+            source=source,
+            raw_id=raw_id,
+            request_id=f"request-{raw_id}",
+        )
+
+    result = await client._search_explicit_raw_layer(
+        "content",
+        user_ids=["u1"],
+        agent_ids=["a1"],
+        session_ids=["s2"],
+        include_sources=["claude"],
+        exclude_sources=["cursor"],
+        limit=5,
+        min_score=0.0,
+        created_after=None,
+        request_id="search-1",
+        t0=0.0,
+        query_embedding=[0.1, 0.2],
+    )
+
+    assert [item["memory_id"] for item in result["memories"]["normal"]] == [
+        "raw-2"
+    ]
+    assert result["memories"]["normal"][0]["layer"] == "l1_raw"
+    assert result["memories"]["normal"][0]["source"] == "claude"

tests/test_write_outcome_cache.py

@@ -0,0 +1,47 @@
+from __future__ import annotations
+
+import pytest
+
+from hy_memory.config import MemoryConfig
+from hy_memory.data.cache_sqlite import SqliteCache
+
+
+@pytest.fixture
+async def cache(tmp_path):
+    config = MemoryConfig.from_dict(
+        {"cache": {"backend": "sqlite", "db_path": str(tmp_path / "cache.sqlite3")}}
+    )
+    value = SqliteCache(config)
+    await value.initialize()
+    yield value
+    await value.close()
+
+
+@pytest.mark.asyncio
+async def test_claim_is_atomic_and_returns_existing_scope(cache):
+    claimed = await cache.claim_write_outcome("r1", "u1", "a1", "s1", "add")
+    assert claimed is None
+
+    existing = await cache.claim_write_outcome("r1", "u1", "a1", "s1", "add")
+    assert existing["status"] == "running"
+    assert existing["user_id"] == "u1"
+    assert existing["operation"] == "add"
+
+
+@pytest.mark.asyncio
+async def test_terminal_result_round_trips(cache):
+    await cache.claim_write_outcome("r2", "u1", "a1", "s1", "add_extracted")
+    assert await cache.finish_write_outcome(
+        "r2",
+        "succeeded",
+        {"success": True, "memory_ids": ["m1", "m2"]},
+    )
+
+    outcome = await cache.get_write_outcome("r2")
+    assert outcome["status"] == "succeeded"
+    assert outcome["result"] == {"success": True, "memory_ids": ["m1", "m2"]}
+
+
+@pytest.mark.asyncio
+async def test_missing_outcome_is_distinct_from_running(cache):
+    assert await cache.get_write_outcome("missing") is None

tests/test_write_outcome_client.py

@@ -0,0 +1,128 @@
+from __future__ import annotations
+
+from unittest.mock import AsyncMock
+
+import pytest
+
+from hy_memory.client import HyMemoryClient
+
+
+class OutcomeCache:
+    def __init__(self):
+        self.rows: dict[str, dict] = {}
+
+    async def claim_write_outcome(
+        self,
+        request_id,
+        user_id,
+        agent_id,
+        session_id,
+        operation,
+    ):
+        existing = self.rows.get(request_id)
+        if existing is not None:
+            return existing
+        self.rows[request_id] = {
+            "request_id": request_id,
+            "user_id": user_id,
+            "agent_id": agent_id,
+            "session_id": session_id,
+            "operation": operation,
+            "status": "running",
+            "result": {},
+        }
+        return None
+
+    async def finish_write_outcome(self, request_id, status, result):
+        row = self.rows.get(request_id)
+        if row is None:
+            return False
+        row.update(status=status, result=dict(result))
+        return True
+
+    async def get_write_outcome(self, request_id):
+        return self.rows.get(request_id)
+
+
+def _client():
+    client = HyMemoryClient.__new__(HyMemoryClient)
+    client._cache = OutcomeCache()
+    return client
+
+
+@pytest.mark.asyncio
+async def test_same_request_replays_terminal_result_without_writing_twice():
+    client = _client()
+    client._async_add_impl = AsyncMock(
+        return_value={"success": True, "request_id": "r1", "memory_id": "m1"}
+    )
+
+    first = await HyMemoryClient.async_add.__wrapped__(
+        client, "hello", user_id="u1", request_id="r1"
+    )
+    second = await HyMemoryClient.async_add.__wrapped__(
+        client, "hello", user_id="u1", request_id="r1"
+    )
+
+    assert first["memory_id"] == "m1"
+    assert second == {**first, "replayed": True}
+    client._async_add_impl.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_exception_marks_outcome_unknown_and_blocks_blind_retry():
+    client = _client()
+    client._async_add_impl = AsyncMock(side_effect=RuntimeError("after commit is unknown"))
+
+    with pytest.raises(RuntimeError):
+        await HyMemoryClient.async_add.__wrapped__(
+            client, "hello", user_id="u1", request_id="r2"
+        )
+
+    replay = await HyMemoryClient.async_add.__wrapped__(
+        client, "hello", user_id="u1", request_id="r2"
+    )
+    assert replay["success"] is False
+    assert replay["outcome_status"] == "unknown"
+    assert replay["error_code"] == "write_outcome_not_terminal"
+    client._async_add_impl.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_request_id_cannot_cross_write_scope():
+    client = _client()
+    client._async_add_impl = AsyncMock(
+        return_value={"success": True, "request_id": "r3", "memory_id": "m1"}
+    )
+    await HyMemoryClient.async_add.__wrapped__(
+        client, "hello", user_id="u1", agent_id="a1", request_id="r3"
+    )
+
+    with pytest.raises(ValueError, match="another write scope"):
+        await HyMemoryClient.async_add.__wrapped__(
+            client, "hello", user_id="u2", agent_id="a1", request_id="r3"
+        )
+
+
+@pytest.mark.asyncio
+async def test_add_extracted_uses_the_same_idempotent_outcome_contract():
+    client = _client()
+    client._async_add_extracted_impl = AsyncMock(
+        return_value={
+            "success": True,
+            "request_id": "r4",
+            "memory_id": "m1",
+            "memory_ids": ["m1"],
+        }
+    )
+
+    first = await HyMemoryClient.async_add_extracted.__wrapped__(
+        client, ["fact"], user_id="u1", request_id="r4"
+    )
+    second = await HyMemoryClient.async_add_extracted.__wrapped__(
+        client, ["fact"], user_id="u1", request_id="r4"
+    )
+
+    assert first["memory_ids"] == ["m1"]
+    assert second == {**first, "replayed": True}
+    client._async_add_extracted_impl.assert_awaited_once()

uv.lock

@@ -1715,7 +1715,7 @@ wheels = [
 
 [[package]]
 name = "hy-memory"
-version = "2.0.4"
+version = "2.0.5.dev4"
 source = { editable = "." }
 dependencies = [
     { name = "chromadb" },