fix(sdk): clarify metadata and allow opt-in raw get

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

CHANGELOG.md

@@ -14,12 +14,6 @@ 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`

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.RLock 保护共享连接的全部操作与关闭
+- threading.Lock 保护写操作
 
 三张表:
 - memory_operations / pipeline_logs / system_metrics: 审计与观测
@@ -122,10 +122,7 @@ class SqliteCache(CacheBase):
     def __init__(self, config: MemoryConfig):
         self.config = config
         self._conn: Optional[sqlite3.Connection] = None
-        # 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()
+        self._lock = threading.Lock()
 
         # 读取 db_path
         cache_cfg = getattr(config, "cache", None)
@@ -147,37 +144,35 @@ class SqliteCache(CacheBase):
         logger.debug(f"SqliteCache initialized: {self._db_path}")
 
     def _init_sync(self) -> None:
-        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()
+        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 列。"""
-        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:
+        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(
-                        "ALTER TABLE pipeline_logs_v2 ADD COLUMN event_at_ms INTEGER NOT NULL DEFAULT 0"
+                        f"ALTER TABLE pipeline_logs_v2 ADD COLUMN {col} TEXT NOT NULL DEFAULT ''"
                     )
-            except Exception as e:
-                logger.debug(f"[sqlite] pipeline_logs_v2 column migration skipped: {e}")
+            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}")
 
     async def close(self) -> None:
         if self._conn is not None:
@@ -185,10 +180,9 @@ class SqliteCache(CacheBase):
         logger.info("SqliteCache closed")
 
     def _close_sync(self) -> None:
-        with self._lock:
-            if self._conn:
-                self._conn.close()
-                self._conn = None
+        if self._conn:
+            self._conn.close()
+            self._conn = None
 
     # ================================================================
     # 统计
@@ -261,40 +255,39 @@ class SqliteCache(CacheBase):
         """
         try:
             def _query():
-                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)
+                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:
@@ -376,38 +369,37 @@ class SqliteCache(CacheBase):
         """
         try:
             def _query():
-                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)
+                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:
@@ -422,27 +414,26 @@ class SqliteCache(CacheBase):
         """存储一个分钟桶的增量指标数据（UPSERT: 同一分钟多次写入会合并）"""
         try:
             def _store():
-                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()
+                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}")
@@ -451,18 +442,17 @@ class SqliteCache(CacheBase):
         """读取指定时间范围内的分钟指标数据"""
         try:
             def _load():
-                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
+                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}")
@@ -472,12 +462,11 @@ class SqliteCache(CacheBase):
         """删除指定时间之前的 metrics 数据"""
         try:
             def _cleanup():
-                with self._lock:
-                    self._conn.execute(
-                        "DELETE FROM system_metrics WHERE minute_ts < ?",
-                        (before_ts,),
-                    )
-                    self._conn.commit()
+                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,7 +16,6 @@ import json
 import logging
 import os
 import sqlite3
-import threading
 from datetime import datetime, timezone
 from typing import Dict, List, Any, Optional
 
@@ -90,10 +89,6 @@ 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:
         """创建表和索引（如果不存在）"""
@@ -102,21 +97,20 @@ class HistoryStore:
 
     def _init_sync(self) -> None:
         """同步初始化：创建目录、连接数据库、建表建索引"""
-        with self._conn_lock:
-            # 确保目录存在
-            db_dir = os.path.dirname(self._db_path)
-            if db_dir:
-                os.makedirs(db_dir, exist_ok=True)
+        # 确保目录存在
+        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()
 
     # ================================================================
     # 记录操作
@@ -224,14 +218,13 @@ 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]]:
-        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()]
+        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,
@@ -251,15 +244,14 @@ 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]]:
-        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()]
+        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:
         """
@@ -274,17 +266,16 @@ class HistoryStore:
         return await _run_in_sqlite_pool(self._count_sync, isolation_key)
 
     def _count_sync(self, isolation_key: str) -> int:
-        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]
+        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:
         """关闭数据库连接"""
@@ -293,10 +284,9 @@ class HistoryStore:
             logger.info("[HistoryStore] closed")
 
     def _close_sync(self) -> None:
-        with self._conn_lock:
-            if self._conn:
-                self._conn.close()
-                self._conn = None
+        if self._conn:
+            self._conn.close()
+            self._conn = None
 
     # ================================================================
     # 内部
@@ -349,34 +339,33 @@ class HistoryStore:
         role: str = "",
         extra: Optional[Dict[str, Any]] = None,
     ) -> int:
-        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
+        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,12 +200,10 @@ 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 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;
+            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;""",
         ),
         (
@@ -296,10 +294,6 @@ 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
@@ -309,15 +303,11 @@ 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. The
-    trigger replacement is one short transaction and happens once per start.
+    Runs through the write connection because the triggers are schema. It is
+    one short transaction and happens once per start.
     """
-    with main_database(database_path).write() as db, db:
+    with main_database(database_path).write() as 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,14 +8,7 @@ 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 (
-    HyCreate,
-    HyLink,
-    HyRevise,
-    MemoryDraft,
-    MemoryGovernanceStore,
-    MemoryKind,
-)
+from one_memory.memory_governance import MemoryDraft, MemoryGovernanceStore, MemoryKind
 from one_memory.memory_tasks import (
     MemoryTaskOutput,
     MemoryTaskStore,
@@ -26,7 +19,6 @@ from one_memory.memory_visibility import (
     install_triggers,
     pending,
     recompute,
-    startup,
 )
 from one_memory.model_profiles import (
     ModelCapabilities,
@@ -344,197 +336,6 @@ 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_sqlite_connection_lifecycle.py

@@ -1,136 +0,0 @@
-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()