fix(core): drop native store write gate on HY search

MemHub expects concurrent native reads during precipitation. The
process-local read slot mostly existed for a Chroma/GIL fail-fast and
was never held by writers on main; activating it (as in #98) would only
push recall onto weaker fallbacks. Remove the lock module and the
hy_memory_store_busy path instead.

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

.env.example

@@ -86,7 +86,6 @@ 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,25 +122,19 @@ 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
-            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"
+            archive="release-assets/memhub-runtime-v${version}-linux-${arch}.tar.gz"
+            test -f "$archive"
+            test -f "$archive.sha256"
             (
-              cd publish-assets
-              sha256sum -c "$name.sha256"
+              cd release-assets
+              sha256sum -c "$(basename "$archive.sha256")"
             )
           done
       - name: Upload Linux assets to a draft GitHub Release
@@ -166,7 +160,7 @@ jobs:
               --title "$GITHUB_REF_NAME"
           fi
           gh release upload "$GITHUB_REF_NAME" \
-            publish-assets/memhub-runtime-v*-linux-*.tar.gz \
-            publish-assets/memhub-runtime-v*-linux-*.tar.gz.sha256 \
+            release-assets/memhub-runtime-v*-linux-*.tar.gz \
+            release-assets/memhub-runtime-v*-linux-*.tar.gz.sha256 \
             --repo "$GITHUB_REPOSITORY" \
             --clobber

CHANGELOG.md

@@ -2,40 +2,6 @@
 
 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
@@ -48,22 +14,6 @@ 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

@@ -65,13 +65,13 @@ Welcome to the **Hy Memory Research Family**, covering cognitive chat memory, lo
 ### 1. Installation
 
 ```bash
-pip install hy-memory==2.0.4              # core (includes ChromaDB and Kuzu)
-pip install "hy-memory[qdrant]==2.0.4"    # + Qdrant
-pip install "hy-memory[faiss]==2.0.4"     # + FAISS
-pip install "hy-memory[graph]==2.0.4"     # + Neo4j
-pip install "hy-memory[mysql]==2.0.4"     # + MySQL cache
-pip install "hy-memory[rerank]==2.0.4"    # + local multilingual cross-encoder rerank
-pip install "hy-memory[all]==2.0.4"       # everything
+pip install hy-memory==2.0.3              # core (includes ChromaDB and Kuzu)
+pip install "hy-memory[qdrant]==2.0.3"    # + Qdrant
+pip install "hy-memory[faiss]==2.0.3"     # + FAISS
+pip install "hy-memory[graph]==2.0.3"     # + Neo4j
+pip install "hy-memory[mysql]==2.0.3"     # + MySQL cache
+pip install "hy-memory[rerank]==2.0.3"    # + local multilingual cross-encoder rerank
+pip install "hy-memory[all]==2.0.3"       # everything
 ```
 
 ### 2. Configuration
@@ -153,7 +153,7 @@ result = client.search(
 
 result["facts"]          # fact layer
 result["experiences"]    # experience layer
-result["chain_block"]    # coding only: version-evolution context (empty unless MEMORY_CODING_CHAIN_BLOCK_ENABLED)
+result["chain_block"]    # coding only: version-evolution context
 ```
 
 ### 5. Routing, Source Isolation & Advanced APIs

README.zh.md

@@ -65,13 +65,13 @@
 ### 1. 安装
 
 ```bash
-pip install hy-memory==2.0.4              # 核心（含 ChromaDB 与 Kuzu）
-pip install "hy-memory[qdrant]==2.0.4"    # + Qdrant
-pip install "hy-memory[faiss]==2.0.4"     # + FAISS
-pip install "hy-memory[graph]==2.0.4"     # + Neo4j
-pip install "hy-memory[mysql]==2.0.4"     # + MySQL 缓存
-pip install "hy-memory[rerank]==2.0.4"    # + 本地多语言 cross-encoder rerank
-pip install "hy-memory[all]==2.0.4"       # 全部
+pip install hy-memory==2.0.3              # 核心（含 ChromaDB 与 Kuzu）
+pip install "hy-memory[qdrant]==2.0.3"    # + Qdrant
+pip install "hy-memory[faiss]==2.0.3"     # + FAISS
+pip install "hy-memory[graph]==2.0.3"     # + Neo4j
+pip install "hy-memory[mysql]==2.0.3"     # + MySQL 缓存
+pip install "hy-memory[rerank]==2.0.3"    # + 本地多语言 cross-encoder rerank
+pip install "hy-memory[all]==2.0.3"       # 全部
 ```
 
 ### 2. 配置
@@ -153,7 +153,7 @@ result = client.search(
 
 result["facts"]          # 事实层
 result["experiences"]    # 经验层
-result["chain_block"]    # 仅 coding：版本演进上下文（默认空；需 MEMORY_CODING_CHAIN_BLOCK_ENABLED）
+result["chain_block"]    # 仅 coding：版本演进上下文
 ```
 
 ### 5. 自动路由、来源过滤与进阶 API

docs/env_reference.md

@@ -197,7 +197,6 @@ 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,8 +78,7 @@ default 多-query 超过 `MEMORY_QUERY_REWRITE_MAX_CHARS` 时不改写：Coding
 Coding 关闭则直接使用最后一条 query。
 
 路由到 `working` / `coding` 后，Search 读取独立生产力记忆库，返回 `facts` /
-`experiences`，不使用 chat 三通道结构。`coding` 还返回 `chain_block` / `chain_used`
-（默认空；需 `MEMORY_CODING_CHAIN_BLOCK_ENABLED` 才注入 diff 块）。
+`experiences`，不使用 chat 三通道结构。`coding` 还返回 `chain_block` / `chain_used`。
 
 ## Chat 成功返回结构
 
@@ -153,7 +152,7 @@ client = HyMemoryClient.from_config({
 安装可选依赖：
 
 ```bash
-pip install 'hy-memory[rerank]==2.0.4'
+pip install 'hy-memory[rerank]==2.0.1'
 ```
 
 启用后，`HyMemoryClient(...)` 构造阶段会完成模型下载/读取、device 放置和一次 dummy

docs/monorepo.md

@@ -37,7 +37,8 @@ do not belong in the public source tree.
 - Root GitHub workflows run SDK and MemHub checks from their respective working
   directories. Native lifecycle workflows produce reviewable Actions artifacts;
   tag releases remain fail-closed on public identity, signing and notarization.
-- Hy Memory is published as `hy-memory==2.0.4`. MemHub pins that public
-  distribution; its selected npm package is `memhub-cli` 1.0.5.
+- Hy Memory is published as `hy-memory==2.0.3`. MemHub pins that public
+  distribution; its selected npm package is `memhub-cli`, which is not yet
+  published.
 - Hy-Memory source uses Apache-2.0 at the repository root. Third-party
   components retain their own licenses and attribution notices.

docs/scene_coding_usage.md

@@ -168,11 +168,9 @@ working/coding 搜索返回：
 
 `limit` 分别限制 facts 与 experiences，因此两层最多各返回 `limit` 条。
 
-`scene="coding"` 额外返回 `chain_block` 和 `chain_used` 字段（键始终存在；默认内容为空）：
+`scene="coding"` 额外返回 `chain_block` 和 `chain_used`：
 
-- 注入 diff / 版本演进块需显式开启 `MEMORY_CODING_CHAIN_BLOCK_ENABLED=true`
-  （或 `CodingConfig.chain_block_enabled=True`）。默认关闭，以节约 prompt token。
-- 开启后，`chain_block` 是与已召回 fact 对齐的版本演进上下文；无法挂链时为 `""`。
+- `chain_block` 是与已召回 fact 对齐的版本演进上下文；无法挂链时为 `""`。
 - `chain_used` 是已经进入 chain block 的 `facts` 下标。
 - SDK 不会从 `facts` 删除这些条目。调用方若要拼成“chain block + 剩余 facts”，需自行按 `chain_used` 过滤。
 
@@ -224,12 +222,10 @@ 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

@@ -5,14 +5,14 @@
 ## 安装与快速开始
 
 ```bash
-pip install hy-memory==2.0.4
+pip install hy-memory==2.0.3
 
 # 可选后端
-pip install 'hy-memory[qdrant]==2.0.4'
-pip install 'hy-memory[faiss]==2.0.4'
-pip install 'hy-memory[graph]==2.0.4'     # Neo4j；Kuzu 已在核心依赖中
-pip install 'hy-memory[mysql]==2.0.4'
-pip install 'hy-memory[rerank]==2.0.4'    # 本地多语言 cross-encoder rerank
+pip install 'hy-memory[qdrant]==2.0.1'
+pip install 'hy-memory[faiss]==2.0.1'
+pip install 'hy-memory[graph]==2.0.1'     # Neo4j；Kuzu 已在核心依赖中
+pip install 'hy-memory[mysql]==2.0.1'
+pip install 'hy-memory[rerank]==2.0.1'    # 本地多语言 cross-encoder rerank
 ```
 
 默认 LLM 和 Embedding 均使用 OpenAI-compatible 接口。使用 OpenAI 时可只设置：
@@ -210,8 +210,7 @@ result = client.search(
 ```
 
 `working` / `coding` 路径返回 `scene="productivity"`、`facts`、`experiences`，不使用
-chat 的三桶结构；`coding` 还会返回 `chain_block` 和 `chain_used` 字段（默认空；
-需 `MEMORY_CODING_CHAIN_BLOCK_ENABLED=true` 才注入 diff 块）。
+chat 的三桶结构；`coding` 还会返回 `chain_block` 和 `chain_used`。
 
 #### `search_mode`
 

hy_memory/__init__.py

@@ -55,13 +55,6 @@ 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
@@ -118,11 +111,6 @@ __all__ = [
     "HyMemoryClient",
     "MemoryInspector",
     "SharedRuntime",
-    "DataPlaneRestore",
-    "DataPlaneSnapshot",
-    "DataPlaneSnapshotRejected",
-    "restore_local_data_plane",
-    "snapshot_local_data_plane",
     # 配置
     "MemoryConfig",
     # Embedding 端点覆盖

hy_memory/agent/abstractor.py

@@ -22,8 +22,7 @@ logger = logging.getLogger(__name__)
 
 # ================================================================
 # 数据结构
-# ================================================================
-@dataclass
+# ================================================================@dataclass
 class SessionSummaryResult:
     """Session 摘要结果 (V2)"""
     success: bool

hy_memory/agent/episode_summary.py

@@ -39,7 +39,6 @@ import logging
 import hashlib
 import json
 import re
-from datetime import datetime
 from typing import Any, Dict, List, Optional
 import uuid
 
@@ -253,17 +252,6 @@ def _series_episode_id(
     return str(uuid.uuid5(uuid.NAMESPACE_URL, "hy-memory:series-episode:" + raw))
 
 
-def _series_memory_at(times: List[int]) -> Optional[datetime]:
-    """Earliest member timestamp as the same local wall clock it came from.
-
-    ``_node_timestamp`` produces ``datetime.timestamp()`` values. Reading them
-    back with ``utcfromtimestamp`` turns them into a naive *UTC* datetime, which
-    the vector payload later re-interprets as local time and shifts by the
-    machine's UTC offset.
-    """
-    return datetime.fromtimestamp(min(times)) if times else None
-
-
 def _node_timestamp(node: Any) -> Optional[int]:
     value = getattr(node, "memory_at", None) or getattr(node, "gmt_created", None)
     if value is None:
@@ -986,6 +974,8 @@ async def finalize_series_episode_containers(
     Stored text is only a title plus a structural intro. Exact L2 bodies remain in
     their canonical nodes and are referenced by ID from dated timeline segments.
     """
+    from datetime import datetime
+
     from ..models.memory import MemoryLayer, MemoryNode, MemoryStatus, SourceType
     from ..pipelines._retrieval.session_episode import (
         add_parent_episode_membership,
@@ -1071,7 +1061,7 @@ async def finalize_series_episode_containers(
             segment["memory_at"] for segment in plan["timeline"]
             if segment.get("memory_at") is not None
         ]
-        memory_at = _series_memory_at(times)
+        memory_at = datetime.utcfromtimestamp(min(times)) if times else None
         content = f"{plan['title']}. {plan['intro']}".strip()
         series = MemoryNode(
             node_id=series_id, user_id=user_id,

hy_memory/coding/chain_store.py

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

hy_memory/config.py

@@ -1136,9 +1136,6 @@ 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:
@@ -1156,10 +1153,6 @@ 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,37 +114,6 @@ 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,21 +81,6 @@ _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,
@@ -474,128 +459,29 @@ 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（分钟粒度落盘）
     # ================================================================
 
     async def store_metrics_minute(self, minute_ts: str, data: dict) -> None:
-        import aiomysql
-
         try:
             now = datetime.now().isoformat()
-            # Read-merge-write must be one transaction: two instances flushing
-            # the same minute used to overwrite each other's counters, and a
-            # concurrent insert could lose the whole batch to a duplicate key.
-            for attempt in range(2):
-                try:
-                    async with self._pool.acquire() as conn:
-                        await conn.begin()
-                        try:
-                            async with conn.cursor(aiomysql.DictCursor) as cur:
-                                await cur.execute(
-                                    "SELECT data FROM system_metrics "
-                                    "WHERE minute_ts = %s FOR UPDATE",
-                                    (minute_ts,),
-                                )
-                                row = await cur.fetchone()
-                                if row is not None:
-                                    old_raw = (
-                                        row["data"]
-                                        if isinstance(row, dict)
-                                        else row[0]
-                                    )
-                                    merged = self._merge_metric_buckets(
-                                        json.loads(old_raw), data,
-                                    )
-                                    await cur.execute(
-                                        "UPDATE system_metrics SET data = %s, "
-                                        "created_at = %s WHERE minute_ts = %s",
-                                        (
-                                            json.dumps(merged, default=str),
-                                            now,
-                                            minute_ts,
-                                        ),
-                                    )
-                                else:
-                                    await cur.execute(
-                                        "INSERT INTO system_metrics "
-                                        "(minute_ts, data, created_at) "
-                                        "VALUES (%s, %s, %s)",
-                                        (
-                                            minute_ts,
-                                            json.dumps(data, default=str),
-                                            now,
-                                        ),
-                                    )
-                            await conn.commit()
-                            return
-                        except Exception:
-                            await conn.rollback()
-                            raise
-                except (aiomysql.IntegrityError, aiomysql.OperationalError) as error:
-                    code = error.args[0] if error.args else None
-                    deadlock = code in (1205, 1213)
-                    if attempt or (isinstance(error, aiomysql.OperationalError) and not deadlock):
-                        raise
-                    # A competing writer inserted the row, or InnoDB chose us
-                    # as the deadlock victim. Retry so this batch merges.
-                    continue
+            # 尝试 UPSERT
+            existing = await self._fetchone(
+                "SELECT data FROM system_metrics WHERE minute_ts = %s", (minute_ts,)
+            )
+            if existing:
+                old_data = json.loads(existing["data"])
+                merged = self._merge_metric_buckets(old_data, data)
+                await self._execute(
+                    "UPDATE system_metrics SET data = %s, created_at = %s WHERE minute_ts = %s",
+                    (json.dumps(merged, default=str), now, minute_ts),
+                )
+            else:
+                await self._execute(
+                    "INSERT INTO system_metrics (minute_ts, data, created_at) VALUES (%s, %s, %s)",
+                    (minute_ts, json.dumps(data, default=str), now),
+                )
         except Exception as e:
             logger.warning(f"store_metrics_minute failed: {e}")
 

hy_memory/data/cache_sqlite.py

@@ -87,20 +87,6 @@ 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,
@@ -428,88 +414,6 @@ 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/data/history_store.py

@@ -66,11 +66,6 @@ def _now_iso() -> str:
     return datetime.now(timezone.utc).isoformat()
 
 
-def _escape_like(value: str) -> str:
-    """Escape LIKE wildcards so a user id is matched literally."""
-    return value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
-
-
 class HistoryStore:
     """
     SQLite 历史存储 - 记录所有内存操作的审计日志。
@@ -266,34 +261,6 @@ class HistoryStore:
             )
             return [dict(row) for row in cursor.fetchall()]
 
-    async def get_recent_for_user(
-        self,
-        user_id: str,
-        limit: int = 50,
-    ) -> List[Dict[str, Any]]:
-        """获取某 user 下所有 agent 的最近操作。
-
-        isolation_key 形如 ``"{user_id}:{agent_id}"``。user_id 里的 LIKE
-        通配符会被转义，保证只匹配这个 user，不会漏到 ``u1`` 之外。
-        """
-        prefix = _escape_like(f"{user_id}:") + "%"
-        return await _run_in_sqlite_pool(
-            self._get_recent_prefix_sync, prefix, limit,
-        )
-
-    def _get_recent_prefix_sync(
-        self, pattern: str, limit: int,
-    ) -> List[Dict[str, Any]]:
-        with self._conn_lock:
-            if self._conn is None:
-                return []
-            cursor = self._conn.execute(
-                "SELECT * FROM memory_history WHERE isolation_key LIKE ? ESCAPE '\\' "
-                "ORDER BY created_at DESC LIMIT ?",
-                (pattern, limit),
-            )
-            return [dict(row) for row in cursor.fetchall()]
-
     async def count(self, isolation_key: str = "") -> int:
         """
         统计历史记录数量。

hy_memory/data/vector_store_faiss.py

@@ -462,39 +462,14 @@ class FaissVectorStore(VectorStoreBase):
             return None
 
     async def get_embeddings(self, node_ids: List[str]) -> Dict[str, List[float]]:
-        """批量取向量，供去重 / episode 摘要计算 cosine。
-
-        ``_data`` 只保存 payload 与 FAISS 内部下标；普通 upsert 不额外存一份
-        embedding，向量本体在 FAISS index 里。这里按当前有效下标 reconstruct，
-        并用 ``_idx_to_id`` 校验，避免读到已被替换的旧下标。
-        """
-
-        def _get() -> Dict[str, List[float]]:
-            out: Dict[str, List[float]] = {}
-            with self._lock:
-                for nid in node_ids:
-                    point_id = self._node_id_to_point_id(nid)
-                    entry = self._data.get(point_id)
-                    if entry is None:
-                        continue
-                    embedding = entry.get("embedding")
-                    if embedding:
-                        out[nid] = list(embedding)
-                        continue
-                    idx = entry.get("idx")
-                    if (
-                        idx is None
-                        or self._index is None
-                        or self._idx_to_id.get(int(idx)) != point_id
-                    ):
-                        continue
-                    try:
-                        out[nid] = self._index.reconstruct(int(idx)).tolist()
-                    except Exception:
-                        continue
-            return out
-
-        return await _run_in_vdb_pool(_get)
+        """批量取向量（in-memory _data 直接读 embedding），供去重算 cosine。"""
+        out: Dict[str, List[float]] = {}
+        for nid in node_ids:
+            point_id = self._node_id_to_point_id(nid)
+            entry = self._data.get(point_id)
+            if entry and entry.get("embedding"):
+                out[nid] = list(entry["embedding"])
+        return out
 
     # ================================================================
     # 删除

hy_memory/pipelines/base.py

@@ -231,10 +231,6 @@ 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,27 +280,6 @@ 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）。
@@ -989,7 +968,7 @@ class MemoryWriter(WritePipeline):
             async def _run():
                 try:
                     nid = await summarize_closed_episode(
-                        vector_store=await self._get_vector_store(),
+                        vector_store=self.vector_store,
                         embed_service=self.embed_service,
                         llm_provider=self.llm_provider,
                         episode_id=episode_id,
@@ -2617,23 +2596,13 @@ class MemoryWriter(WritePipeline):
             from ..metrics import MetricsCollector
             MetricsCollector.get().sys1_start()
 
-            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）
+            # 2. 向量化 + 持久化（L1_RAW）
             _t_embed = datetime.now()
-            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)})
+            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(
@@ -2656,43 +2625,27 @@ class MemoryWriter(WritePipeline):
             with tracer.span("qdrant_upsert") as s:
                 try:
                     vector_store = await self._get_vector_store()
-                    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)
+                    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] {'reused' if existing_raw_ids else 'persisted'}: "
-                        f"id={memory_id} layer={suggested_layer}"
-                    )
+                    logger.debug(f"[write] persisted: id={memory_id} layer={suggested_layer}")
                 except Exception as persist_err:
                     _l1_error = str(persist_err)
                     s.set_error(_l1_error)
@@ -2756,12 +2709,6 @@ 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,
@@ -2771,13 +2718,12 @@ class MemoryWriter(WritePipeline):
                             "error": response.extra.get("agent_error", ""),
                         },
                     )
-                    if not existing_raw_ids:
-                        await self._rollback_failed_l1(
-                            request=request,
-                            response=response,
-                            vector_store=vector_store,
-                            memory_id=memory_id,
-                        )
+                    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/server.py

@@ -29,7 +29,7 @@ import logging
 import os
 import re
 import signal
-import threading
+import sys
 from datetime import datetime
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from typing import Any, Dict, Optional
@@ -495,16 +495,6 @@ class MemoryHTTPHandler(BaseHTTPRequestHandler):
         _json_response(self, 200, result)
 
 
-def _request_shutdown(server: ThreadingHTTPServer) -> None:
-    """Ask ``serve_forever`` to stop without blocking the calling thread.
-
-    ``BaseServer.shutdown()`` waits for the ``serve_forever`` loop to exit. A
-    signal handler runs on the very thread executing that loop, so calling it
-    there deadlocks forever. Hand the call to a helper thread instead.
-    """
-    threading.Thread(target=server.shutdown, daemon=True).start()
-
-
 def run_server(port: int = 19527, host: str = "127.0.0.1"):
     """Start the HTTP server (blocking)."""
     # Configure logging — basicConfig for root; hy_memory level set after client init
@@ -533,12 +523,13 @@ def run_server(port: int = 19527, host: str = "127.0.0.1"):
     logger.info(f"[server] Listening on http://{host}:{port}")
     logger.info(f"[server] Health check: http://{host}:{port}/healthz")
 
-    # Graceful shutdown. The signal handler only asks a helper thread to stop
-    # the loop; cleanup runs after serve_forever() returns so it cannot race
-    # the request threads or deadlock the serving thread.
+    # Graceful shutdown
     def _shutdown(signum, frame):
         logger.info("[server] Shutting down...")
-        _request_shutdown(server)
+        server.shutdown()
+        if _client is not None:
+            _client.close()
+        sys.exit(0)
 
     signal.signal(signal.SIGINT, _shutdown)
     signal.signal(signal.SIGTERM, _shutdown)
@@ -547,10 +538,6 @@ def run_server(port: int = 19527, host: str = "127.0.0.1"):
         server.serve_forever()
     except KeyboardInterrupt:
         _shutdown(None, None)
-    finally:
-        server.server_close()
-        if _client is not None:
-            _client.close()
 
 
 def main():

hy_memory/snapshot.py

@@ -1,150 +0,0 @@
-"""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/README.md

@@ -18,13 +18,13 @@ If you run into a problem, please open an Issue with the version, platform,repro
 
 ## 发行与平台合同
 
-公共 Python SDK 为 `hy-memory==2.0.4`。MemHub 源码版本为 `1.0.4`
-（build 13），GitHub 仓库为
+公共 Python SDK 为 `hy-memory==2.0.3`。MemHub 源码版本为 `1.0.3`
+（build 7），GitHub 仓库为
 [`Tencent-Hunyuan/Hy-Memory`](https://github.com/Tencent-Hunyuan/Hy-Memory)，npm 包名为
 `memhub-cli`。当前代码架构见
 [`docs/ARCHITECTURE.md`](docs/ARCHITECTURE.md)。
 
-`v1.0.4` 的发行目标是 Linux x64/arm64 Runtime 与 macOS arm64/x64 App 和 Runtime。
+`v1.0.3` 的发行目标是 Linux x64/arm64 Runtime 与 macOS arm64/x64 App 和 Runtime。
 Linux 制品由 GitHub Actions 在原生架构 runner 上构建并完成生命周期验收；macOS
 制品由发行流水线完成 Developer ID 签名、公证和验收后手动上传。Windows 不支持。
 只有同时出现在 GitHub Release 与 `release-manifest.json` 中、且通过公开校验的制品才属于
@@ -113,7 +113,7 @@ MemHub 的 SQLite 数据默认位于 `~/.one-memory`，不在 npm 包或 Runtime
 
 ## Hy Memory
 
-当前开发依赖固定为公开 PyPI 包 `hy-memory==2.0.4`。SDK 配置、权限边界及 search
+当前开发依赖固定为公开 PyPI 包 `hy-memory==2.0.3`。SDK 配置、权限边界及 search
 返回格式见 [`MEMORY.md`](MEMORY.md)。HY search 返回的 `content` 已包含 SDK 统一生成的
 逻辑时间和演化上下文；调用方不应再次转换时间戳或重复拼接 `When`。
 

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

@@ -76,7 +76,8 @@ 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.
+    # Same local revocation/ownership predicate as SourceReadAccess.memories,
+    # evaluated BEFORE freshness (which can read support text), in this snapshot.
     # Its public methods open their own connections, so cannot be used here.
     access = db.execute(
         "SELECT EXISTS (SELECT 1 FROM governed_memory_evidence "
@@ -120,8 +121,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. Evidence freshness
-    proofs are retired and do not affect availability.
+    A memory needs at least one accessible active citation. No proof preserves
+    shared freshness compatibility, not a claim of verified historical freshness.
 
     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/imports/adapters.py

@@ -34,11 +34,10 @@ class ScanIssueCode(StrEnum):
     SOURCE_BUSY = "source_busy"
     UNSUPPORTED_SCHEMA = "unsupported_schema"
     BUDGET_EXCEEDED = "budget_exceeded"
-    #: The file is bigger than a per-file ceiling: either the import's own
-    #: ceiling or the access gate's per-root `max_file_bytes`. Both are skipped
-    #: with this reason rather than read, because one oversized history file
-    #: can hold the import open long enough to take the process with it, and
-    #: the user-facing cause is the same size limit.
+    #: The file is bigger than the import's own per-file ceiling. Skipped with
+    #: a reason rather than read, because one oversized history file can hold
+    #: the import open long enough to take the process with it. Distinct from
+    #: the access layer's `FILE_TOO_LARGE`, which is a per-root limit.
     FILE_TOO_LARGE = "file_too_large"
     HISTORY_WINDOW_EXCLUDED = "history_window_excluded"
 

memhub/apps/core/src/one_memory/imports/sources/common.py

@@ -1120,11 +1120,6 @@ def discovery_failure_code(
 ) -> ScanIssueCode:
     if failure.reason_code is SourceAccessErrorCode.BUDGET_EXCEEDED:
         return ScanIssueCode.BUDGET_EXCEEDED
-    if failure.reason_code is SourceAccessErrorCode.FILE_TOO_LARGE:
-        # The file is readable and its format is fine; only its size is over
-        # the access gate. Reporting that as `unsupported_schema` sends the
-        # user looking for a parser problem that does not exist.
-        return ScanIssueCode.FILE_TOO_LARGE
     return ScanIssueCode.UNSUPPORTED_SCHEMA
 
 
@@ -1132,8 +1127,6 @@ def _scan_failure_code(code: SourceAccessErrorCode) -> ScanIssueCode:
     """Map a per-file open failure onto a machine code, never onto free text."""
     if code is SourceAccessErrorCode.BUDGET_EXCEEDED:
         return ScanIssueCode.BUDGET_EXCEEDED
-    if code is SourceAccessErrorCode.FILE_TOO_LARGE:
-        return ScanIssueCode.FILE_TOO_LARGE
     if code is SourceAccessErrorCode.SYMLINK_NOT_ALLOWED:
         return ScanIssueCode.UNSUPPORTED_SCHEMA
     if code in {

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

@@ -1,25 +1,36 @@
-"""Memory evidence freshness checks are retired.
+"""Read-only, exact-revision support checks inside the caller's SQLite snapshot.
 
-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.
+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.
 """
 
-from __future__ import annotations
-
+import json
+import logging
+import os
 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
 
-#: Historical budget error codes. Retained so diagnostics/env docs do not break
-#: mid-teardown; nothing raises them on the read path anymore.
+_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.
 BUDGET_ERRORS = frozenset(
     {
         "memory_freshness_proof_budget",
@@ -28,23 +39,80 @@ 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:
-    """Former opt-in to raise on budget overrun. Always false after retirement."""
-    return False
+    """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"
 
 
-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 _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 _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,
@@ -52,16 +120,328 @@ def freshness_proofs_prefetched(
     *,
     evidence_ids: Sequence[str] = (),
 ) -> Iterator[None]:
-    """No-op. Kept so batched read callers need not change in this PR."""
-    del db, memory_ids, evidence_ids
-    yield
+    """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")
 
 
 def register_memory_freshness(db: sqlite3.Connection) -> str:
-    """Register a always-true UDF; return SQL predicate ``1``.
+    """Register UDF; return a predicate for alias m. Does not change row_factory.
 
-    Callers that embed the return value in ``WHERE`` keep working. Callers that
-    invoke ``is_memory_evidence_current`` by name also get current.
+    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).
     """
-    register_function(db, "is_memory_evidence_current", 3, lambda *_: 1)
-    return "1"
+    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)"

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` | 新鲜度拆除后：读路径恒为 current；证据修订 / 畸形 proof / segment ref 不再藏记忆或 503。 |
+| `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_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,9 +355,8 @@ 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 result[1].available
+    assert_hidden(result[1], reference, "changed")
     corrected = as_person(MemoryGovernanceStore(path)).correct(
         memory.memory_id,
         user_id="local",
@@ -370,7 +369,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_keeps_memory_readable_when_evidence_changes(fresh):
+def test_scope_only_revision_still_checks_original_support(fresh):
     path, evidence, value, record, memory, tasks, output = fresh
     updated = as_person(MemoryGovernanceStore(path)).set_scopes(
         memory.memory_id,
@@ -382,27 +381,25 @@ def test_scope_only_revision_keeps_memory_readable_when_evidence_changes(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 resolve_call_references(path, event(reference))[0].available
+    assert_hidden(resolve_call_references(path, event(reference))[0], reference, "changed")
 
 
-def test_evidence_identity_change_without_revision_keeps_memory_readable(fresh):
+def test_freshness_identity_change_without_revision_change(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 resolve_call_references(path, event(reference))[0].available
+    assert_hidden(resolve_call_references(path, event(reference))[0], reference, "changed")
 
 
-def test_invalid_proof_is_ignored_but_revocation_still_hides_memory(fresh):
+def test_invalid_freshness_proof_fails_whole_response_but_revoked_never_reads_it(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'")
-    assert all(
-        item.available
-        for item in resolve_call_references(path, event(ref(key=record.evidence_id), reference))
-    )
+    with pytest.raises(CallReferencesUnavailable, match="^call_references_unavailable$"):
+        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))
@@ -458,13 +455,9 @@ 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))
-    # 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_hidden(
+        current[0], reference, "changed" if change in ("evidence", "memory") else "unavailable"
+    )
     assert current[1].available == (change in ("memory", "link"))
 
 

memhub/apps/core/tests/test_context_raw_scale_benchmark.py

@@ -8,11 +8,6 @@ import pytest
 pytestmark = [pytest.mark.performance, pytest.mark.serial]
 
 SCRIPT = Path(__file__).resolve().parents[3] / "scripts" / "bench_context_raw_scale.py"
-if not SCRIPT.exists():
-    pytest.skip(
-        f"benchmark script is not part of this checkout: {SCRIPT.name}",
-        allow_module_level=True,
-    )
 spec = importlib.util.spec_from_file_location("bench_context_raw_scale", SCRIPT)
 benchmark = importlib.util.module_from_spec(spec)
 sys.modules["bench_context_raw_scale"] = benchmark

memhub/apps/core/tests/test_memory_evidence_freshness.py

@@ -1,5 +1,3 @@
-"""Freshness retirement: read path always admits, even when evidence changed."""
-
 import json
 import sqlite3
 
@@ -7,11 +5,7 @@ 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 (
-    budget_errors_raise,
-    freshness_proofs_prefetched,
-    register_memory_freshness,
-)
+from one_memory.memory_evidence_freshness import 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
@@ -67,45 +61,46 @@ def check(path, memory):
         ).fetchone()[0]
 
 
-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):
+def test_fresh_then_revised_and_identity_only_change(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_segment_support_refs_do_not_fail_reads(case):
+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):
     path, evidence, value, record, memory, tasks, output = case
     body = output.model_dump(mode="json")
-    body["support_refs"][0].update(
-        segment_index=0,
-        segment_start=0,
-        segment_end=len(record.verbatim),
-        segment_hash="0" * 64,
-    )
+    if broken == "identity":
+        body["memory_id"] = "other"
+    if broken == "ref":
+        body["support_refs"][0]["revision"] = "1"
+    payload = json.dumps(body)
+    if broken == "json":
+        payload = "{"
     with sqlite3.connect(path) as db:
-        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
+        db.execute("UPDATE memory_task_outputs SET body=?", (payload,))
+    with pytest.raises(sqlite3.OperationalError):
+        check(path, memory)
 
 
 def test_missing_table_compatibility_is_readonly():
@@ -116,19 +111,239 @@ def test_missing_table_compatibility_is_readonly():
         assert db.execute("SELECT name FROM sqlite_master").fetchall() == []
 
 
-def test_prefetch_is_noop(case):
+def test_table_presence_is_checked_once_at_registration(case):
     path, evidence, value, record, memory, tasks, output = case
     with sqlite3.connect(path) as db:
         db.execute("BEGIN")
-        with freshness_proofs_prefetched(db, [memory.memory_id], evidence_ids=[record.evidence_id]):
-            assert register_memory_freshness(db) == "1"
-            assert check(path, memory) == 1
+        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
 
 
-def test_budget_raise_env_is_inert(case, monkeypatch):
+def test_proof_budget_reads_current_instead_of_failing(case):
     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}"}))
     assert check(path, memory) == 1
+
+
+def test_proof_budget_still_raises_when_opted_back(case, monkeypatch):
+    path, evidence, value, record, memory, tasks, output = case
+    monkeypatch.setenv("ONE_MEMORY_FRESHNESS_BUDGET_RAISES", "1")
+    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,11 +102,17 @@ def revise(case):
     ).record
 
 
-def test_revised_evidence_remains_readable_after_freshness_retirement(case):
-    """Evidence revision no longer hides library / context / source reads."""
+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).
     assert visible(case) == (1, 1, 1, 1)
     revise(case)
-    assert visible(case) == (1, 1, 1, 1)
+    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 case.governance.get(case.memory.memory_id, user_id="local") == case.memory
     assert case.tasks.outputs(case.output.task_id) == (case.output,)
 
@@ -114,10 +120,11 @@ def test_revised_evidence_remains_readable_after_freshness_retirement(case):
 @pytest.mark.parametrize(
     "revise_first", [False, True], ids=["scope-before-revision", "scope-after-revision"]
 )
-def test_scope_only_revision_keeps_memory_readable_after_evidence_change(case, revise_first):
+def test_scope_only_revision_must_not_resurrect_stale_derived_content(case, revise_first):
     if revise_first:
         revise(case)
-        assert visible(case) == (1, 1, 1, 1)
+        _, home, context_items, source_items = visible(case)
+        assert (context_items, source_items) == (0, 0)
     changed = as_person(case.governance).set_scopes(
         case.memory.memory_id,
         user_id="local",
@@ -129,7 +136,8 @@ def test_scope_only_revision_keeps_memory_readable_after_evidence_change(case, r
     if not revise_first:
         assert visible(case, changed) == (1, 1, 1, 1)
         revise(case)
-    assert visible(case, changed) == (1, 1, 1, 1)
+    _, home, context_items, source_items = visible(case, changed)
+    assert (context_items, source_items) == (0, 0)
 
 
 def test_explicit_correct_remains_readable_but_never_overrides_revocation(case):
@@ -144,10 +152,7 @@ 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
-            (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)""",
+            "INSERT INTO source_grants VALUES ('local','cursor','/fixture','default',?,?)",
             ("2026-09-20T00:00:00Z", "2026-09-20T00:00:01Z"),
         )
     assert visible(case, corrected) == (0, 0, 0, 0)
@@ -176,10 +181,7 @@ 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
-                (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)""",
+                "INSERT INTO source_grants VALUES ('local','cursor','/fixture','default',?,?)",
                 ("2026-09-20T00:00:00Z", "2026-09-20T00:00:01Z"),
             )
         changes.append(True)
@@ -192,7 +194,7 @@ def test_revocation_before_freshness_snapshot_cannot_release_memory(case, monkey
     assert returned == []
 
 
-def test_udf_stays_current_across_evidence_revision(case):
+def test_udf_uses_caller_snapshot_then_observes_revision_in_next_transaction(case):
     with closing(sqlite3.connect(case.path)) as reader:
         reader.execute("PRAGMA journal_mode=WAL")
         reader.execute("BEGIN")
@@ -204,7 +206,7 @@ def test_udf_stays_current_across_evidence_revision(case):
         reader.commit()
         reader.execute("BEGIN")
         register_memory_freshness(reader)
-        assert reader.execute(sql).fetchone() == (1,)
+        assert reader.execute(sql).fetchone() == (0,)
 
 
 def _oversize(case, bad):
@@ -225,16 +227,20 @@ def test_material_too_large_to_verify_reads_as_current(case, bad):
 
 
 @pytest.mark.parametrize("bad", ["duplicate-key"])
-def test_bad_proof_material_no_longer_makes_reads_unavailable(case, bad):
+def test_bad_material_is_unavailable_not_successful_empty(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"
-    assert len(result.items) == 1
+    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])
 
 
 def test_route_serves_memories_whose_proof_is_too_large(case, tmp_path):

memhub/apps/core/tests/test_memory_library.py

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

memhub/apps/core/tests/test_processing_requests.py

@@ -8,11 +8,8 @@ from uuid import UUID
 import pytest
 from pydantic import ValidationError
 
-requests_module = pytest.importorskip(
-    "one_memory.processing_requests",
-    reason="processing-request store is not part of the public Core snapshot",
-)
-from one_memory.processing_requests import (  # noqa: E402 - importorskip must run first
+import one_memory.processing_requests as requests_module
+from one_memory.processing_requests import (
     ProcessingRequest,
     ProcessingRequestConflictError,
     ProcessingRequestState,

memhub/apps/core/tests/test_raw_vector_scale_benchmark.py

@@ -7,11 +7,6 @@ import pytest
 pytestmark = [pytest.mark.performance, pytest.mark.serial]
 
 SCRIPT = Path(__file__).resolve().parents[3] / "scripts" / "bench_raw_vector_scale.py"
-if not SCRIPT.exists():
-    pytest.skip(
-        f"benchmark script is not part of this checkout: {SCRIPT.name}",
-        allow_module_level=True,
-    )
 spec = importlib.util.spec_from_file_location("bench_raw_vector_scale", SCRIPT)
 benchmark = importlib.util.module_from_spec(spec)
 sys.modules["bench_raw_vector_scale"] = benchmark

memhub/apps/core/tests/test_retrieval_eligibility_benchmark.py

@@ -8,11 +8,6 @@ import pytest
 pytestmark = [pytest.mark.performance, pytest.mark.serial]
 
 SCRIPT = Path(__file__).resolve().parents[3] / "scripts" / "bench_retrieval_eligibility_scale.py"
-if not SCRIPT.exists():
-    pytest.skip(
-        f"benchmark script is not part of this checkout: {SCRIPT.name}",
-        allow_module_level=True,
-    )
 spec = importlib.util.spec_from_file_location("bench_retrieval_eligibility_scale", SCRIPT)
 benchmark = importlib.util.module_from_spec(spec)
 sys.modules["bench_retrieval_eligibility_scale"] = benchmark

memhub/apps/core/tests/test_source_access.py

@@ -224,11 +224,6 @@ def test_live_adapter_keeps_symlink_refusal_in_existing_issue_taxonomy() -> None
     )
 
 
-def test_live_adapter_reports_an_oversized_file_as_file_too_large() -> None:
-    """The open path and the discovery path must agree on the size reason."""
-    assert _scan_failure_code(SourceAccessErrorCode.FILE_TOO_LARGE) is ScanIssueCode.FILE_TOO_LARGE
-
-
 def test_live_adapter_runner_persists_symlink_refusal_as_unsupported_schema(
     tmp_path, monkeypatch
 ) -> None:

memhub/apps/core/tests/test_source_discovery.py

@@ -3,12 +3,10 @@ from one_memory.imports.access import (
     AuthorizedSourceRoot,
     SourceAccessErrorCode,
 )
-from one_memory.imports.adapters import ScanIssueCode
 from one_memory.imports.discovery import (
     SourceFileDiscoveryFailure,
     discover_source_files,
 )
-from one_memory.imports.sources.common import discovery_failure_code
 
 
 def grant(root, home):
@@ -62,35 +60,3 @@ def test_discovery_does_not_follow_directory_or_file_symlinks(tmp_path) -> None:
     assert len(events) == 1
     assert isinstance(events[0], SourceFileDiscoveryFailure)
     assert "private" not in events[0].file_key
-
-
-def test_discovery_reports_an_oversized_file_as_file_too_large(tmp_path) -> None:
-    """A size-gate refusal is a size problem, not a schema problem.
-
-    The Codex report in #102 shows the access gate refusing a file over the
-    per-root limit while the UI said `unsupported_schema`, which sends the user
-    looking for a parser that is working fine.
-    """
-    root = tmp_path / "source"
-    root.mkdir()
-    (root / "big.jsonl").write_text('{"content":"too big"}\n')
-    granted = AuthorizedSourceRoot.grant(
-        source_id="fixture",
-        root=root,
-        user_confirmed=True,
-        home=tmp_path / "home",
-        max_file_bytes=8,
-    )
-
-    events = list(
-        discover_source_files(
-            granted,
-            suffixes=frozenset({".jsonl"}),
-            max_targets=10,
-        )
-    )
-
-    assert len(events) == 1
-    assert isinstance(events[0], SourceFileDiscoveryFailure)
-    assert events[0].reason_code is SourceAccessErrorCode.FILE_TOO_LARGE
-    assert discovery_failure_code(events[0]) is ScanIssueCode.FILE_TOO_LARGE

memhub/docs/ARCHITECTURE.md

@@ -363,7 +363,7 @@ Support is declared per immutable GitHub Release manifest:
 - macOS arm64 and x64 App/Runtime artifacts must carry Developer ID signatures,
   notarization tickets, and native-platform acceptance receipts.
 - Windows is not supported.
-- `hy-memory==2.0.4` is the public Python SDK dependency.
+- `hy-memory==2.0.3` is the public Python SDK dependency.
 
 See [`INSTALLATION.md`](INSTALLATION.md) for the release matrix. A source test,
 fake hosting test, or cross-build does not substitute for a native-platform

memhub/docs/INSTALLATION.md

@@ -1,6 +1,6 @@
 # MemHub 安装与发行
 
-MemHub `v1.0.4`（build 13）的发行目标如下：
+MemHub `v1.0.3`（build 7）的发行目标如下：
 
 | 平台 | 发行物 | 构建与验收 |
 | --- | --- | --- |

memhub/packages/runtime-control/src/core-supervisor-reader-start.test.ts

@@ -1,66 +0,0 @@
-import type { ChildProcess, SpawnOptions } from "node:child_process";
-import { EventEmitter } from "node:events";
-import { describe, expect, it, vi } from "vitest";
-
-import { CoreSupervisor } from "./core-supervisor.js";
-
-function fakeChild(): ChildProcess {
-  const child = new EventEmitter() as ChildProcess;
-  child.stderr = new EventEmitter() as ChildProcess["stderr"];
-  child.stdout = new EventEmitter() as ChildProcess["stdout"];
-  child.kill = vi.fn(() => {
-    setImmediate(() => child.emit("exit", 0, null));
-    return true;
-  });
-  return child;
-}
-
-function fakeReaderChild(): ChildProcess {
-  return fakeChild();
-}
-
-describe("CoreSupervisor reader startup", () => {
-  it("starts the reader after spawning its own Core", async () => {
-    const core = fakeChild();
-    const reader = fakeReaderChild();
-    const spawned: string[][] = [];
-    const spawnProcess = vi.fn(
-      (_command: string, args: readonly string[], _options: SpawnOptions) => {
-        spawned.push([...args]);
-        if (args.includes("--reader")) {
-          // The Reader attaches its stderr listener synchronously after this
-          // call returns, so publish the port on the next tick.
-          setImmediate(() =>
-            reader.stderr?.emit("data", Buffer.from("reader_port:43210\n")),
-          );
-          return reader;
-        }
-        return core;
-      },
-    );
-    const supervisor = new CoreSupervisor({
-      launch: { command: "one-memory", args: [] },
-      dataDir: "/tmp/one-memory-reader-start-test",
-      instanceId: "owned",
-      port: 9600,
-      attempts: 1,
-      delayMs: 0,
-      readerEnabled: true,
-      spawnProcess,
-      probeHealth: async () => ({
-        status: "ok",
-        service: "one-memory-core",
-        version: "0.0.0",
-        instance_id: "owned",
-      }),
-    });
-
-    await supervisor.start();
-
-    expect(spawnProcess).toHaveBeenCalledTimes(2);
-    expect(spawned[1]).toContain("--reader");
-    expect(supervisor.readerPort).toBe(43210);
-
-    await supervisor.stop();
-  });
-});

memhub/packages/runtime-control/src/core-supervisor.ts

@@ -339,11 +339,7 @@ export class CoreSupervisor {
     }
     this.useOwnedIdentity();
     try {
-      const health = await this.attempt();
-      if (this.options.readerEnabled) {
-        await this.startReader();
-      }
-      return health;
+      return await this.attempt();
     } catch (error) {
       // A standalone Runtime may have won the race after our initial
       // discovery read but before our child acquired the data lock. Verify it
@@ -351,12 +347,7 @@ export class CoreSupervisor {
       let raced: HealthResponse | null = null;
       try {
         raced = await this.attachExisting();
-        if (raced !== null) {
-          if (this.options.readerEnabled) {
-            await this.startReader();
-          }
-          return raced;
-        }
+        if (raced !== null) return raced;
       } catch (attachError) {
         // A stale receipt can fail identity validation after the owned Core
         // exited. Treat that as a cleanup candidate, not as a new startup
@@ -396,11 +387,7 @@ export class CoreSupervisor {
       this.exitSignal = null;
       this.spawnError = null;
       try {
-        const health = await this.attempt();
-        if (this.options.readerEnabled) {
-          await this.startReader();
-        }
-        return health;
+        return await this.attempt();
       } catch (retryError) {
         await this.logStartFailure(retryError);
         throw retryError;
@@ -709,9 +696,6 @@ export class CoreSupervisor {
           await delay(delayMs);
           if (this.stopping) return;
           try {
-            // The old Reader pointed at the Core that just died. Stop it so
-            // the relaunched Core gets a Reader with the current token/port.
-            await this.stopReader();
             // Keep the configured port stable across relaunches. The packaged
             // renderer and discovery receipts point at this port, so moving to
             // an ephemeral port would make a recovered Core unreachable.
@@ -722,9 +706,6 @@ export class CoreSupervisor {
             this.exited = false;
             this.spawnError = null;
             const health = await this.attempt();
-            if (this.options.readerEnabled) {
-              await this.startReader();
-            }
             await this.options.diagnostics?.log({
               level: "info",
               component: "runtime.core",
@@ -843,14 +824,6 @@ export class CoreSupervisor {
       cwd: this.options.launch.cwd,
     });
     this.readerChild = child;
-    // A Reader exits when its Core goes away. Clear the handle so the restart
-    // path can spawn a fresh one instead of returning early forever.
-    child.on("exit", () => {
-      if (this.readerChild === child) {
-        this.readerChild = null;
-        this.activeReaderPort = null;
-      }
-    });
     // Wait briefly for the reader to publish its port on stderr.
     this.activeReaderPort = await new Promise<number | null>((resolve) => {
       let settled = false;

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\.4`/);
-  assert.match(readme, /`v1\.0\.4` 的发行目标是 Linux x64\/arm64 Runtime/);
+  assert.match(readme, /公共 Python SDK 为 `hy-memory==2\.0\.3`/);
+  assert.match(readme, /`v1\.0\.3` 的发行目标是 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,19 +117,9 @@ 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.5.dev4"
+version = "2.0.4"
 description = "HY Memory - Industrial-grade dual-system cognitive memory framework"
 readme = "README.md"
 requires-python = ">=3.9"

tests/coding/test_chain_block_config.py

@@ -1,19 +0,0 @@
-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,21 +62,7 @@ def _seed_chain(tmpdir, branch="feat/q4", path=_PATH):
     return H
 
 
-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):
+def _bind_chain(ctx, store):
     """把链层挂到 search_ctx 这个 SimpleNamespace 上。
 
     search_ctx 是裸 namespace，不是真 client，所以 _decorate_with_chain 这个
@@ -87,8 +73,6 @@ def _bind_chain(ctx, store, *, enable_block=True):
     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
 
 
@@ -145,17 +129,6 @@ 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,34 +326,12 @@ 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_disabled_by_default(self):
+    def test_returns_empty_when_chain_unavailable(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,
         )
@@ -361,12 +339,13 @@ class TestDecorateGlue:
 
     def test_store_exception_degrades_not_raises(self):
         # 这层是增量, 挂了要退回纯向量双层而不是把整次 search 拖失败。
+        import types
 
         class _Boom:
             def decorate(self, *a, **k):
                 raise RuntimeError("boom")
 
-        ns = _ns_with_chain_block(_get_chain_store=lambda owner="": _Boom())
+        ns = types.SimpleNamespace(_get_chain_store=lambda owner="": _Boom())
         assert HyMemoryClient._decorate_with_chain(ns, [dict(_FACT)]) == ("", [])
 
 

tests/test_abstractor_dataclass.py

@@ -1,13 +0,0 @@
-# -*- coding: utf-8 -*-
-"""Public Abstractor result types must remain constructible dataclasses."""
-
-from hy_memory.agent.abstractor import SessionSummaryResult
-
-
-def test_session_summary_result_accepts_fields():
-    result = SessionSummaryResult(success=True, summary="session summary")
-
-    assert result.success is True
-    assert result.summary == "session summary"
-    assert result.key_topics == []
-    assert result.error is None

tests/test_cache_backend_parity.py

@@ -24,9 +24,6 @@ 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 = [
@@ -73,16 +70,6 @@ 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_client_loop_lifecycle.py

@@ -1,45 +0,0 @@
-# -*- coding: utf-8 -*-
-"""Owned event-loop threads must be stopped and closed with the client."""
-
-from hy_memory.client import HyMemoryClient, _LoopThread
-
-
-def test_loop_thread_stop_closes_the_loop_and_is_idempotent():
-    loop_thread = _LoopThread()
-    loop = loop_thread._loop
-
-    loop_thread.stop()
-    assert not loop_thread._thread.is_alive()
-    assert loop.is_closed()
-
-    # A second stop must not touch a closed loop.
-    loop_thread.stop()
-
-
-def test_close_stops_both_loops_owned_by_a_solo_client():
-    client = HyMemoryClient.__new__(HyMemoryClient)
-    client._registry = None
-    client._owns_runtime = True
-    client._loop_thread = _LoopThread()
-    client._search_loop_thread = _LoopThread()
-    loops = [client._loop_thread._loop, client._search_loop_thread._loop]
-
-    client.close()
-
-    assert all(loop.is_closed() for loop in loops)
-
-
-def test_shared_runtime_loops_are_left_to_the_runtime():
-    client = HyMemoryClient.__new__(HyMemoryClient)
-    client._registry = None
-    client._owns_runtime = False
-    client._loop_thread = _LoopThread()
-    client._search_loop_thread = _LoopThread()
-    loops = [client._loop_thread._loop, client._search_loop_thread._loop]
-
-    try:
-        client.close()
-        assert all(not loop.is_closed() for loop in loops)
-    finally:
-        client._loop_thread.stop()
-        client._search_loop_thread.stop()

tests/test_data_plane_snapshot.py

@@ -1,125 +0,0 @@
-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

@@ -1,120 +0,0 @@
-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_episode.py

@@ -173,41 +173,6 @@ class TestEpisodeCloseTriggersSummary:
         await w._resolve_episode_id(_req(t0 + timedelta(hours=3)), 3)
         assert scheduled == [ep1], "只在关闭时对刚关闭的那个 episode 触发一次"
 
-    @pytest.mark.asyncio
-    async def test_summary_uses_the_writer_vector_store(self, monkeypatch):
-        """后台摘要必须拿到 writer 已初始化的 vector store。
-
-        旧实现访问了不存在的 ``self.vector_store``，AttributeError 被
-        except 吞掉，L3 episode 摘要从不生成。
-        """
-        from types import SimpleNamespace
-
-        import hy_memory.agent.episode_summary as episode_summary
-
-        captured = {}
-
-        async def fake_summarize(**kwargs):
-            captured.update(kwargs)
-            return "episode-node"
-
-        monkeypatch.setattr(
-            episode_summary, "summarize_closed_episode", fake_summarize,
-        )
-
-        w = _writer(_FakeCache())
-        w.config.recall.episode_summary_enabled = True
-        w._mem_agent = SimpleNamespace(llm_provider=object())
-        w._embed_service = object()
-        vector_store = object()
-        w._external_vector_store = vector_store
-        w._vector_store = vector_store
-        w._vector_store_initialized = True
-
-        w._schedule_episode_summary(_req(datetime(2026, 9, 13)), "ep-1")
-        await asyncio.sleep(0.05)
-
-        assert captured["vector_store"] is vector_store
-
 
 # ======================================================================
 # Episode 摘要（写侧，省 token）
@@ -710,20 +675,3 @@ class TestInstrumentation:
         assert logged[1]["count"] == 5, "累计条数要落下来"
         assert logged[2]["reused"] is False
         assert all(r["gap"] == 1800.0 for r in logged)
-
-
-def test_series_memory_at_round_trips_local_wall_clock(monkeypatch):
-    """Series timestamps must not shift by the machine's UTC offset."""
-    import time as _time
-
-    from hy_memory.agent.episode_summary import _series_memory_at
-
-    monkeypatch.setenv("TZ", "Asia/Shanghai")
-    _time.tzset()
-    try:
-        expected = datetime(2026, 1, 1, 12, 0)
-        assert _series_memory_at([int(expected.timestamp())]) == expected
-        assert _series_memory_at([]) is None
-    finally:
-        monkeypatch.delenv("TZ", raising=False)
-        _time.tzset()

tests/test_faiss_get_embeddings.py

@@ -1,57 +0,0 @@
-# -*- coding: utf-8 -*-
-"""FAISS 后端的 get_embeddings 必须能取回普通 upsert 写入的向量。
-
-``_data`` 只保存 payload 与内部下标，向量本体在 FAISS index 里。去重和
-episode 摘要依赖 get_embeddings，取不到向量会让这些能力静默失效。
-"""
-
-import asyncio
-
-import pytest
-
-pytest.importorskip("faiss")
-
-from hy_memory.config import MemoryConfig  # noqa: E402
-from hy_memory.data.vector_store_faiss import FaissVectorStore  # noqa: E402
-from hy_memory.models.memory import MemoryLayer, MemoryNode  # noqa: E402
-
-
-def _config(tmp_path):
-    return MemoryConfig.from_dict({
-        "vector_store": {
-            "provider": "faiss",
-            "collection_name": "get_embeddings_test",
-            "persist_directory": str(tmp_path / "faiss"),
-            "embedding_dims": 4,
-        },
-    })
-
-
-def test_get_embeddings_reconstructs_upserted_and_updated_vectors(tmp_path):
-    async def _run():
-        store = FaissVectorStore(_config(tmp_path))
-        await store.initialize()
-        try:
-            node = MemoryNode(
-                user_id="u1",
-                agent_id="a1",
-                session_id="s1",
-                source="default",
-                layer=MemoryLayer.L1_RAW,
-                content="hello",
-                embedding=[1.0, 0.0, 0.0, 0.0],
-            )
-            await store.upsert(node)
-
-            got = await store.get_embeddings([node.node_id])
-            assert got[node.node_id] == pytest.approx([1.0, 0.0, 0.0, 0.0])
-
-            assert await store.update_embedding(node.node_id, [0.0, 1.0, 0.0, 0.0])
-            got = await store.get_embeddings([node.node_id])
-            assert got[node.node_id] == pytest.approx([0.0, 1.0, 0.0, 0.0])
-
-            assert await store.get_embeddings(["missing"]) == {}
-        finally:
-            await store.close()
-
-    asyncio.run(_run())

tests/test_history_recent_for_user.py

@@ -1,60 +0,0 @@
-# -*- coding: utf-8 -*-
-"""History queries without agent_ids must cover every agent of the user."""
-
-import asyncio
-
-from hy_memory.config import MemoryConfig
-from hy_memory.data.history_store import HistoryStore
-
-
-def _store(tmp_path):
-    config = MemoryConfig()
-    config.history.db_path = str(tmp_path / "history.db")
-    return HistoryStore(config)
-
-
-def test_get_recent_for_user_matches_all_agents(tmp_path):
-    async def _run():
-        store = _store(tmp_path)
-        await store.initialize()
-        try:
-            keys = [
-                "u1:default_agent",
-                "u1:agent_b",
-                "u10:default_agent",
-                "u2:default_agent",
-            ]
-            for index, key in enumerate(keys):
-                await store.record_add(
-                    memory_id=f"m{index}",
-                    content="x",
-                    layer="l1_raw",
-                    isolation_key=key,
-                )
-            rows = await store.get_recent_for_user("u1", limit=10)
-        finally:
-            await store.close()
-        return [row["isolation_key"] for row in rows]
-
-    assert set(asyncio.run(_run())) == {"u1:default_agent", "u1:agent_b"}
-
-
-def test_get_recent_for_user_escapes_like_wildcards(tmp_path):
-    async def _run():
-        store = _store(tmp_path)
-        await store.initialize()
-        try:
-            await store.record_add(
-                memory_id="m1", content="x", layer="l1_raw",
-                isolation_key="u_1:default_agent",
-            )
-            await store.record_add(
-                memory_id="m2", content="x", layer="l1_raw",
-                isolation_key="uX1:default_agent",
-            )
-            rows = await store.get_recent_for_user("u_1", limit=10)
-        finally:
-            await store.close()
-        return [row["isolation_key"] for row in rows]
-
-    assert asyncio.run(_run()) == ["u_1:default_agent"]

tests/test_list_layer_filter_client.py

@@ -31,12 +31,7 @@ def run(coro):
 DIMS = 32
 
 
-def _node(
-    user_id="u1",
-    content="x",
-    layer=MemoryLayer.L2_FACT,
-    status=MemoryStatus.ACTIVE,
-):
+def _node(user_id="u1", content="x", layer=MemoryLayer.L2_FACT):
     import random
     n = MemoryNode(
         node_id=str(uuid.uuid4()),
@@ -45,8 +40,8 @@ def _node(
         session_id="s1",
         content=content,
         layer=layer,
-        status=status,
-        is_latest=status == MemoryStatus.ACTIVE,
+        status=MemoryStatus.ACTIVE,
+        is_latest=True,
     )
     vec = [random.gauss(0, 1) for _ in range(DIMS)]
     norm = sum(v * v for v in vec) ** 0.5
@@ -109,22 +104,6 @@ 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([
@@ -184,24 +163,6 @@ 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([
@@ -254,37 +215,3 @@ 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_mysql_metrics_atomic.py

@@ -1,111 +0,0 @@
-# -*- coding: utf-8 -*-
-"""MySQL minute metrics must merge under one row lock."""
-
-import asyncio
-import json
-
-import pytest
-
-pytest.importorskip("aiomysql")
-
-from hy_memory.config import MemoryConfig  # noqa: E402
-from hy_memory.data.cache_mysql import MysqlCache  # noqa: E402
-
-
-class FakeCursor:
-    def __init__(self, row):
-        self.row = row
-        self.executed = []
-
-    async def __aenter__(self):
-        return self
-
-    async def __aexit__(self, *args):
-        return None
-
-    async def execute(self, sql, args=None):
-        self.executed.append((sql, args))
-
-    async def fetchone(self):
-        return self.row
-
-
-class FakeConnection:
-    def __init__(self, row):
-        self.cursor_obj = FakeCursor(row)
-        self.begun = False
-        self.committed = False
-        self.rolled_back = False
-
-    async def begin(self):
-        self.begun = True
-
-    async def commit(self):
-        self.committed = True
-
-    async def rollback(self):
-        self.rolled_back = True
-
-    def cursor(self, *args, **kwargs):
-        return self.cursor_obj
-
-
-class _Acquire:
-    def __init__(self, connection):
-        self.connection = connection
-
-    async def __aenter__(self):
-        return self.connection
-
-    async def __aexit__(self, *args):
-        return None
-
-
-class FakePool:
-    def __init__(self, connection):
-        self.connection = connection
-
-    def acquire(self):
-        return _Acquire(self.connection)
-
-
-def _cache_with(row):
-    cache = MysqlCache(MemoryConfig())
-    connection = FakeConnection(row)
-    cache._pool = FakePool(connection)
-    return cache, connection
-
-
-def test_metrics_insert_takes_the_row_lock_and_commits():
-    cache, connection = _cache_with(None)
-
-    asyncio.run(
-        cache.store_metrics_minute("2026-01-01T00:00", {"sys1_started": 2})
-    )
-
-    statements = [sql for sql, _ in connection.cursor_obj.executed]
-    assert any("FOR UPDATE" in sql for sql in statements)
-    assert any(sql.startswith("INSERT") for sql in statements)
-    assert connection.begun and connection.committed
-    assert not connection.rolled_back
-
-
-def test_metrics_update_merges_existing_buckets():
-    old = {"sys1_started": 1, "sys1_timing_sums": {"a": 1}}
-    cache, connection = _cache_with({"data": json.dumps(old)})
-
-    asyncio.run(
-        cache.store_metrics_minute(
-            "2026-01-01T00:00",
-            {"sys1_started": 2, "sys1_timing_sums": {"a": 3}},
-        )
-    )
-
-    update = next(
-        args for sql, args in connection.cursor_obj.executed
-        if sql.startswith("UPDATE")
-    )
-    merged = json.loads(update[0])
-    assert merged["sys1_started"] == 3
-    assert merged["sys1_timing_sums"]["a"] == 4
-    assert connection.committed

tests/test_raw_public_api.py

@@ -1,405 +0,0 @@
-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_search_layer_filter.py

@@ -77,28 +77,6 @@ class TestReadRequestWiring:
             HyMemoryClient._normalize_layers(["l3_identity"])
 
 
-class TestDedupKindPairing:
-    """dedup 的 _memory_kind 必须与它的 public item 来自同一条 reader 结果。"""
-
-    def test_memory_kind_follows_source_after_layer_filter(self):
-        memories = [
-            {"memory_id": "raw", "layer": "l1_raw", "content": "raw"},
-            {
-                "memory_id": "canon",
-                "layer": "l2_fact",
-                "content": "canon",
-                "_memory_kind": "canonical_state",
-            },
-        ]
-        all_mems, dedup_mems = HyMemoryClient._public_search_views(
-            memories,
-            wanted_layers={"l2_fact"},
-            lite_mode=False,
-        )
-        assert [m["memory_id"] for m in all_mems] == ["canon"]
-        assert dedup_mems[0]["_memory_kind"] == "canonical_state"
-
-
 class TestHybridDirectedRecall:
     """共享 reader 主流程必须把 request.layers 下推到召回阶段。"""
 

tests/test_server_shutdown.py

@@ -1,76 +0,0 @@
-# -*- coding: utf-8 -*-
-"""HTTP server signal shutdown must not deadlock the serving thread."""
-
-import subprocess
-import sys
-import textwrap
-import threading
-import time
-from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
-
-from hy_memory.server import _request_shutdown
-
-
-class _Handler(BaseHTTPRequestHandler):
-    def do_GET(self):  # pragma: no cover - the test only needs a live server
-        self.send_response(200)
-        self.end_headers()
-
-    def log_message(self, *args):  # pragma: no cover - silence test output
-        pass
-
-
-def test_request_shutdown_returns_promptly():
-    server = ThreadingHTTPServer(("127.0.0.1", 0), _Handler)
-    server.daemon_threads = True
-    thread = threading.Thread(target=server.serve_forever)
-    thread.start()
-    try:
-        started = time.monotonic()
-        _request_shutdown(server)
-        assert time.monotonic() - started < 1.0
-        thread.join(timeout=2.0)
-        assert not thread.is_alive()
-    finally:
-        server.server_close()
-
-
-def test_signal_handler_does_not_deadlock_the_serving_thread():
-    """A signal handler calling shutdown() directly hangs forever.
-
-    Run the serving loop on the main thread (as run_server does), fire an
-    alarm, and require the process to exit on its own.
-    """
-    script = textwrap.dedent(
-        """
-        import signal
-        from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
-        from hy_memory.server import _request_shutdown
-
-        class H(BaseHTTPRequestHandler):
-            def do_GET(self):
-                self.send_response(200)
-                self.end_headers()
-            def log_message(self, *args):
-                pass
-
-        server = ThreadingHTTPServer(("127.0.0.1", 0), H)
-
-        def _handler(signum, frame):
-            _request_shutdown(server)
-
-        signal.signal(signal.SIGALRM, _handler)
-        signal.alarm(1)
-        server.serve_forever()
-        server.server_close()
-        print("exited")
-        """
-    )
-    result = subprocess.run(
-        [sys.executable, "-c", script],
-        capture_output=True,
-        text=True,
-        timeout=10,
-    )
-    assert result.returncode == 0, result.stderr
-    assert "exited" in result.stdout

tests/test_write_outcome_cache.py

@@ -1,47 +0,0 @@
-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

@@ -1,128 +0,0 @@
-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.5.dev4"
+version = "2.0.4"
 source = { editable = "." }
 dependencies = [
     { name = "chromadb" },

新鲜度_todo.md

@@ -1,70 +0,0 @@
-# 新鲜度（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」；先删，再按需重设计。