fix(memhub): make trigger upgrade atomic

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

CHANGELOG.md

@@ -14,6 +14,17 @@ Hy Memory and MemHub are versioned and released independently.
 
 ## Unreleased
 
+- 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
+  or governed and visibility data.
+- Serialized each SDK SQLite store's shared connection lifecycle so concurrent
+  history/cache operations cannot race with `close()` and crash the process.
+- Clarified that `add(metadata=...)` carries legacy SDK processing options,
+  while persistent caller business context belongs in `meta`.
+- Added an opt-in `get(..., include_raw=True)` path for reading `L1_RAW`
+  memories, including memories written in lite mode. The default remains
+  hidden for backward compatibility.
 - Removed MemHub's task-scoped HyPermission store wrappers and global
   model-content consent switch. The trusted embedded Hy Memory SDK now uses
   its native local stores directly.

docs/usage.md

@@ -106,6 +106,15 @@ result = client.add(
 )
 ```
 
+`metadata` 与 `meta` 不是同一字段：
+
+- `metadata` 是兼容旧调用方保留的 SDK 处理选项，会进入内部
+  `WriteRequest.extra`，不保证持久化或随查询结果返回。
+- `meta` 是调用方的业务上下文，会持久化并在 `search()`、`get()` 等结果中返回。
+  同一记忆被多次合并时，各次 `meta` 会按写入顺序保存在列表中。
+
+需要以后读回的数据必须放在 `meta`，不要放在 `metadata`。
+
 `scene` 的公开取值：
 
 - `None`：普通文本直接走 chat；messages 含 tool 时执行快速场景判断。
@@ -263,6 +272,8 @@ catalog 也可以在没有基础 L0 时形成只读的合成画像。
 
 ```python
 memory = client.get("memory_id")
+# lite 模式写入的是 L1_RAW；按 ID 读取时必须显式选择返回原文。
+raw_memory = client.get("memory_id", include_raw=True)
 result = client.update("memory_id", "更新后的内容")
 result = client.delete("memory_id")
 result = client.delete_all(user_id="user_001")
@@ -282,6 +293,10 @@ result = client.clone_user("user_001", "user_002", agent_id=None)
 对应异步方法为 `async_get()`、`async_update()`、`async_delete()`、
 `async_delete_all()`、`async_list_memories()`、`async_clone_user()`。
 
+`get()` / `async_get()` 默认继续隐藏 `L1_RAW`，因此不会改变现有调用方的原文访问
+边界。只有显式传入 `include_raw=True` 才返回该层；这也是 lite 模式按 ID 读取
+写入内容及其 `meta` 的方式。MemHub 的 HTTP 读取接口不开放此参数。
+
 普通 `list_memories()` 返回 `vdb` 桶，并在 Graph 有结果时附带 `graph`；传入
 `workspace_id` 时返回生产力记忆的 `scene="productivity"` 和 `experiences`。
 

hy_memory/client.py

@@ -1597,7 +1597,9 @@ class HyMemoryClient:
             session_id:   Session ID（三级隔离 key）
             source:       调用来源，例如 openclaw / cursor / codex；与 agent_id 正交。
                           写入前执行 strip().lower()，默认 "default"。
-            metadata:     自定义元数据
+            metadata:     旧版 SDK 处理选项，会并入内部 WriteRequest.extra。
+                          不保证持久化或随查询返回；调用方需要保存并读回的业务字段
+                          请使用 meta。
             memory_at:    记忆时间戳（不传则用当前时间），用于导入历史记忆
             scene:        显式指定写入路由，跳过 LLM 自动判定。
                           - None（默认）：含 tool 时 LLM 判 chat/working/coding；
@@ -1623,8 +1625,8 @@ class HyMemoryClient:
             branch:       Coding memory 路径专用。分支名。同 workspace_id。chat 路径忽略。
             meta:         调用方自定义业务 meta，随本次产出的记忆存储、查询时原样返回。
                           不 embed、不参与召回、不作过滤条件。
-                          与 metadata 不同：metadata 会并入 extra（含 SDK 内部控制键），
-                          meta 独立存放，只做透传。
+                          与 metadata 不同：metadata 是旧版内部处理选项，
+                          meta 独立持久化，只做业务透传。
                           语义为「该次 write 时刻的上下文」，故存为 list——同一节点被
                           多次写入合并时会追加而非覆盖，详见 get() 返回的 meta 字段。
 
@@ -1791,17 +1793,26 @@ class HyMemoryClient:
             query_embedding=query_embedding,
         ))
 
-    def get(self, memory_id: str) -> Optional[Dict[str, Any]]:
+    def get(
+        self,
+        memory_id: str,
+        *,
+        include_raw: bool = False,
+    ) -> Optional[Dict[str, Any]]:
         """
         获取单条记忆。
 
         Args:
             memory_id: 记忆 ID
+            include_raw: 是否允许返回 L1_RAW 原始记忆。默认 False，保持原有隐藏行为；
+                         lite 模式的记忆均为 L1_RAW，需显式设为 True 才能按 ID 读取。
 
         Returns:
             {"memory_id", "content", "layer", "request_id", "elapsed_ms"} 或 None（不存在时）
         """
