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 数
 
 # ── 写入流程开关 ──

.github/workflows/linux-lifecycle.yml

@@ -122,19 +122,25 @@ jobs:
         shell: bash
         run: |
           set -euo pipefail
+          shopt -s globstar nullglob
           version=$(node -p 'require("./package.json").version')
           node scripts/verify-release-version.mjs \
             --channel stable \
             --version "$version" \
             --build-number "$(node -p 'require("./package.json").release.build_number')" \
             --tag "$GITHUB_REF_NAME"
+          mkdir -p publish-assets
           for arch in x64 arm64; do
-            archive="release-assets/memhub-runtime-v${version}-linux-${arch}.tar.gz"
-            test -f "$archive"
-            test -f "$archive.sha256"
+            name="memhub-runtime-v${version}-linux-${arch}.tar.gz"
+            archives=(release-assets/**/"$name")
+            checksums=(release-assets/**/"$name.sha256")
+            test "${#archives[@]}" -eq 1
+            test "${#checksums[@]}" -eq 1
+            cp -- "${archives[0]}" "publish-assets/$name"
+            cp -- "${checksums[0]}" "publish-assets/$name.sha256"
             (
-              cd release-assets
-              sha256sum -c "$(basename "$archive.sha256")"
+              cd publish-assets
+              sha256sum -c "$name.sha256"
             )
           done
       - name: Upload Linux assets to a draft GitHub Release
@@ -160,7 +166,7 @@ jobs:
               --title "$GITHUB_REF_NAME"
           fi
           gh release upload "$GITHUB_REF_NAME" \