-        return self._loop_thread.run(self.async_get(memory_id))
+        return self._loop_thread.run(
+            self.async_get(memory_id, include_raw=include_raw)
+        )
 
     def update(
         self,
@@ -3650,12 +3661,21 @@ class HyMemoryClient:
     # ================================================================
 
     @_ensure_internal_loop
-    async def async_get(self, memory_id: str) -> Optional[Dict[str, Any]]:
+    async def async_get(
+        self,
+        memory_id: str,
+        *,
+        include_raw: bool = False,
+    ) -> Optional[Dict[str, Any]]:
         """
         获取单条记忆（异步）。
 
         VDB + Graph 双路并行查询，哪个有结果返回哪个。
 
+        Args:
+            memory_id: 记忆 ID
+            include_raw: 是否允许返回 L1_RAW 原始记忆，默认 False。
+
         Returns:
             {"memory_id", "content", "layer", "source", "request_id", "elapsed_ms"} 或 None
         """
@@ -3683,17 +3703,24 @@ class HyMemoryClient:
 
         vdb_node, graph_node = await _asyncio_get.gather(_get_vdb(), _get_graph())
 
-        # L1_RAW 不对外可见：按 ID 直取走主键查询、绕过向量库的 layer filter，
-        # 故在此统一挡掉。async_get 是 GET /v1/memories/{id} 与
-        # /v1/memories/batch 两个对外接口的共同必经处，而内部写链路不经过它
-        # （writer 写后校验直接调 vector_store.get_by_id），挡在此处不影响内部。
+        # L1_RAW 默认不对外可见：按 ID 直取走主键查询、绕过向量库的 layer
+        # filter，故在此统一挡掉。SDK 调用方可通过 include_raw=True 显式读取；
+        # HTTP 接口不传该参数，继续保持隐藏。内部写链路不经过此方法。
         #
         # 置于 gather 之后、埋点之前：命中 raw 仍照常落 GET 埋点（保留可观测性，
         # 便于发现有调用方拿 raw id 试探），只是不把内容返回给调用方。
         # 对调用方的表现与"该 id 不存在"完全一致，不额外暴露"存在但被挡"的信息。
-        if vdb_node is not None and vdb_node.layer == MemoryLayer.L1_RAW:
+        if (
+            not include_raw
+            and vdb_node is not None
+            and vdb_node.layer == MemoryLayer.L1_RAW
+        ):
             vdb_node = None
-        if graph_node is not None and getattr(graph_node, "layer", None) == MemoryLayer.L1_RAW:
+        if (
+            not include_raw
+            and graph_node is not None
+            and getattr(graph_node, "layer", None) == MemoryLayer.L1_RAW
+        ):
             graph_node = None
 
         elapsed = round((time.perf_counter() - t0) * 1000, 2)

hy_memory/data/cache_sqlite.py

@@ -7,7 +7,7 @@ Agent Memory V2 - SqliteCache
 设计模式跟随 history_store.py:
 - 标准库 sqlite3 + _run_in_sqlite_pool()
 - WAL 模式 + check_same_thread=False
-- threading.Lock 保护写操作
+- 每实例 threading.RLock 保护共享连接的全部操作与关闭
 
 三张表:
 - memory_operations / pipeline_logs / system_metrics: 审计与观测
@@ -122,7 +122,10 @@ class SqliteCache(CacheBase):
     def __init__(self, config: MemoryConfig):
         self.config = config
         self._conn: Optional[sqlite3.Connection] = None
-        self._lock = threading.Lock()
+        # One connection is shared by the executor workers. Use the same lock
+        # for reads, writes, initialization and close so close() cannot free
+        # the connection while another worker is inside sqlite3.
+        self._lock = threading.RLock()
 
         # 读取 db_path
         cache_cfg = getattr(config, "cache", None)
@@ -144,35 +147,37 @@ class SqliteCache(CacheBase):
         logger.debug(f"SqliteCache initialized: {self._db_path}")
 
     def _init_sync(self) -> None:
-        db_dir = os.path.dirname(self._db_path)
-        if db_dir:
-            os.makedirs(db_dir, exist_ok=True)
-        self._conn = sqlite3.connect(self._db_path, check_same_thread=False)
-        self._conn.row_factory = sqlite3.Row
-        self._conn.execute("PRAGMA journal_mode=WAL;")
-        self._conn.executescript(_CREATE_TABLES_SQL)
-        self._migrate_pipeline_logs_columns()
-        self._conn.commit()
+        with self._lock:
+            db_dir = os.path.dirname(self._db_path)
+            if db_dir:
+                os.makedirs(db_dir, exist_ok=True)
+            self._conn = sqlite3.connect(self._db_path, check_same_thread=False)
+            self._conn.row_factory = sqlite3.Row
+            self._conn.execute("PRAGMA journal_mode=WAL;")
+            self._conn.executescript(_CREATE_TABLES_SQL)
+            self._migrate_pipeline_logs_columns()
+            self._conn.commit()
 
     def _migrate_pipeline_logs_columns(self) -> None:
         """给旧库的 pipeline_logs_v2 补 namespace/session/event time 列。"""
-        try:
-            cols = {
-                row["name"]
-                for row in self._conn.execute("PRAGMA table_info(pipeline_logs_v2)")
-            }
-            _text_cols = ("namespace", "session_id")
-            for col in _text_cols:
-                if col not in cols:
+        with self._lock:
+            try:
+                cols = {
+                    row["name"]
+                    for row in self._conn.execute("PRAGMA table_info(pipeline_logs_v2)")
+                }
+                _text_cols = ("namespace", "session_id")
+                for col in _text_cols:
+                    if col not in cols:
+                        self._conn.execute(
+                            f"ALTER TABLE pipeline_logs_v2 ADD COLUMN {col} TEXT NOT NULL DEFAULT ''"
+                        )
+                if "event_at_ms" not in cols:
                     self._conn.execute(
-                        f"ALTER TABLE pipeline_logs_v2 ADD COLUMN {col} TEXT NOT NULL DEFAULT ''"
+                        "ALTER TABLE pipeline_logs_v2 ADD COLUMN event_at_ms INTEGER NOT NULL DEFAULT 0"
                     )
-            if "event_at_ms" not in cols:
-                self._conn.execute(
-                    "ALTER TABLE pipeline_logs_v2 ADD COLUMN event_at_ms INTEGER NOT NULL DEFAULT 0"
-                )
-        except Exception as e:
-            logger.debug(f"[sqlite] pipeline_logs_v2 column migration skipped: {e}")
+            except Exception as e:
+                logger.debug(f"[sqlite] pipeline_logs_v2 column migration skipped: {e}")
 
     async def close(self) -> None:
         if self._conn is not None:
@@ -180,9 +185,10 @@ class SqliteCache(CacheBase):
         logger.info("SqliteCache closed")
 
     def _close_sync(self) -> None:
-        if self._conn:
-            self._conn.close()
-            self._conn = None
+        with self._lock:
+            if self._conn:
+                self._conn.close()
+                self._conn = None
 
     # ================================================================
     # 统计
@@ -255,39 +261,40 @@ class SqliteCache(CacheBase):
         """
         try:
             def _query():
-                conditions = []
-                params = []
-                if request_id:
-                    conditions.append("request_id = ?")
-                    params.append(request_id)
-                if memory_id:
-                    conditions.append("memory_id = ?")
-                    params.append(memory_id)
-                if user_id:
-                    conditions.append("user_id = ?")
-                    params.append(user_id)
+                with self._lock:
+                    conditions = []
+                    params = []
+                    if request_id:
+                        conditions.append("request_id = ?")
+                        params.append(request_id)
+                    if memory_id:
+                        conditions.append("memory_id = ?")
+                        params.append(memory_id)
+                    if user_id:
+                        conditions.append("user_id = ?")
+                        params.append(user_id)
 
-                if not conditions:
-                    return []
+                    if not conditions:
+                        return []
 
-                where = " AND ".join(conditions)
-                rows = self._conn.execute(
-                    f"SELECT * FROM memory_operations_v2 WHERE {where} "
-                    f"ORDER BY id DESC LIMIT ?",
-                    params + [limit],
-                ).fetchall()
-                result = []
-                for r in rows:
-                    row = dict(r)
-                    # 反序列化 supersedes JSON 字符串 → list
-                    import json as _json
-                    raw_sup = row.get("supersedes", "")
-                    try:
-                        row["supersedes"] = _json.loads(raw_sup) if raw_sup else []
-                    except Exception:
-                        row["supersedes"] = []
-                    result.append(row)
-                return result
+                    where = " AND ".join(conditions)
+                    rows = self._conn.execute(
+                        f"SELECT * FROM memory_operations_v2 WHERE {where} "
+                        f"ORDER BY id DESC LIMIT ?",
+                        params + [limit],
+                    ).fetchall()
+                    result = []
+                    for r in rows:
+                        row = dict(r)
+                        # 反序列化 supersedes JSON 字符串 → list
+                        import json as _json
+                        raw_sup = row.get("supersedes", "")
+                        try:
+                            row["supersedes"] = _json.loads(raw_sup) if raw_sup else []
+                        except Exception:
+                            row["supersedes"] = []
+                        result.append(row)
+                    return result
 
             return await _run_in_sqlite_pool(_query)
         except Exception as e:
@@ -369,37 +376,38 @@ class SqliteCache(CacheBase):
         """
         try:
             def _query():
-                conditions = []
-                params = []
-                if request_id:
-                    conditions.append("request_id = ?")
-                    params.append(request_id)
-                if user_id:
-                    conditions.append("user_id = ?")
-                    params.append(user_id)
-                if step:
-                    conditions.append("step = ?")
-                    params.append(step)
+                with self._lock:
+                    conditions = []
+                    params = []
+                    if request_id:
+                        conditions.append("request_id = ?")
+                        params.append(request_id)
+                    if user_id:
+                        conditions.append("user_id = ?")
+                        params.append(user_id)
+                    if step:
+                        conditions.append("step = ?")
+                        params.append(step)
 
-                if not conditions:
-                    return []
+                    if not conditions:
+                        return []
 
-                where = " AND ".join(conditions)
-                rows = self._conn.execute(
-                    f"SELECT * FROM pipeline_logs_v2 WHERE {where} "
-                    f"ORDER BY id ASC LIMIT ?",
-                    params + [limit],
-                ).fetchall()
-                out = [dict(r) for r in rows]
-                # 派生可读的 event_at（ISO ms）供查询接口直接用；event_at_ms 为 0 时留空
-                from datetime import datetime as _dt
-                for _r in out:
-                    _ms = _r.get("event_at_ms") or 0
-                    _r["event_at"] = (
-                        _dt.fromtimestamp(_ms / 1000.0).isoformat(timespec="milliseconds")
-                        if _ms else ""
-                    )
-                return out
+                    where = " AND ".join(conditions)
+                    rows = self._conn.execute(
+                        f"SELECT * FROM pipeline_logs_v2 WHERE {where} "
+                        f"ORDER BY id ASC LIMIT ?",
+                        params + [limit],
+                    ).fetchall()
+                    out = [dict(r) for r in rows]
+                    # 派生可读的 event_at（ISO ms）供查询接口直接用；event_at_ms 为 0 时留空
+                    from datetime import datetime as _dt
+                    for _r in out:
+                        _ms = _r.get("event_at_ms") or 0
+                        _r["event_at"] = (
+                            _dt.fromtimestamp(_ms / 1000.0).isoformat(timespec="milliseconds")
+                            if _ms else ""
+                        )
+                    return out
 
             return await _run_in_sqlite_pool(_query)
         except Exception as e:
@@ -414,26 +422,27 @@ class SqliteCache(CacheBase):
         """存储一个分钟桶的增量指标数据（UPSERT: 同一分钟多次写入会合并）"""
         try:
             def _store():
-                now = datetime.now().isoformat()
-                # UPSERT: 如果同一分钟已存在，合并数据
-                existing = self._conn.execute(
-                    "SELECT data FROM system_metrics WHERE minute_ts = ?",
-                    (minute_ts,),
-                ).fetchone()
-                if existing:
-                    import json as _json
-                    old_data = _json.loads(existing["data"])
-                    merged = self._merge_metric_buckets(old_data, data)
-                    self._conn.execute(
-                        "UPDATE system_metrics SET data = ?, created_at = ? WHERE minute_ts = ?",
-                        (_json.dumps(merged, default=str), now, minute_ts),
-                    )
-                else:
-                    self._conn.execute(
-                        "INSERT INTO system_metrics (minute_ts, data, created_at) VALUES (?, ?, ?)",
-                        (minute_ts, json.dumps(data, default=str), now),
-                    )
-                self._conn.commit()
+                with self._lock:
+                    now = datetime.now().isoformat()
+                    # UPSERT: 如果同一分钟已存在，合并数据
+                    existing = self._conn.execute(
+                        "SELECT data FROM system_metrics WHERE minute_ts = ?",
+                        (minute_ts,),
+                    ).fetchone()
+                    if existing:
+                        import json as _json
+                        old_data = _json.loads(existing["data"])
+                        merged = self._merge_metric_buckets(old_data, data)
+                        self._conn.execute(
+                            "UPDATE system_metrics SET data = ?, created_at = ? WHERE minute_ts = ?",
+                            (_json.dumps(merged, default=str), now, minute_ts),
+                        )
+                    else:
+                        self._conn.execute(
+                            "INSERT INTO system_metrics (minute_ts, data, created_at) VALUES (?, ?, ?)",
+                            (minute_ts, json.dumps(data, default=str), now),
+                        )
+                    self._conn.commit()
             await _run_in_sqlite_pool(_store)
         except Exception as e:
             logger.warning(f"store_metrics_minute failed: {e}")
@@ -442,17 +451,18 @@ class SqliteCache(CacheBase):
         """读取指定时间范围内的分钟指标数据"""
         try:
             def _load():
-                rows = self._conn.execute(
-                    "SELECT data FROM system_metrics WHERE minute_ts >= ? AND minute_ts <= ? ORDER BY minute_ts",
-                    (start_ts, end_ts),
-                ).fetchall()
-                results = []
-                for row in rows:
-                    try:
-                        results.append(json.loads(row["data"]))
-                    except (json.JSONDecodeError, TypeError):
-                        pass
-                return results
+                with self._lock:
+                    rows = self._conn.execute(
+                        "SELECT data FROM system_metrics WHERE minute_ts >= ? AND minute_ts <= ? ORDER BY minute_ts",
+                        (start_ts, end_ts),
+                    ).fetchall()
+                    results = []
+                    for row in rows:
+                        try:
+                            results.append(json.loads(row["data"]))
+                        except (json.JSONDecodeError, TypeError):
+                            pass
+                    return results
             return await _run_in_sqlite_pool(_load)
         except Exception as e:
             logger.warning(f"load_metrics_range failed: {e}")
@@ -462,11 +472,12 @@ class SqliteCache(CacheBase):
         """删除指定时间之前的 metrics 数据"""
         try:
             def _cleanup():
-                self._conn.execute(
-                    "DELETE FROM system_metrics WHERE minute_ts < ?",
-                    (before_ts,),
-                )
-                self._conn.commit()
+                with self._lock:
+                    self._conn.execute(
+                        "DELETE FROM system_metrics WHERE minute_ts < ?",
+                        (before_ts,),
+                    )
+                    self._conn.commit()
             await _run_in_sqlite_pool(_cleanup)
         except Exception as e:
             logger.warning(f"cleanup_old_metrics failed: {e}")

hy_memory/data/history_store.py

@@ -16,6 +16,7 @@ import json
 import logging
 import os
 import sqlite3
+import threading
 from datetime import datetime, timezone
 from typing import Dict, List, Any, Optional
 
@@ -89,6 +90,10 @@ class HistoryStore:
         """
         self._db_path = config.history.db_path
         self._conn: Optional[sqlite3.Connection] = None
+        # sqlite3.Connection.close() must never race with an execute() on the
+        # same connection. The lock also serializes operations on this shared
+        # connection; separate HistoryStore instances remain independent.
+        self._conn_lock = threading.RLock()
 
     async def initialize(self) -> None:
         """创建表和索引（如果不存在）"""
@@ -97,20 +102,21 @@ class HistoryStore:
 
     def _init_sync(self) -> None:
         """同步初始化：创建目录、连接数据库、建表建索引"""
-        # 确保目录存在
-        db_dir = os.path.dirname(self._db_path)
-        if db_dir:
-            os.makedirs(db_dir, exist_ok=True)
+        with self._conn_lock:
+            # 确保目录存在
+            db_dir = os.path.dirname(self._db_path)
+            if db_dir:
+                os.makedirs(db_dir, exist_ok=True)
 
-        self._conn = sqlite3.connect(self._db_path, check_same_thread=False)
-        self._conn.row_factory = sqlite3.Row
-        # 开启 WAL 模式，提高并发读写性能
-        self._conn.execute("PRAGMA journal_mode=WAL;")
+            self._conn = sqlite3.connect(self._db_path, check_same_thread=False)
+            self._conn.row_factory = sqlite3.Row
+            # 开启 WAL 模式，提高并发读写性能
+            self._conn.execute("PRAGMA journal_mode=WAL;")
 
-        self._conn.execute(_CREATE_TABLE_SQL)
-        for idx_sql in _CREATE_INDEXES_SQL:
-            self._conn.execute(idx_sql)
-        self._conn.commit()
+            self._conn.execute(_CREATE_TABLE_SQL)
+            for idx_sql in _CREATE_INDEXES_SQL:
+                self._conn.execute(idx_sql)
+            self._conn.commit()
 
     # ================================================================
     # 记录操作
@@ -218,13 +224,14 @@ class HistoryStore:
         return await _run_in_sqlite_pool(self._get_history_sync, memory_id)
 
     def _get_history_sync(self, memory_id: str) -> List[Dict[str, Any]]:
-        if self._conn is None:
-            return []
-        cursor = self._conn.execute(
-            "SELECT * FROM memory_history WHERE memory_id = ? ORDER BY created_at ASC",
-            (memory_id,),
-        )
-        return [dict(row) for row in cursor.fetchall()]
+        with self._conn_lock:
+            if self._conn is None:
+                return []
+            cursor = self._conn.execute(
+                "SELECT * FROM memory_history WHERE memory_id = ? ORDER BY created_at ASC",
+                (memory_id,),
+            )
+            return [dict(row) for row in cursor.fetchall()]
 
     async def get_recent(
         self,
@@ -244,14 +251,15 @@ class HistoryStore:
         return await _run_in_sqlite_pool(self._get_recent_sync, isolation_key, limit)
 
     def _get_recent_sync(self, isolation_key: str, limit: int) -> List[Dict[str, Any]]:
-        if self._conn is None:
-            return []
-        cursor = self._conn.execute(
-            "SELECT * FROM memory_history WHERE isolation_key = ? "
-            "ORDER BY created_at DESC LIMIT ?",
-            (isolation_key, limit),
-        )
-        return [dict(row) for row in cursor.fetchall()]
+        with self._conn_lock:
+            if self._conn is None:
+                return []
+            cursor = self._conn.execute(
+                "SELECT * FROM memory_history WHERE isolation_key = ? "
+                "ORDER BY created_at DESC LIMIT ?",
+                (isolation_key, limit),
+            )
+            return [dict(row) for row in cursor.fetchall()]
 
     async def count(self, isolation_key: str = "") -> int:
         """
@@ -266,16 +274,17 @@ class HistoryStore:
         return await _run_in_sqlite_pool(self._count_sync, isolation_key)
 
     def _count_sync(self, isolation_key: str) -> int:
-        if self._conn is None:
-            return 0
-        if isolation_key:
-            cursor = self._conn.execute(
-                "SELECT COUNT(*) FROM memory_history WHERE isolation_key = ?",
-                (isolation_key,),
-            )
-        else:
-            cursor = self._conn.execute("SELECT COUNT(*) FROM memory_history")
-        return cursor.fetchone()[0]
+        with self._conn_lock:
+            if self._conn is None:
+                return 0
+            if isolation_key:
+                cursor = self._conn.execute(
+                    "SELECT COUNT(*) FROM memory_history WHERE isolation_key = ?",
+                    (isolation_key,),
+                )
+            else:
+                cursor = self._conn.execute("SELECT COUNT(*) FROM memory_history")
+            return cursor.fetchone()[0]
 
     async def close(self) -> None:
         """关闭数据库连接"""
@@ -284,9 +293,10 @@ class HistoryStore:
             logger.info("[HistoryStore] closed")
 
     def _close_sync(self) -> None:
-        if self._conn:
-            self._conn.close()
-            self._conn = None
+        with self._conn_lock:
+            if self._conn:
+                self._conn.close()
+                self._conn = None
 
     # ================================================================
     # 内部
@@ -339,33 +349,34 @@ class HistoryStore:
         role: str = "",
         extra: Optional[Dict[str, Any]] = None,
     ) -> int:
-        if self._conn is None:
-            return -1
-        now = _now_iso()
-        extra_json = json.dumps(extra or {}, ensure_ascii=False)
-        cursor = self._conn.execute(
-            """INSERT INTO memory_history
-               (memory_id, isolation_key, event,
-                old_memory, new_memory, old_status, new_status,
-                change_reason, layer, actor_id, role, extra,
-                created_at, updated_at)
-               VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
-            (
-                memory_id,
-                isolation_key,
-                event,
-                old_memory,
-                new_memory,
-                old_status,
-                new_status,
-                change_reason,
-                layer,
-                actor_id,
-                role,
-                extra_json,
-                now,
-                now,
-            ),
-        )
-        self._conn.commit()
-        return cursor.lastrowid
+        with self._conn_lock:
+            if self._conn is None:
+                return -1
+            now = _now_iso()
+            extra_json = json.dumps(extra or {}, ensure_ascii=False)
+            cursor = self._conn.execute(
+                """INSERT INTO memory_history
+                   (memory_id, isolation_key, event,
+                    old_memory, new_memory, old_status, new_status,
+                    change_reason, layer, actor_id, role, extra,
+                    created_at, updated_at)
+                   VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
+                (
+                    memory_id,
+                    isolation_key,
+                    event,
+                    old_memory,
+                    new_memory,
+                    old_status,
+                    new_status,
+                    change_reason,
+                    layer,
+                    actor_id,
+                    role,
+                    extra_json,
+                    now,
+                    now,
+                ),
+            )
+            self._conn.commit()
+            return cursor.lastrowid

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

@@ -200,10 +200,12 @@ def _trigger_sql() -> list[tuple[str, str]]:
         CREATE TRIGGER IF NOT EXISTS memory_visibility_dirty_on_link_update
         AFTER UPDATE ON governed_memory_evidence
         BEGIN
-            INSERT OR IGNORE INTO memory_visibility_dirty (memory_id)
-            VALUES (NEW.memory_id);
-            INSERT OR IGNORE INTO memory_visibility_dirty (memory_id)
-            VALUES (OLD.memory_id);
+            INSERT INTO memory_visibility_dirty (memory_id)
+            VALUES (NEW.memory_id)
+            ON CONFLICT(memory_id) DO NOTHING;
+            INSERT INTO memory_visibility_dirty (memory_id)
+            VALUES (OLD.memory_id)
+            ON CONFLICT(memory_id) DO NOTHING;
         END;""",
         ),
         (
@@ -294,6 +296,10 @@ def install_triggers(db: sqlite3.Connection) -> None:
         str(row["name"])
         for row in cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
     }
+    # CREATE TRIGGER IF NOT EXISTS cannot repair the legacy definition. Replace
+    # this derived-cache trigger transactionally on every startup so old
+    # libraries get the row-level conflict handling without a schema migration.
+    cursor.execute("DROP TRIGGER IF EXISTS memory_visibility_dirty_on_link_update")
     for table, statement in _trigger_sql():
         if table not in present:
             continue
@@ -303,11 +309,15 @@ def install_triggers(db: sqlite3.Connection) -> None:
 def startup(database_path: Path) -> None:
     """Idempotent: create the derived tables and triggers. No version change.
 