-            release-assets/memhub-runtime-v*-linux-*.tar.gz \
-            release-assets/memhub-runtime-v*-linux-*.tar.gz.sha256 \
+            publish-assets/memhub-runtime-v*-linux-*.tar.gz \
+            publish-assets/memhub-runtime-v*-linux-*.tar.gz.sha256 \
             --repo "$GITHUB_REPOSITORY" \
             --clobber

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:
-                embedding, embed_meta = await self.embed_service.embed_queued_with_meta(
-                    request.content
-                )
-                s.set_output({"dims": len(embedding), "content_len": len(request.content)})
+            if existing_raw_ids:
+                embedding, embed_meta = [], {"reused_raw": True}
+            else:
+                with tracer.span("embed") as s:
+                    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)
-                    _raw_custom: Dict[str, Any] = {}
-                    self._attach_client_meta(_raw_custom, request)
-                    mem_node = MemoryNode(
-                        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)
+                    if existing_raw_ids:
+                        memory_id = existing_raw_ids[-1]
+                        mem_node = await vector_store.get_by_id(memory_id)
+                        if (
+                            mem_node is None
+                            or mem_node.layer != MemoryLayer.L1_RAW
+                            or mem_node.user_id != request.user_id
+                            or mem_node.agent_id != (request.agent_id or "default_agent")
+                            or mem_node.session_id != (request.session_id or "default_session")
+                            or mem_node.source != request.source
+                        ):
+                            raise ValueError("existing raw record is outside the write scope")
+                    else:
+                        layer_enum = MemoryLayer.from_string(suggested_layer)
+                        _raw_custom: Dict[str, Any] = {}
+                        self._attach_client_meta(_raw_custom, request)
+                        mem_node = MemoryNode(
+                            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(
-                        request=request,
-                        response=response,
-                        vector_store=vector_store,
-                        memory_id=memory_id,
-                    )
+                    if not existing_raw_ids:
+                        await self._rollback_failed_l1(
+                            request=request,
+                            response=response,
+                            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,
+    )

memhub/apps/core/src/one_memory/call_event_references.py

@@ -76,8 +76,7 @@ def _memory(db: sqlite3.Connection, reference: CallReference) -> SourceRecordRef
     if row["status"] == "deleted" or row["deleted_at"] is not None:
         return _hidden(reference)
 
-    # Same local revocation/ownership predicate as SourceReadAccess.memories,
-    # evaluated BEFORE freshness (which can read support text), in this snapshot.
+    # Same local revocation/ownership predicate as SourceReadAccess.memories.
     # Its public methods open their own connections, so cannot be used here.
     access = db.execute(
         "SELECT EXISTS (SELECT 1 FROM governed_memory_evidence "
@@ -121,8 +120,8 @@ def resolve_call_references(database_path: Path, event: CallEvent) -> list[Sourc
 
     Local-owner only: event host attribution is NOT authorization. Explicit local
     source revocations deny access; absent grants retain legacy local semantics.
-    A memory needs at least one accessible active citation. No proof preserves
-    shared freshness compatibility, not a claim of verified historical freshness.
+    A memory needs at least one accessible active citation. Evidence freshness
+    proofs are retired and do not affect availability.
 
     Any read/parse/budget failure raises a fixed whole-request error, never a
     partial result or per-item deletion. The route owner must map this separately

memhub/apps/core/src/one_memory/memory_evidence_freshness.py

@@ -1,36 +1,25 @@
-"""Read-only, exact-revision support checks inside the caller's SQLite snapshot.
+"""Memory evidence freshness checks are retired.
 
-No proof is legacy/unknown, not stale. One complete fresh proof suffices; all
-complete proofs stale means false. Malformed records raise. An exceeded budget
-means the memory is too large to verify, not that it is stale: it reads as
-current, the same as having no proof, and is reported once in a while as a
-content-free diagnostic. One such memory used to fail every COUNT over the
-library, so the homepage returned 503 for as long as it existed (D76).
-Set ONE_MEMORY_FRESHNESS_BUDGET_RAISES=1 to restore the old raising behavior.
-The caller owns BEGIN/deadline/progress-handler and error-to-unavailable mapping.
+Read paths used to re-validate ``memory_task_outputs.support_refs`` against
+live ``evidence`` rows via ``is_memory_evidence_current``. That design hid
+memories when evidence changed, but at high cost (library 503s, per-row SQL,
+segment-key breakage). Call sites still import this module; registration is
+now a no-op that always admits the row.
+
+Proof rows may still be written by the ingest path; they are ignored on read.
+See ``新鲜度_todo.md`` for the remaining teardown steps.
 """
 
-import json
-import logging
-import os
+from __future__ import annotations
+
 import sqlite3
-import threading
-import time
 from collections.abc import Iterator, Sequence
 from contextlib import contextmanager
 
 from one_memory.database_access import register_function
-from one_memory.diagnostic_log import diagnostic_event
-from one_memory.imports.models import EvidenceRecord
-from one_memory.memory_tasks import MemoryTaskOutput, matches_evidence_ref
 
-_PROOFS = 64
-_REFERENCES = 128
-_BODY_BYTES = 65_536
-_EVIDENCE_BYTES = 8 * 1024 * 1024
-_SCOPE_STEPS = 64
-
-#: Budget refusals: the check could not finish, the data is not malformed.
+#: Historical budget error codes. Retained so diagnostics/env docs do not break
+#: mid-teardown; nothing raises them on the read path anymore.
 BUDGET_ERRORS = frozenset(
     {
         "memory_freshness_proof_budget",
@@ -39,80 +28,23 @@ BUDGET_ERRORS = frozenset(
         "memory_freshness_scope_budget",
     }
 )
-_REPORT_INTERVAL_SECONDS = 60.0
-_reported: dict[str, tuple[float, int]] = {}
-_reported_lock = threading.Lock()
 
 
 def budget_errors_raise() -> bool:
-    """Opt back into the pre-D76 behavior, where one large memory fails the query."""
-    return os.environ.get("ONE_MEMORY_FRESHNESS_BUDGET_RAISES", "").strip() == "1"
+    """Former opt-in to raise on budget overrun. Always false after retirement."""
+    return False
 
 
-def _report_budget(code: str) -> None:
-    """Count every budget skip; write at most one line per reason per interval."""
-    now = time.monotonic()
-    with _reported_lock:
-        last, pending = _reported.get(code, (float("-inf"), 0))
-        pending += 1
-        if now - last < _REPORT_INTERVAL_SECONDS:
-            _reported[code] = (last, pending)
-            return
-        _reported[code] = (now, 0)
-    diagnostic_event(
-        level="warn",
-        component="memory.freshness",
-        event="memory_freshness.budget_exceeded",
-        outcome="skipped",
-        details={"reason": code, "count": pending},
+def has_proof_table(db: sqlite3.Connection) -> bool:
+    """Whether ``memory_task_outputs`` exists. Unused by the retired check."""
+    return (
+        db.execute(
+            "SELECT 1 FROM sqlite_master WHERE type='table' AND name='memory_task_outputs'"
+        ).fetchone()
+        is not None
     )
 
 
-def _object(pairs: list[tuple[str, object]]) -> dict[str, object]:
-    result: dict[str, object] = {}
-    for key, value in pairs:
-        if key in result:
-            raise ValueError("memory_freshness_invalid_proof")
-        result[key] = value
-    return result
-
-
-class _Prefetched:
-    """Prefetched rows for the current batch, private to thread and connection.
-
-    Scoped, never global: a cache that outlived the transaction would answer
-    from rows a later write had already replaced.
-    """
-
-    def __init__(self) -> None:
-        self._local = threading.local()
-
-    def get(self) -> dict[str, list[sqlite3.Row]] | None:
-        return getattr(self._local, "rows", None)
-
-    def evidence(self) -> dict[str, sqlite3.Row] | None:
-        return getattr(self._local, "evidence", None)
-
-    @contextmanager
-    def using(
-        self,
-        rows: dict[str, list[sqlite3.Row]],
-        evidence: dict[str, sqlite3.Row] | None = None,
-    ) -> Iterator[None]:
-        previous = getattr(self._local, "rows", None)
-        previous_evidence = getattr(self._local, "evidence", None)
-        self._local.rows = rows
-        self._local.evidence = evidence
-        try:
-            yield
-        finally:
-            self._local.rows = previous
-            self._local.evidence = previous_evidence
-
-
-_prefetched_proofs = _Prefetched()
-
-
 @contextmanager
 def freshness_proofs_prefetched(
     db: sqlite3.Connection,
@@ -120,328 +52,16 @@ def freshness_proofs_prefetched(
     *,
     evidence_ids: Sequence[str] = (),
 ) -> Iterator[None]:
-    """Serve the freshness check's proof reads from one query.
-
-    The decisions are unchanged — the rows come from the same transaction —
-    but the per-memory reads collapse into the one in `proof_rows_for`.
-    """
-    with _prefetched_proofs.using(
-        proof_rows_for(db, memory_ids),
-        evidence_rows_for(db, evidence_ids) if evidence_ids else None,
-    ):
-        yield
-
-
-def has_proof_table(db: sqlite3.Connection) -> bool:
-    """Whether this database keeps proofs at all.
-
-    Without the table the registered UDF answers 1 unconditionally, so there
-    is nothing to prefetch and nothing to read.
-    """
-    return (
-        db.execute(
-            "SELECT 1 FROM sqlite_master WHERE type='table' AND name='memory_task_outputs'"
-        ).fetchone()
-        is not None
-    )
-
-
-_EVIDENCE_COLUMNS = (
-    "evidence_id,user_id,source_id,workspace,project,conversation_id,"
-    "message_id,correlation_id,source_ordinal,content_kind,role,written_by,"
-    "created_at,verbatim,content_hash,revision,imported_at,updated_at"
-)
-
-#: The budget measure the single read applies, kept as one string so the
-#: batched read cannot compute it differently.
-_EVIDENCE_SIZE_SQL = (
-    "length(CAST(verbatim AS BLOB)) + "
-    "length(CAST(json_array(evidence_id,user_id,source_id,workspace,project,"
-    "conversation_id,message_id,correlation_id,source_ordinal,content_kind,"
-    "role,written_by,created_at,content_hash,revision,imported_at,updated_at) "
-    "AS BLOB))"
-)
-
-
-def evidence_rows_for(
-    db: sqlite3.Connection,
-    evidence_ids: Sequence[str],
-) -> dict[str, sqlite3.Row]:
-    """Read the cited evidence rows for a page in one query.
-
-    The check reads every citation twice — once for its size, against the
-    evidence budget, and once for the row it validates against the proof's
-    reference. Both are per citation, so both grow with the page.
-
-    The size comes from the same expression the single read uses, so the
-    budget arithmetic cannot drift between the two paths.
-    """
-    if not evidence_ids:
-        return {}
-    cursor = db.cursor()
-    cursor.row_factory = sqlite3.Row
-    by_id: dict[str, sqlite3.Row] = {}
-    for start in range(0, len(evidence_ids), 900):
-        batch = list(dict.fromkeys(evidence_ids[start : start + 900]))
-        placeholders = ",".join("?" for _ in batch)
-        for row in cursor.execute(
-            f"SELECT {_EVIDENCE_COLUMNS},{_EVIDENCE_SIZE_SQL} AS measure "
-            f"FROM evidence WHERE evidence_id IN ({placeholders})",
-            batch,
-        ):
-            by_id[str(row["evidence_id"])] = row
-    return by_id
-
-
-def proof_rows_for(
-    db: sqlite3.Connection,
-    memory_ids: Sequence[str],
-) -> dict[str, list[sqlite3.Row]]:
-    """Read the proof rows for a whole page in one query.
-
-    The freshness check reads these per memory, and that is the read path's
-    whole cost: measured on the old-user library, 160 memories cost 0.69 s in
-    the UDF against 0.08 s for the citation subquery, and grew further while
-    the library was also being written to — enough to cross the five second
-    interrupt on `/api/v1/memories` and on the map.
-
-    Returns only ids that have rows; a caller falling back to a single read
-    for anything absent keeps the same answer.
-    """
-    if not memory_ids:
-        return {}
-    if not has_proof_table(db):
-        return {}
-    cursor = db.cursor()
-    cursor.row_factory = sqlite3.Row
-    by_id: dict[str, list[sqlite3.Row]] = {}
-    for start in range(0, len(memory_ids), 400):
-        batch = list(memory_ids[start : start + 400])
-        placeholders = ",".join("?" for _ in batch)
-        for row in cursor.execute(
-            "SELECT task_id,memory_id,length(CAST(body AS BLOB)) AS size,"
-            "substr(CAST(body AS BLOB),1,?) AS body FROM memory_task_outputs "
-            f"WHERE memory_id IN ({placeholders})",
-            (_BODY_BYTES + 1, *batch),
-        ):
-            by_id.setdefault(str(row["memory_id"]), []).append(row)
-    return by_id
-
-
-def _proof_revision(
-    cursor: sqlite3.Cursor,
-    memory_id: str,
-    revision: int,
-    user_id: str,
-    available: set[int],
-) -> int:
-    if revision in available or not available:
-        return revision
-    if (
-        cursor.execute(
-            "SELECT 1 FROM sqlite_master WHERE type='table' AND name='governed_memory_versions'"
-        ).fetchone()
-        is None
-    ):
-        return revision
-    original = revision
-    for step in range(_SCOPE_STEPS + 1):
-        if revision in available:
-            return revision
-        row = cursor.execute(
-            "SELECT v.action, typeof(v.content)='text' AND typeof(p.content)='text' "
-            "AND CAST(v.content AS BLOB)=CAST(p.content AS BLOB) AS same_content, "
-            "v.status=p.status AS same_status, "
-            "CAST(v.content AS BLOB)=CAST(m.content AS BLOB) AS current_content "
-            "FROM governed_memory_versions v "
-            "JOIN governed_memory_versions p ON p.memory_id=v.memory_id "
-            "AND p.revision=v.revision-1 "
-            "JOIN governed_memories m ON m.memory_id=v.memory_id "
-            "WHERE v.memory_id=? AND v.revision=? AND m.user_id=? AND m.revision=?",
-            (memory_id, revision, user_id, original),
-        ).fetchone()
-        if row is None or row[0] != "set_scopes" or not all(row[index] == 1 for index in (1, 2, 3)):
-            return original
-        if step == _SCOPE_STEPS:
-            raise ValueError("memory_freshness_scope_budget")
-        revision -= 1
-    raise ValueError("memory_freshness_scope_budget")
+    """No-op. Kept so batched read callers need not change in this PR."""
+    del db, memory_ids, evidence_ids
+    yield
 
 
 def register_memory_freshness(db: sqlite3.Connection) -> str:
-    """Register UDF; return a predicate for alias m. Does not change row_factory.
+    """Register a always-true UDF; return SQL predicate ``1``.
 
-    Production initialization supplies memory_task_outputs(memory_id) index.
-    Old no-index databases depend on the caller's SQL deadline. Per evaluation:
-    65 rows / 64KiB each, 129 active IDs, 8MiB total Evidence material. No library
-    bodies are loaded; Evidence objects live only for this evaluation. No cache
-    persists across transactions (safe when a caller reuses its connection).
+    Callers that embed the return value in ``WHERE`` keep working. Callers that
+    invoke ``is_memory_evidence_current`` by name also get current.
     """
-    if not has_proof_table(db):
-        register_function(db, "is_memory_evidence_current", 3, lambda *_: 1)
-        return "1"
-
-    raising = budget_errors_raise()
-
-    def current(memory_id: str, revision: int, user_id: str) -> int:
-        try:
-            return checked(memory_id, revision, user_id)
-        except ValueError as error:
-            code = str(error)
-            if raising or code not in BUDGET_ERRORS:
-                raise
-            _report_budget(code)
-            return 1
-        except sqlite3.Error as error:
-            # A freshness proof is an enhancement, never the reason a person
-            # cannot see their library. Under a concurrent history import the
-            # 5s read deadline interrupts this UDF mid-statement; the old
-            # behaviour let that escape, and the caller turned it into
-            # `memory_library_unavailable`, so the memory list stayed empty for
-            # as long as the import ran. Answer "current" instead and record
-            # why, which is the same trade the D76 budget skip already makes.
-            _report_budget("memory_freshness_sql_error")
-            logging.getLogger(__name__).warning(
-                "memory_freshness_udf_failed %s: %s", type(error).__name__, error
-            )
-            return 1
-
-    def checked(memory_id: str, revision: int, user_id: str) -> int:
-        if not isinstance(memory_id, str) or type(revision) is not int or revision < 1:
-            raise ValueError("memory_freshness_invalid_identity")
-        if not isinstance(user_id, str):
-            raise ValueError("memory_freshness_invalid_identity")
-        cursor = db.cursor()
-        cursor.row_factory = sqlite3.Row
-        prefetched = _prefetched_proofs.get()
-        if prefetched is not None:
-            rows = prefetched.get(memory_id, ())
-        else:
-            rows = cursor.execute(
-                "SELECT task_id,memory_id,length(CAST(body AS BLOB)) AS size,"
-                "substr(CAST(body AS BLOB),1,?) AS body FROM memory_task_outputs "
-                "WHERE memory_id=? LIMIT ?",
-                (_BODY_BYTES + 1, memory_id, _PROOFS + 1),
-            ).fetchall()
-        rows = list(rows)
-        if len(rows) > _PROOFS:
-            raise ValueError("memory_freshness_proof_budget")
-        decoded = []
-        for row in rows:
-            if type(row["size"]) is not int or row["size"] > _BODY_BYTES:
-                raise ValueError("memory_freshness_proof_budget")
-            body = json.loads(row["body"], object_pairs_hook=_object)
-            if not isinstance(body, dict) or type(body.get("memory_revision")) is not int:
-                raise ValueError("memory_freshness_invalid_proof")
-            decoded.append((row, body))
-        revision = _proof_revision(
-            cursor,
-            memory_id,
-            revision,
-            user_id,
-            {body["memory_revision"] for _, body in decoded},
-        )
-        applicable = []
-        for row, body in decoded:
-            if body["memory_revision"] != revision:
-                continue
-            raw_refs = body.get("support_refs", [])
-            if not isinstance(raw_refs, list) or any(
-                not isinstance(ref, dict)
-                or set(ref) - {"evidence_id", "revision", "content_hash", "identity_hash"}
-                for ref in raw_refs
-            ):
-                raise ValueError("memory_freshness_invalid_reference")
-            output = MemoryTaskOutput.model_validate_json(row["body"], strict=True)
-            if output.memory_id != memory_id or output.task_id != row["task_id"]:
-                raise ValueError("memory_freshness_proof_identity")
-            if len(output.support_refs) > _REFERENCES:
-                raise ValueError("memory_freshness_reference_budget")
-            for ref in output.support_refs:
-                if (
-                    set(ref) - {"evidence_id", "revision", "content_hash", "identity_hash"}
-                    or not ref["evidence_id"]
-                    or ref["revision"] < 1
-                    or any(
-                        len(ref[key]) != 64
-                        or any(char not in "0123456789abcdef" for char in ref[key])
-                        for key in ("content_hash", "identity_hash")
-                        if key in ref
-                    )
-                ):
-                    raise ValueError("memory_freshness_invalid_reference")
-            proof_ids = [ref["evidence_id"] for ref in output.support_refs]
-            if len(proof_ids) != len(set(proof_ids)):
-                raise ValueError("memory_freshness_duplicate_reference")
-            applicable.append(output)
-        if not applicable:
-            return 1
-        active = cursor.execute(
-            "SELECT evidence_id FROM governed_memory_evidence "
-            "WHERE memory_id=? AND active=1 LIMIT ?",
-            (memory_id, _REFERENCES + 1),
-        ).fetchall()
-        if len(active) > _REFERENCES:
-            raise ValueError("memory_freshness_reference_budget")
-        ids = {row[0] for row in active}
-        complete = [
-            output
-            for output in applicable
-            if output.support_refs and {ref["evidence_id"] for ref in output.support_refs} == ids
-        ]
-        if not complete:
-            return 1
-        records: dict[str, EvidenceRecord | None] = {}
-        used = 0
-        prefetched_evidence = _prefetched_proofs.evidence()
-        for identifier in ids:
-            evidence_row = (
-                prefetched_evidence.get(identifier) if prefetched_evidence is not None else None
-            )
-            if evidence_row is not None:
-                size = evidence_row["measure"]
-                if type(size) is not int or size < 0:
-                    raise ValueError("memory_freshness_invalid_evidence")
-            else:
-                measured = cursor.execute(
-                    f"SELECT {_EVIDENCE_SIZE_SQL} FROM evidence WHERE evidence_id=?",
-                    (identifier,),
-                ).fetchone()
-                if measured is None:
-                    records[identifier] = None
-                    continue
-                if type(measured[0]) is not int or measured[0] < 0:
-                    raise ValueError("memory_freshness_invalid_evidence")
-                size = measured[0]
-            used += size
-            if used > _EVIDENCE_BYTES:
-                raise ValueError("memory_freshness_evidence_budget")
-            if evidence_row is None:
-                evidence_row = cursor.execute(
-                    f"SELECT {_EVIDENCE_COLUMNS} FROM evidence WHERE evidence_id=?",
-                    (identifier,),
-                ).fetchone()
-                if evidence_row is None:
-                    records[identifier] = None
-                    continue
-            # evidence_ref explicitly excludes origin_meta/imported_at/updated_at.
-            # `sqlite3.Row` iterates values, not column names, so pair the
-            # names with the values explicitly.
-            fields = {
-                name: value
-                for name, value in zip(evidence_row.keys(), tuple(evidence_row), strict=False)
-                if name != "measure"
-            }
-            records[identifier] = EvidenceRecord.model_validate({**fields, "origin_meta": {}})
-        for output in complete:
-            if all(
-                (record := records[ref["evidence_id"]]) is not None
-                and record.user_id == user_id
-                and matches_evidence_ref(record, ref)
-                for ref in output.support_refs
-            ):
-                return 1
-        return 0
-
-    register_function(db, "is_memory_evidence_current", 3, current)
-    return "is_memory_evidence_current(m.memory_id,m.revision,m.user_id)"
+    register_function(db, "is_memory_evidence_current", 3, lambda *_: 1)
+    return "1"

memhub/apps/core/tests/TEST_INDEX.md

@@ -203,7 +203,7 @@
 | `test_memory_correction_sync.py` | `integration` | R-27: a correction that only lands locally, while HY keeps the old text. |
 | `test_memory_deletion_outbox.py` | `contract` | 删除发件箱：一条删除同一时刻只归一个执行者，结果只由持有领取凭证的那个写。 |
 | `test_memory_dispatch_lifecycle.py` | `contract` | Exercise the automatic enqueue boundary without producing substitute memories. |
-| `test_memory_evidence_freshness.py` | `contract` | 覆盖 fresh then revised and identity only change、any complete fresh proof wins over old proof、other revision and incomplete proof do not hide legacy or correction 等 21 个场景。 |
+| `test_memory_evidence_freshness.py` | `contract` | 新鲜度拆除后：读路径恒为 current；证据修订 / 畸形 proof / segment ref 不再藏记忆或 503。 |
 | `test_memory_freshness_integration_review.py` | `integration` | Independent synthetic read-path review; no lifespan, model or user database. |
 | `test_memory_governance.py` | `contract` | 覆盖 correction keeps verbatim and appends history、mutations are revision checked and request idempotent、delete result survives restart and request replay 等 42 个场景。 |
 | `test_memory_governance_api.py` | `integration` | 覆盖 memory governance api preserves evidence and reports cleanup、memory validation does not echo content 场景。 |

memhub/apps/core/tests/test_call_event_references.py

@@ -355,8 +355,9 @@ def test_actual_evidence_proof_revision_and_corrected_memory(fresh):
     assert all(item.available for item in resolve_call_references(path, event(raw, reference)))
     evidence.put(value.model_copy(update={"verbatim": "revised evidence"}))
     result = resolve_call_references(path, event(raw, reference))
+    # Raw refs still track evidence.revision; memory refs no longer use freshness.
     assert_hidden(result[0], raw, "changed")
-    assert_hidden(result[1], reference, "changed")
+    assert result[1].available
     corrected = as_person(MemoryGovernanceStore(path)).correct(
         memory.memory_id,
         user_id="local",
@@ -369,7 +370,7 @@ def test_actual_evidence_proof_revision_and_corrected_memory(fresh):
     assert resolve_call_references(path, event(current))[0].title == corrected.content
 
 
-def test_scope_only_revision_still_checks_original_support(fresh):
+def test_scope_only_revision_keeps_memory_readable_when_evidence_changes(fresh):
     path, evidence, value, record, memory, tasks, output = fresh
     updated = as_person(MemoryGovernanceStore(path)).set_scopes(
         memory.memory_id,
@@ -381,25 +382,27 @@ def test_scope_only_revision_still_checks_original_support(fresh):
     reference = memory_ref(updated.memory_id, updated.revision)
     assert resolve_call_references(path, event(reference))[0].available
     evidence.put(value.model_copy(update={"verbatim": "revised evidence"}))
-    assert_hidden(resolve_call_references(path, event(reference))[0], reference, "changed")
+    assert resolve_call_references(path, event(reference))[0].available
 
 
-def test_freshness_identity_change_without_revision_change(fresh):
+def test_evidence_identity_change_without_revision_keeps_memory_readable(fresh):
     path, evidence, value, record, memory, tasks, output = fresh
     reference = memory_ref(memory.memory_id, memory.revision)
     assert resolve_call_references(path, event(reference))[0].available
     with closing(sqlite3.connect(path)) as db, db:
         db.execute("UPDATE evidence SET workspace='workspace:changed'")
-    assert_hidden(resolve_call_references(path, event(reference))[0], reference, "changed")
+    assert resolve_call_references(path, event(reference))[0].available
 
 
-def test_invalid_freshness_proof_fails_whole_response_but_revoked_never_reads_it(fresh):
+def test_invalid_proof_is_ignored_but_revocation_still_hides_memory(fresh):
     path, evidence, value, record, memory, tasks, output = fresh
     reference = memory_ref(memory.memory_id, memory.revision)
     with closing(sqlite3.connect(path)) as db, db:
         db.execute("UPDATE memory_task_outputs SET body='invalid secret proof'")
-    with pytest.raises(CallReferencesUnavailable, match="^call_references_unavailable$"):
-        resolve_call_references(path, event(ref(key=record.evidence_id), reference))
+    assert all(
+        item.available
+        for item in resolve_call_references(path, event(ref(key=record.evidence_id), reference))
+    )
     with closing(sqlite3.connect(path)) as db, db:
         grant(db, "cursor")
     result = resolve_call_references(path, event(reference))
@@ -455,9 +458,13 @@ def test_authorization_revision_freshness_and_titles_share_one_snapshot(fresh, m
     assert [item.title for item in snapshot] == [memory.content, record.verbatim]
     assert changed == [True] and reads and all(reads)
     current = resolve_call_references(path, event(reference, raw))
-    assert_hidden(
-        current[0], reference, "changed" if change in ("evidence", "memory") else "unavailable"
-    )
+    # Memory refs no longer hide on evidence revision/identity via freshness.
+    if change == "evidence":
+        assert current[0].available
+    else:
+        assert_hidden(
+            current[0], reference, "changed" if change == "memory" else "unavailable"
+        )
     assert current[1].available == (change in ("memory", "link"))
 
 

memhub/apps/core/tests/test_memory_evidence_freshness.py

@@ -1,3 +1,5 @@
+"""Freshness retirement: read path always admits, even when evidence changed."""
+
 import json
 import sqlite3
 
@@ -5,7 +7,11 @@ import pytest
 
 from one_memory.imports.models import EvidenceInput, EvidenceOrigin, OriginMetadata, WrittenBy
 from one_memory.imports.storage import EvidenceStore
-from one_memory.memory_evidence_freshness import register_memory_freshness
+from one_memory.memory_evidence_freshness import (
+    budget_errors_raise,
+    freshness_proofs_prefetched,
+    register_memory_freshness,
+)
 from one_memory.memory_governance import MemoryDraft, MemoryGovernanceStore, MemoryKind
 from one_memory.memory_tasks import MemoryTaskOutput, MemoryTaskStore, evidence_ref
 from tests.person_door import as_person
@@ -61,46 +67,45 @@ def check(path, memory):
         ).fetchone()[0]
 
 
-def test_fresh_then_revised_and_identity_only_change(case):
+def test_register_returns_constant_true_predicate(case):
+    path, evidence, value, record, memory, tasks, output = case
+    with sqlite3.connect(path) as db:
+        db.execute("BEGIN")
+        assert register_memory_freshness(db) == "1"
+        assert db.execute("SELECT is_memory_evidence_current(?,?,?)", ("x", 1, "local")).fetchone() == (
+            1,
+        )
+
+
+def test_revised_evidence_no_longer_hides_memory(case):
     path, evidence, value, record, memory, tasks, output = case
     assert check(path, memory) == 1
     evidence.put(value.model_copy(update={"verbatim": "new statement"}))
-    assert check(path, memory) == 0
-
-
-def test_any_complete_fresh_proof_wins_over_old_proof(case):
-    path, evidence, value, record, memory, tasks, output = case
-    revised = evidence.put(value.model_copy(update={"verbatim": "new statement"})).record
-    tasks.save_output(
-        output.model_copy(update={"task_id": "task2", "support_refs": (evidence_ref(revised),)})
-    )
     assert check(path, memory) == 1
 
 
-def test_other_revision_and_incomplete_proof_do_not_hide_legacy_or_correction(case):
-    path, evidence, value, record, memory, tasks, output = case
-    evidence.put(value.model_copy(update={"verbatim": "new statement"}))
-    tasks.save_output(output.model_copy(update={"memory_revision": memory.revision + 1}))
-    assert check(path, memory) == 1
-    tasks.save_output(output.model_copy(update={"support_refs": ()}))
-    assert check(path, memory) == 1
-
-
-@pytest.mark.parametrize("broken", ["json", "identity", "ref"])
-def test_bad_proof_raises_not_false_empty(case, broken):
+def test_segment_support_refs_do_not_fail_reads(case):
     path, evidence, value, record, memory, tasks, output = case
     body = output.model_dump(mode="json")
-    if broken == "identity":
-        body["memory_id"] = "other"
-    if broken == "ref":
-        body["support_refs"][0]["revision"] = "1"
-    payload = json.dumps(body)
-    if broken == "json":
-        payload = "{"
+    body["support_refs"][0].update(
+        segment_index=0,
+        segment_start=0,
+        segment_end=len(record.verbatim),
+        segment_hash="0" * 64,
+    )
     with sqlite3.connect(path) as db:
-        db.execute("UPDATE memory_task_outputs SET body=?", (payload,))
-    with pytest.raises(sqlite3.OperationalError):
-        check(path, memory)
+        db.execute("UPDATE memory_task_outputs SET body=?", (json.dumps(body),))
+    assert check(path, memory) == 1
+
+
+def test_malformed_proof_does_not_fail_reads(case):
+    path, evidence, value, record, memory, tasks, output = case
+    with sqlite3.connect(path) as db:
+        db.execute(
+            "UPDATE memory_task_outputs SET body=? WHERE memory_id=?",
+            ("{", memory.memory_id),
+        )
+    assert check(path, memory) == 1
 
 
 def test_missing_table_compatibility_is_readonly():
@@ -111,239 +116,19 @@ def test_missing_table_compatibility_is_readonly():
         assert db.execute("SELECT name FROM sqlite_master").fetchall() == []
 
 
-def test_table_presence_is_checked_once_at_registration(case):
+def test_prefetch_is_noop(case):
     path, evidence, value, record, memory, tasks, output = case
     with sqlite3.connect(path) as db:
         db.execute("BEGIN")
-        statements = []
-        db.set_trace_callback(statements.append)
-        predicate = register_memory_freshness(db)
-        for _ in range(3):
-            assert db.execute(f"SELECT {predicate} FROM governed_memories m").fetchone() == (1,)
-        assert sum("sqlite_master" in statement for statement in statements) == 1
+        with freshness_proofs_prefetched(db, [memory.memory_id], evidence_ids=[record.evidence_id]):
+            assert register_memory_freshness(db) == "1"
+            assert check(path, memory) == 1
 
 
-def test_proof_budget_reads_current_instead_of_failing(case):
-    path, evidence, value, record, memory, tasks, output = case
-    for index in range(65):
-        tasks.save_output(output.model_copy(update={"task_id": f"task-{index}"}))
-    assert check(path, memory) == 1
-
-
-def test_proof_budget_still_raises_when_opted_back(case, monkeypatch):
+def test_budget_raise_env_is_inert(case, monkeypatch):
     path, evidence, value, record, memory, tasks, output = case
     monkeypatch.setenv("ONE_MEMORY_FRESHNESS_BUDGET_RAISES", "1")
+    assert budget_errors_raise() is False
     for index in range(65):
         tasks.save_output(output.model_copy(update={"task_id": f"task-{index}"}))
-    with pytest.raises(sqlite3.OperationalError):
-        check(path, memory)
-
-
-def test_snapshot_cache_does_not_hide_changes_across_transactions(case):
-    path, evidence, value, record, memory, tasks, output = case
-    with sqlite3.connect(path) as db:
-        db.execute("BEGIN")
-        predicate = register_memory_freshness(db)
-        query = f"SELECT {predicate} FROM governed_memories m"
-        assert db.execute(query).fetchone() == (1,)
-        db.commit()
-        evidence.put(value.model_copy(update={"verbatim": "new statement"}))
-        db.execute("BEGIN")
-        register_memory_freshness(db)
-        assert db.execute(query).fetchone() == (0,)
-
-
-def test_identity_hash_detects_metadata_change_without_revision_or_content_change(case):
-    path, evidence, value, record, memory, tasks, output = case
-    with sqlite3.connect(path) as db:
-        db.execute("UPDATE evidence SET workspace='workspace:changed'")
-    assert check(path, memory) == 0
-
-
-def test_incomplete_fresh_proof_cannot_override_complete_stale(case):
-    path, evidence, value, record, memory, tasks, output = case
-    evidence.put(value.model_copy(update={"verbatim": "new statement"}))
-    tasks.save_output(output.model_copy(update={"task_id": "partial", "support_refs": ()}))
-    assert check(path, memory) == 0
-
-
-def test_sql_query_only_and_missing_proof_compatibility(case):
-    path, evidence, value, record, memory, tasks, output = case
-    with sqlite3.connect(path) as db:
-        db.execute("DELETE FROM memory_task_outputs")
-    with sqlite3.connect(path) as db:
-        db.execute("PRAGMA query_only=ON")
-        db.execute("BEGIN")
-        before = db.total_changes
-        predicate = register_memory_freshness(db)
-        assert db.execute(f"SELECT {predicate} FROM governed_memories m").fetchone() == (1,)
-        assert db.total_changes == before
-
-
-def test_oversized_evidence_reads_current_without_loading_body(case):
-    path, evidence, value, record, memory, tasks, output = case
-    with sqlite3.connect(path) as db:
-        db.execute("UPDATE evidence SET verbatim=zeroblob(?)", (8 * 1024 * 1024 + 1,))
     assert check(path, memory) == 1
-
-
-@pytest.mark.parametrize("revise_first", [False, True])
-def test_scope_only_chain_preserves_old_support_guard(case, revise_first):
-    path, evidence, value, record, memory, tasks, output = case
-    governance = MemoryGovernanceStore(path)
-    if revise_first:
-        evidence.put(value.model_copy(update={"verbatim": "new statement"}))
-    for index in range(3):
-        memory = as_person(governance).set_scopes(
-            memory.memory_id,
-            user_id="local",
-            request_id=f"scope-{index}",
-            expected_revision=memory.revision,
-            scopes=(f"workspace:{index}",),
-        )
-    if not revise_first:
-        assert check(path, memory) == 1
-        evidence.put(value.model_copy(update={"verbatim": "new statement"}))
-    assert check(path, memory) == 0
-
-
-@pytest.mark.parametrize("action", ["correct", "undo", "hy_reconcile", "unknown"])
-def test_scope_chain_does_not_cross_non_scope_action(case, action):
-    path, evidence, value, record, memory, tasks, output = case
-    governance = MemoryGovernanceStore(path)
-    memory = as_person(governance).correct(
-        memory.memory_id,
-        user_id="local",
-        request_id="correct",
-        expected_revision=memory.revision,
-        content=memory.content,
-    )
-    with sqlite3.connect(path) as db:
-        db.execute(
-            "UPDATE governed_memory_versions SET action=? WHERE revision=?",
-            (action, memory.revision),
-        )
-    memory = as_person(governance).set_scopes(
-        memory.memory_id,
-        user_id="local",
-        request_id="scope",
-        expected_revision=memory.revision,
-        scopes=("workspace:changed",),
-    )
-    evidence.put(value.model_copy(update={"verbatim": "new statement"}))
-    assert check(path, memory) == 1
-
-
-def test_scope_chain_budget_reads_current(case):
-    path, evidence, value, record, memory, tasks, output = case
-    governance = MemoryGovernanceStore(path)
-    for index in range(65):
-        memory = as_person(governance).set_scopes(
-            memory.memory_id,
-            user_id="local",
-            request_id=f"scope-{index}",
-            expected_revision=memory.revision,
-            scopes=(f"workspace:{index}",),
-        )
-    assert check(path, memory) == 1
-
-
-def test_changed_content_in_scope_history_is_not_inherited(case):
-    path, evidence, value, record, memory, tasks, output = case
-    governance = MemoryGovernanceStore(path)
-    memory = as_person(governance).set_scopes(
-        memory.memory_id,
-        user_id="local",
-        request_id="scope",
-        expected_revision=memory.revision,
-        scopes=("workspace:changed",),
-    )
-    with sqlite3.connect(path) as db:
-        db.execute(
-            "UPDATE governed_memory_versions SET content='different' WHERE revision=?",
-            (output.memory_revision,),
-        )
-    evidence.put(value.model_copy(update={"verbatim": "new statement"}))
-    assert check(path, memory) == 1
-
-
-def test_scope_chain_keeps_large_governed_content_inside_sql(case):
-    path, evidence, value, record, memory, tasks, output = case
-    with sqlite3.connect(path) as db:
-        content = "x" * 1_000_000
-        db.execute("UPDATE governed_memories SET content=?", (content,))
-        db.execute("UPDATE governed_memory_versions SET content=?", (content,))
-    governance = MemoryGovernanceStore(path)
-    memory = as_person(governance).set_scopes(
-        memory.memory_id,
-        user_id="local",
-        request_id="scope",
-        expected_revision=memory.revision,
-        scopes=("workspace:changed",),
-    )
-    evidence.put(value.model_copy(update={"verbatim": "new statement"}))
-
-    class BoundedCursor(sqlite3.Cursor):
-        def fetchone(self):
-            row = super().fetchone()
-            if row is not None:
-                assert all(not isinstance(item, str) or len(item) < 65_536 for item in row)
-            return row
-
-    class BoundedConnection(sqlite3.Connection):
-        def cursor(self, factory=BoundedCursor):
-            return super().cursor(factory)
-
-    with sqlite3.connect(path, factory=BoundedConnection) as db:
-        db.execute("BEGIN")
-        predicate = register_memory_freshness(db)
-        assert db.execute(f"SELECT {predicate} FROM governed_memories m").fetchone() == (0,)
-
-
-def _confirmed_with_oversized_proof(case):
-    path, evidence, value, record, memory, tasks, output = case
-    with sqlite3.connect(path) as db:
-        db.execute(
-            "UPDATE memory_task_outputs SET body=? WHERE memory_id=?",
-            (
-                json.dumps({"memory_revision": memory.revision, "pad": "x" * 70_000}),
-                memory.memory_id,
-            ),
-        )
-    return path, memory
-
-
-def test_one_oversized_memory_no_longer_fails_the_library_count(case):
-    """D76: one memory whose proof is too large used to 503 the homepage forever."""
-    path, memory = _confirmed_with_oversized_proof(case)
-    with sqlite3.connect(path) as db:
-        db.execute("BEGIN")
-        predicate = register_memory_freshness(db)
-        count = db.execute(
-            f"SELECT COUNT(*) FROM governed_memories m WHERE {predicate}"
-        ).fetchone()[0]
-    assert count == 1
-
-
-def test_malformed_proof_still_raises(case):
-    path, evidence, value, record, memory, tasks, output = case
-    with sqlite3.connect(path) as db:
-        db.execute(
-            "UPDATE memory_task_outputs SET body=? WHERE memory_id=?", ("{}", memory.memory_id)
-        )
-    with pytest.raises(sqlite3.OperationalError):
-        check(path, memory)
-
-
-def test_budget_skip_is_reported_without_content(case, monkeypatch):
-    import one_memory.memory_evidence_freshness as freshness
-
-    events = []
-    monkeypatch.setattr(freshness, "diagnostic_event", lambda **values: events.append(values))
-    monkeypatch.setattr(freshness, "_reported", {})
-    path, memory = _confirmed_with_oversized_proof(case)
-    for _ in range(3):
-        assert check(path, memory) == 1
-    assert len(events) == 1
-    assert events[0]["details"] == {"reason": "memory_freshness_proof_budget", "count": 1}
-    assert "pad" not in json.dumps(events[0])

memhub/apps/core/tests/test_memory_freshness_integration_review.py

@@ -102,17 +102,11 @@ def revise(case):
     ).record
 
 
-def test_complete_stale_proof_hides_all_four_reads_without_mutation(case):
-    # D22: the app-facing totals (library total, homepage count) count with
-    # the cheap permission filters only, so a stale proof no longer removes a
-    # row from the totals; the freshness gate still hides it from every read
-    # that returns rows (context, source read access).
+def test_revised_evidence_remains_readable_after_freshness_retirement(case):
+    """Evidence revision no longer hides library / context / source reads."""
     assert visible(case) == (1, 1, 1, 1)
     revise(case)
-    total, home, context_items, source_items = visible(case)
-    # D22: home total counts cheaply and may keep the stale row; the reads
-    # that return rows still hide it.
-    assert (home, context_items, source_items) == (1, 0, 0)
+    assert visible(case) == (1, 1, 1, 1)
     assert case.governance.get(case.memory.memory_id, user_id="local") == case.memory
     assert case.tasks.outputs(case.output.task_id) == (case.output,)
 
@@ -120,11 +114,10 @@ def test_complete_stale_proof_hides_all_four_reads_without_mutation(case):
 @pytest.mark.parametrize(
     "revise_first", [False, True], ids=["scope-before-revision", "scope-after-revision"]
 )
-def test_scope_only_revision_must_not_resurrect_stale_derived_content(case, revise_first):
+def test_scope_only_revision_keeps_memory_readable_after_evidence_change(case, revise_first):
     if revise_first:
         revise(case)
-        _, home, context_items, source_items = visible(case)
-        assert (context_items, source_items) == (0, 0)
+        assert visible(case) == (1, 1, 1, 1)
     changed = as_person(case.governance).set_scopes(
         case.memory.memory_id,
         user_id="local",
@@ -136,8 +129,7 @@ def test_scope_only_revision_must_not_resurrect_stale_derived_content(case, revi
     if not revise_first:
         assert visible(case, changed) == (1, 1, 1, 1)
         revise(case)
-    _, home, context_items, source_items = visible(case, changed)
-    assert (context_items, source_items) == (0, 0)
+    assert visible(case, changed) == (1, 1, 1, 1)
 
 
 def test_explicit_correct_remains_readable_but_never_overrides_revocation(case):
@@ -152,7 +144,10 @@ def test_explicit_correct_remains_readable_but_never_overrides_revocation(case):
     assert visible(case, corrected) == (1, 1, 1, 1)
     with closing(sqlite3.connect(case.path)) as db, db:
         db.execute(
-            "INSERT INTO source_grants VALUES ('local','cursor','/fixture','default',?,?)",
+            """INSERT INTO source_grants
+            (user_id,source_id,root_path,requested_path,root_identity,
+             candidate_kind,granted_at,revoked_at,revision)
+            VALUES ('local','cursor','/fixture','/fixture','fixture','default',?,?,1)""",
             ("2026-09-20T00:00:00Z", "2026-09-20T00:00:01Z"),
         )
     assert visible(case, corrected) == (0, 0, 0, 0)
@@ -181,7 +176,10 @@ def test_revocation_before_freshness_snapshot_cannot_release_memory(case, monkey
         # Revocation commits before any SELECT pins the freshness transaction.
         with closing(sqlite3.connect(case.path)) as writer, writer:
             writer.execute(
-                "INSERT INTO source_grants VALUES ('local','cursor','/fixture','default',?,?)",
+                """INSERT INTO source_grants
+                (user_id,source_id,root_path,requested_path,root_identity,
+                 candidate_kind,granted_at,revoked_at,revision)
+                VALUES ('local','cursor','/fixture','/fixture','fixture','default',?,?,1)""",
                 ("2026-09-20T00:00:00Z", "2026-09-20T00:00:01Z"),
             )
         changes.append(True)
@@ -194,7 +192,7 @@ def test_revocation_before_freshness_snapshot_cannot_release_memory(case, monkey
     assert returned == []
 
 
-def test_udf_uses_caller_snapshot_then_observes_revision_in_next_transaction(case):
+def test_udf_stays_current_across_evidence_revision(case):
     with closing(sqlite3.connect(case.path)) as reader:
         reader.execute("PRAGMA journal_mode=WAL")
         reader.execute("BEGIN")
@@ -206,7 +204,7 @@ def test_udf_uses_caller_snapshot_then_observes_revision_in_next_transaction(cas
         reader.commit()
         reader.execute("BEGIN")
         register_memory_freshness(reader)
-        assert reader.execute(sql).fetchone() == (0,)
+        assert reader.execute(sql).fetchone() == (1,)
 
 
 def _oversize(case, bad):
@@ -227,20 +225,16 @@ def test_material_too_large_to_verify_reads_as_current(case, bad):
 
 
 @pytest.mark.parametrize("bad", ["duplicate-key"])
-def test_bad_material_is_unavailable_not_successful_empty(case, bad):
+def test_bad_proof_material_no_longer_makes_reads_unavailable(case, bad):
     with closing(sqlite3.connect(case.path)) as db, db:
         db.execute(
             "UPDATE memory_task_outputs SET body=?",
             ('{"memory_revision":2,"memory_revision":2}',),
         )
+    assert visible(case) == (1, 1, 1, 1)
     result = context(case)
-    assert result.status == "unavailable" and result.reason == "read_unavailable"
-    assert result.items == ()
-    for read in (read_memory_library, read_home_summary):
-        with pytest.raises(sqlite3.Error):
-            read(case.path)
-    with pytest.raises(sqlite3.Error):
-        SourceReadAccess(case.path).memories([case.memory])
+    assert result.status != "unavailable"
+    assert len(result.items) == 1
 
 
 def test_route_serves_memories_whose_proof_is_too_large(case, tmp_path):

memhub/apps/core/tests/test_memory_library.py

@@ -73,8 +73,9 @@ def put(
 def grant(db, source, *, user="local", revoked=NOW):
     db.execute(
         """INSERT OR REPLACE INTO source_grants
-        (user_id,source_id,root_path,candidate_kind,granted_at,revoked_at)
-        VALUES (?,?, '/synthetic', 'default', ?, ?)""",
+        (user_id,source_id,root_path,requested_path,root_identity,
+         candidate_kind,granted_at,revoked_at,revision)
+        VALUES (?,?, '/synthetic', '/synthetic', 'synthetic', 'default', ?, ?, 1)""",
         (user, source, NOW, revoked),
     )
 

memhub/scripts/install-docs.test.mjs

@@ -12,8 +12,8 @@ test("installation docs disclose publication and lifecycle boundaries", async ()
     "utf8",
   );
 
-  assert.match(readme, /公共 Python SDK 为 `hy-memory==2\.0\.3`/);
-  assert.match(readme, /`v1\.0\.3` 的发行目标是 Linux x64\/arm64 Runtime/);
+  assert.match(readme, /公共 Python SDK 为 `hy-memory==2\.0\.4`/);
+  assert.match(readme, /`v1\.0\.4` 的发行目标是 Linux x64\/arm64 Runtime/);
   assert.match(readme, /Linux 制品由 GitHub Actions/);
   assert.match(readme, /macOS.*Developer ID 签名、公证/s);
   assert.match(readme, /release-manifest\.json/);

memhub/scripts/platform/linux/linux-runtime-scripts.test.mjs

@@ -117,9 +117,19 @@ describe("Linux runtime scripts", () => {
       workflow,
       /verify-release-version\.mjs[\s\S]*--channel stable/,
     );
+    assert.match(workflow, /shopt -s globstar nullglob/);
+    assert.match(workflow, /archives=\(release-assets\/\*\*\/"\$name"\)/);
+    assert.match(
+      workflow,
+      /cp -- "\$\{archives\[0\]\}" "publish-assets\/\$name"/,
+    );
     assert.match(workflow, /sha256sum -c/);
     assert.match(workflow, /--json isDraft[\s\S]*--jq \.isDraft/);
     assert.match(workflow, /gh release create[\s\S]*--draft/);
     assert.match(workflow, /gh release upload[\s\S]*--clobber/);
+    assert.match(
+      workflow,
+      /publish-assets\/memhub-runtime-v\*-linux-\*\.tar\.gz/,
+    );
   });
 });

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,
-        is_latest=True,
+        status=status,
+        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" },

新鲜度_todo.md

@@ -0,0 +1,70 @@
+# 新鲜度（memory evidence freshness）拆除 TODO
+
+状态：PR1 已开（读路径短路）——分支 `chore/retire-memory-freshness`。#86 不合并。
+
+## 结论
+
+- 新鲜度 = 写记忆时记下 `support_refs` 指纹，读时在 MemHub SQLite 里逐条核对 evidence 是否仍匹配；对不上则滤掉该 memory。
+- 这是过度设计：读路径成本高、失败面大（曾导致记忆库整页 503），收益仅是「证据被改过就藏记忆」。
+- 业务层为喂 SDK 再做长证据分段（`segment_*`），与 SDK 自身阶段/窗口处理重复；#86 只是继续兼容这套 proof，不应合入。
+
+## 范围（MemHub Core，非 SDK）
+
+权威库：`~/.one-memory/one-memory.sqlite3`  
+不涉及 SDK `hy-memory/**` 内部 store。
+
+### 读路径（优先拆）
+
+| 位置 | 作用 |
+|---|---|
+| `memhub/apps/core/src/one_memory/memory_evidence_freshness.py` | UDF `is_memory_evidence_current`、proof prefetch、预算跳过 |
+| `source_read_access.py` | 列表/过滤时 per-memory freshness |
+| `context_memory.py` | search/context 候选行 `_memory_row_is_fresh` |
+| `memory_library.py` / `home_summary.py` | 记忆库 / 首页可读谓词 |
+| `memory_visibility.py` | 可见性缓存与 freshness 注册 |
+| `call_event_references.py` | 引用解析时的 freshness |
+
+### 写路径 / 数据（后拆，可分 PR）
+
+| 位置 | 作用 |
+|---|---|
+| `memory_tasks.py` → `MemoryTaskOutput.support_refs` / `TaskEvidenceRef` | 抽取输出里的证明 |
+| `memory_task_outputs` 表 | proof 持久化 |
+| `memory_processing.py` 分段（`segment_evidence_record` / `_semantic_segment_spans`） | 为 HY 窗口切段；与 freshness 耦合但不全是 freshness——分段是否保留需单独拍板 |
+
+### 测试（随拆除改/删）
+
+- `tests/test_memory_evidence_freshness.py`
+- `tests/test_memory_freshness_integration_review.py`
+- `tests/test_read_path_freshness_batch.py`
+- 以及 mock `register_memory_freshness` 的 library / context / visibility / home_summary / call_event 测试
+
+## 建议落地步骤
+
+1. **读路径短路（最小 PR）** — 本 PR  
+   `register_memory_freshness` 恒返回 `"1"`（UDF 恒 `1`），`freshness_proofs_prefetched` 为空操作。行为：证据改过也不再藏 memory；畸形/segment proof 不再 503。  
+   同步改「期望 stale=0」的单测为「始终可读」。
+
+2. **拆调用点**  
+   去掉 `AND is_memory_evidence_current(...)`、`freshness_proofs_prefetched`、`_memory_row_is_fresh` 等死代码，避免假接口长期残留。
+
+3. **停写 / 弃用 proof（可选第二 PR）**  
+   新任务不再依赖 `support_refs` 做读决策；表与历史行可先保留只读，再决定是否 migration 清理。
+
+4. **分段策略另议**  
+   长证据切段是 ingest 窗口问题，不是新鲜度问题。拆除 freshness 后单独决定：继续由 MemHub 切 messages，还是交给 SDK。
+
+5. **对外沟通**  
+   - #86：已留言不合并、将拆除设计。  
+   - 相关 issue（如 #74 里 freshness 503）关闭或改指向「拆除」而非「修 segment」。
+
+## 验收
+
+- 记忆库 / search / context 在存在「证据已修订 + 旧 proof」时仍返回该 memory，且不再因 malformed/segment ref 503。
+- 无 `ONE_MEMORY_FRESHNESS_*` 行为依赖；文档/env 若有提及一并删。
+- Core 相关测试绿。
+
+## 非目标
+
+- 不改 SDK 内部抽取/存储。
+- 不在本轮做「用别的一致性模型替换 freshness」；先删，再按需重设计。