-    Runs through the write connection because the triggers are schema. It is
-    one short transaction and happens once per start.
+    Runs through the write connection because the triggers are schema. The
+    trigger replacement is one short transaction and happens once per start.
     """
-    with main_database(database_path).write() as db:
+    with main_database(database_path).write() as db, db:
         ensure_tables(db)
+        # ensure_tables uses executescript(), which commits any pending legacy
+        # transaction. Begin only afterward so replacing the installed trigger
+        # is one atomic schema change and a failed CREATE restores the old one.
+        db.execute("BEGIN IMMEDIATE")
         install_triggers(db)
 
 

memhub/apps/core/tests/test_memory_visibility.py

@@ -8,7 +8,14 @@ import pytest
 from one_memory.imports.grants import GRANT_SCHEMA
 from one_memory.imports.models import EvidenceInput, EvidenceOrigin, OriginMetadata, WrittenBy
 from one_memory.imports.storage import EvidenceStore
-from one_memory.memory_governance import MemoryDraft, MemoryGovernanceStore, MemoryKind
+from one_memory.memory_governance import (
+    HyCreate,
+    HyLink,
+    HyRevise,
+    MemoryDraft,
+    MemoryGovernanceStore,
+    MemoryKind,
+)
 from one_memory.memory_tasks import (
     MemoryTaskOutput,
     MemoryTaskStore,
@@ -19,6 +26,7 @@ from one_memory.memory_visibility import (
     install_triggers,
     pending,
     recompute,
+    startup,
 )
 from one_memory.model_profiles import (
     ModelCapabilities,
@@ -336,6 +344,197 @@ def test_starting_the_app_creates_the_derived_tables(tmp_path):
     asyncio.run(scenario())
 
 
+@pytest.mark.parametrize("visibility_mode", ["exact", "derived"])
+def test_hy_revise_and_repeated_link_reuse_the_same_evidence(library, monkeypatch, visibility_mode):
+    """The outer evidence UPSERT must not override the trigger's duplicate handling."""
+    database, evidence, _tasks, governance, _profile = library
+    monkeypatch.setenv("ONE_MEMORY_READ_VISIBILITY", visibility_mode)
+    startup(database)
+    record = evidence.put(
+        EvidenceInput(
+            source_id="cursor",
+            conversation_id="trigger-regression",
+            message_id="trigger-regression",
+            correlation_id="trigger-regression",
+            role="user",
+            written_by=WrittenBy.PERSON,
+            verbatim="用户住在杭州。",
+            origin_meta={EvidenceOrigin.FILE: OriginMetadata()},
+        )
+    ).record
+    created = governance.write_from_hy(
+        HyCreate(
+            draft=MemoryDraft(
+                user_id="local",
+                kind=MemoryKind.CHAT,
+                content="用户住在杭州。",
+                scopes=("all-apps",),
+                evidence_ids=(record.evidence_id,),
+            ),
+            request_id="trigger-create",
+            memory_id="trigger-memory",
+        ),
+        user_id="local",
+        derivation=governance.hy_derivation((record.evidence_id,)),
+    )
+    revision = HyRevise(
+        memory_id=created.memory_id,
+        request_id="trigger-revise",
+        expected_revision=created.revision,
+        content="用户住在杭州并在那里工作。",
+        evidence_ids=(record.evidence_id,),
+    )
+    revised = governance.write_from_hy(
+        revision,
+        user_id="local",
+        derivation=governance.hy_derivation((record.evidence_id,)),
+    )
+    replayed = governance.write_from_hy(
+        revision,
+        user_id="local",
+        derivation=governance.hy_derivation((record.evidence_id,)),
+    )
+    assert replayed == revised
+    link = HyLink(memory_id=created.memory_id, evidence_id=record.evidence_id)
+    for _ in range(2):
+        governance.write_from_hy(
+            link,
+            user_id="local",
+            derivation=governance.hy_derivation((record.evidence_id,)),
+        )
+
+    current = governance.get(created.memory_id, user_id="local")
+    assert current is not None
+    assert current.content == "用户住在杭州并在那里工作。"
+    assert current.revision == created.revision + 1
+    with main_database_read(database) as db:
+        assert db.execute(
+            "SELECT memory_id, COUNT(*) FROM memory_visibility_dirty GROUP BY memory_id"
+        ).fetchall() == [(created.memory_id, 1)]
+
+    if visibility_mode == "derived":
+        monkeypatch.setenv("ONE_MEMORY_READ_VISIBILITY", "exact")
+        exact = {memory.memory_id for memory in SourceReadAccess(database).filter([current])}
+        with main_database_write(database) as db:
+            assert recompute(db, [created.memory_id], readable_sql=_MEMORY_READABLE) == 1
+            db.commit()
+        monkeypatch.setenv("ONE_MEMORY_READ_VISIBILITY", "derived")
+        derived = {memory.memory_id for memory in SourceReadAccess(database).filter([current])}
+        assert derived == exact
+        with main_database_read(database) as db:
+            assert created.memory_id not in pending(db)
+
+
+def test_startup_atomically_replaces_the_legacy_link_update_trigger(tmp_path, monkeypatch):
+    import one_memory.memory_visibility as visibility_module
+    from one_memory.schema_gate import guard_schema_version
+
+    database = tmp_path / "one-memory.sqlite3"
+    expected_version = guard_schema_version(database)
+    evidence = EvidenceStore(database)
+    governance = MemoryGovernanceStore(database)
+    record = evidence.put(
+        EvidenceInput(
+            source_id="cursor",
+            conversation_id="legacy-trigger",
+            message_id="legacy-trigger",
+            correlation_id="legacy-trigger",
+            role="user",
+            written_by=WrittenBy.PERSON,
+            verbatim="升级失败不能改变这条原文。",
+            origin_meta={EvidenceOrigin.FILE: OriginMetadata()},
+        )
+    ).record
+    memory = as_person(governance).create(
+        MemoryDraft(
+            user_id="local",
+            kind=MemoryKind.CHAT,
+            content="升级失败不能改变这条记忆。",
+            scopes=("all-apps",),
+            evidence_ids=(record.evidence_id,),
+        ),
+        request_id="legacy-trigger-memory",
+    )
+    startup(database)
+    with sqlite3.connect(database) as db:
+        db.execute("DROP TRIGGER memory_visibility_dirty_on_link_update")
+        db.executescript(
+            """
+            CREATE TRIGGER memory_visibility_dirty_on_link_update
+            AFTER UPDATE ON governed_memory_evidence
+            BEGIN
+                INSERT OR IGNORE INTO memory_visibility_dirty (memory_id)
+                VALUES (NEW.memory_id);
+                INSERT OR IGNORE INTO memory_visibility_dirty (memory_id)
+                VALUES (OLD.memory_id);
+            END;
+            """
+        )
+        db.execute(
+            "INSERT INTO memory_visibility (memory_id, visible, reason, computed_at) "
+            "VALUES (?, 1, 'ok', '2026-10-19T00:00:00+00:00')",
+            (memory.memory_id,),
+        )
+        db.execute(
+            "INSERT INTO memory_visibility_dirty (memory_id) VALUES (?)",
+            (memory.memory_id,),
+        )
+        legacy_trigger = db.execute(
+            "SELECT sql FROM sqlite_master "
+            "WHERE type='trigger' AND name='memory_visibility_dirty_on_link_update'"
+        ).fetchone()[0]
+        before = tuple(db.iterdump())
+
+    real_trigger_sql = visibility_module._trigger_sql
+
+    def fail_after_drop() -> list[tuple[str, str]]:
+        statements = real_trigger_sql()
+        return [
+            (table, "CREATE TRIGGER")
+            if "CREATE TRIGGER IF NOT EXISTS memory_visibility_dirty_on_link_update" in statement
+            else (table, statement)
+            for table, statement in statements
+        ]
+
+    with monkeypatch.context() as failure:
+        failure.setattr(visibility_module, "_trigger_sql", fail_after_drop)
+        with pytest.raises(sqlite3.OperationalError):
+            startup(database)
+
+    with sqlite3.connect(database) as db:
+        assert tuple(db.iterdump()) == before
+        assert (
+            db.execute(
+                "SELECT sql FROM sqlite_master "
+                "WHERE type='trigger' AND name='memory_visibility_dirty_on_link_update'"
+            ).fetchone()[0]
+            == legacy_trigger
+        )
+        assert db.execute("SELECT MAX(version) FROM schema_migrations").fetchone()[0] == (
+            expected_version
+        )
+
+    startup(database)
+    startup(database)
+
+    with sqlite3.connect(database) as db:
+        trigger_sql = db.execute(
+            "SELECT sql FROM sqlite_master "
+            "WHERE type='trigger' AND name='memory_visibility_dirty_on_link_update'"
+        ).fetchone()[0]
+        assert trigger_sql.count("ON CONFLICT(memory_id) DO NOTHING") == 2
+        assert "INSERT OR IGNORE" not in trigger_sql
+        assert db.execute("SELECT MAX(version) FROM schema_migrations").fetchone()[0] == (
+            expected_version
+        )
+        assert db.execute("SELECT * FROM memory_visibility").fetchall() == [
+            (memory.memory_id, 1, "ok", "2026-10-19T00:00:00+00:00")
+        ]
+        assert db.execute("SELECT * FROM memory_visibility_dirty").fetchall() == [
+            (memory.memory_id,)
+        ]
+
+
 def test_the_audit_notices_a_conclusion_that_no_longer_holds(library):
     """The ongoing proof: recompute a slice and compare with what is stored.
 

tests/test_block_l1_raw_in_read_filter.py

@@ -12,8 +12,9 @@ L1_RAW 对外读路径不可见：闸门焊在向量库唯一查询入口。
 hybrid_search_native / list_by_user 四个查询方法的共同必经处，
 上层无法绕开，新增接口自动受保护。
 
-按 ID 直取走主键查询、不经过 filter，故在 `client.async_get` 处单独挡
-（GET /v1/memories/{id} 与 /v1/memories/batch 的共同必经处）。
+按 ID 直取走主键查询、不经过 filter，故在 `client.async_get` 处默认挡住。
+Python SDK 调用方可显式传 `include_raw=True` 读取；HTTP 接口不传该参数，
+继续保持原有隐藏行为。
 
 内部管线需读原始对话时显式 allow_raw_layer=True 放行：
 - `writer._get_recent_dialogue_history`：提取前拼多轮对话历史喂 LLM
@@ -124,28 +125,63 @@ class TestListNoLongerFiltersInProcess:
             "list 又出现进程内过滤 raw 的写法，应交由 vdb 侧 filter 完成"
 
 
-class TestGetByIdPathBlocksRaw:
-    """按 ID 直取：走主键查询不经 filter，须在 async_get 处挡住。"""
+class TestGetByIdRawOptIn:
+    """按 ID 直取默认隐藏 raw，只有 SDK 显式选择时才返回。"""
 
-    def test_async_get_drops_raw_node(self):
+    @staticmethod
+    def _client(node):
+        from hy_memory.client import HyMemoryClient
+
+        class Store:
+            async def get_by_id(self, memory_id):
+                return node if memory_id == node.node_id else None
+
+        async def no_op_log(**kwargs):
+            return None
+
+        client = HyMemoryClient.__new__(HyMemoryClient)
+        client._vector_store = Store()
+        client._graph_store = None
+        client._backend_identity = lambda: {}
+        client._maybe_store_op_log = no_op_log
+        return client
+
+    @staticmethod
+    def _raw_node():
+        return MemoryNode(
+            node_id="raw-1",
+            user_id="u1",
+            agent_id="a1",
+            session_id="s1",
+            content="原始对话",
+            layer=MemoryLayer.L1_RAW,
+            status=MemoryStatus.ACTIVE,
+            custom={"meta": [{"source": "test"}]},
+        )
+
+    def test_default_still_hides_raw(self):
+        from hy_memory.client import HyMemoryClient
+
+        client = self._client(self._raw_node())
+        raw_get = HyMemoryClient.async_get.__wrapped__
+        assert run(raw_get(client, "raw-1")) is None
+
+    def test_explicit_include_raw_returns_lite_memory(self):
+        from hy_memory.client import HyMemoryClient
+
+        client = self._client(self._raw_node())
+        raw_get = HyMemoryClient.async_get.__wrapped__
+        result = run(raw_get(client, "raw-1", include_raw=True))
+
+        assert result["memory_id"] == "raw-1"
+        assert result["content"] == "原始对话"
+        assert result["layer"] == MemoryLayer.L1_RAW.value
+        assert result["meta"] == [{"source": "test"}]
+
+    def test_public_get_exposes_keyword_only_opt_in(self):
         import inspect
         from hy_memory.client import HyMemoryClient
-        src = inspect.getsource(HyMemoryClient.async_get)
-        assert "MemoryLayer.L1_RAW" in src, \
-            "async_get 未挡 raw，batch 与按 ID 单取会漏出原始对话"
 
-    def test_block_happens_before_return(self):
-        """挡的位置必须在两处 return 之前，否则等于没挡。"""
-        import inspect
-        from hy_memory.client import HyMemoryClient
-        lines = inspect.getsource(HyMemoryClient.async_get).splitlines()
-        block_at = next(
-            i for i, l in enumerate(lines)
-            if "MemoryLayer.L1_RAW" in l and "vdb_node" in l
-        )
-        first_return_at = next(
-            i for i, l in enumerate(lines)
-            if l.strip().startswith("return {")
-        )
-        assert block_at < first_return_at, \
-            f"挡 raw 在第 {block_at} 行、首个 return 在第 {first_return_at} 行，顺序错了"
+        param = inspect.signature(HyMemoryClient.get).parameters["include_raw"]
+        assert param.default is False
+        assert param.kind is inspect.Parameter.KEYWORD_ONLY

tests/test_sqlite_connection_lifecycle.py

@@ -0,0 +1,136 @@
+import asyncio
+import threading
+
+import pytest
+
+from hy_memory.config import MemoryConfig
+from hy_memory.data.cache_sqlite import SqliteCache
+from hy_memory.data.history_store import HistoryStore
+
+
+class _BlockingCursor:
+    def fetchone(self):
+        return (1,)
+
+    def fetchall(self):
+        return []
+
+
+class _BlockingConnection:
+    """Hold execute() open so a concurrent close() can be observed."""
+
+    def __init__(self):
+        self.entered = threading.Event()
+        self.release = threading.Event()
+        self.closed = threading.Event()
+
+    def execute(self, *_args, **_kwargs):
+        self.entered.set()
+        if not self.release.wait(timeout=2):
+            raise TimeoutError("test did not release the blocking query")
+        if self.closed.is_set():
+            raise AssertionError("connection was closed while execute() was active")
+        return _BlockingCursor()
+
+    def close(self):
+        self.closed.set()
+
+
+def _config(tmp_path) -> MemoryConfig:
+    config = MemoryConfig()
+    config.history.db_path = str(tmp_path / "history.db")
+    config.cache.db_path = str(tmp_path / "cache.db")
+    return config
+
+
+async def _wait_for(event: threading.Event) -> None:
+    assert await asyncio.wait_for(asyncio.to_thread(event.wait, 1), timeout=2)
+
+
+@pytest.mark.asyncio
+async def test_history_close_waits_for_active_query(tmp_path):
+    store = HistoryStore(_config(tmp_path))
+    connection = _BlockingConnection()
+    store._conn = connection
+
+    query = asyncio.create_task(store.count())
+    await _wait_for(connection.entered)
+    closing = asyncio.create_task(store.close())
+    await asyncio.sleep(0.05)
+
+    assert not connection.closed.is_set()
+    connection.release.set()
+    assert await query == 1
+    await closing
+    assert connection.closed.is_set()
+
+
+@pytest.mark.asyncio
+async def test_cache_close_waits_for_active_query(tmp_path):
+    store = SqliteCache(_config(tmp_path))
+    connection = _BlockingConnection()
+    store._conn = connection
+
+    query = asyncio.create_task(store.load_metrics_range("0", "9"))
+    await _wait_for(connection.entered)
+    closing = asyncio.create_task(store.close())
+    await asyncio.sleep(0.05)
+
+    assert not connection.closed.is_set()
+    connection.release.set()
+    assert await query == []
+    await closing
+    assert connection.closed.is_set()
+
+
+@pytest.mark.asyncio
+async def test_history_concurrent_reads_and_writes_are_serialized(tmp_path):
+    store = HistoryStore(_config(tmp_path))
+    await store.initialize()
+    try:
+        inserted = await asyncio.gather(
+            *(
+                store.record_add(
+                    memory_id=f"memory-{index}",
+                    content="content",
+                    layer="l2_fact",
+                    isolation_key="user",
+                )
+                for index in range(200)
+            )
+        )
+        assert all(row_id > 0 for row_id in inserted)
+        assert len(set(inserted)) == 200
+
+        counts = await asyncio.gather(*(store.count() for _ in range(100)))
+        assert counts == [200] * 100
+    finally:
+        await store.close()
+
+
+@pytest.mark.asyncio
+async def test_cache_concurrent_pipeline_log_reads_and_writes_are_serialized(tmp_path):
+    store = SqliteCache(_config(tmp_path))
+    await store.initialize()
+    try:
+        written = await asyncio.gather(
+            *(
+                store.store_pipeline_log(
+                    request_id="request",
+                    user_id="user",
+                    agent_id="agent",
+                    step="EXTRACT",
+                    prompt=f"prompt-{index}",
+                    response="response",
+                )
+                for index in range(200)
+            )
+        )
+        assert written == [True] * 200
+
+        pages = await asyncio.gather(
+            *(store.get_pipeline_logs(request_id="request", limit=200) for _ in range(50))
+        )
+        assert all(len(page) == 200 for page in pages)
+    finally:
+        await store.close()