fix(coding): default-off search chain_block (diff) injection

Medium/heavy coding users pay large prompt tokens for the version-evolution
diff block; make MEMORY_CODING_CHAIN_BLOCK_ENABLED opt-in (default false).
Write-side observe is unchanged; HCM_BLOCK_OFF remains an advanced kill switch.

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

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,18 +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

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/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,62 +459,6 @@ class MysqlCache(CacheBase):
             logger.warning(f"get_pipeline_logs failed: {e}")
             return []
 
-    async def claim_write_outcome(
-        self,
-        request_id: str,
-        user_id: str,
-        agent_id: str,
-        session_id: str,
-        operation: str,
-    ) -> Optional[Dict[str, Any]]:
-        now = datetime.now().isoformat()
-        inserted = await self._execute(
-            """
-            INSERT IGNORE INTO write_outcomes (
-                request_id, user_id, agent_id, session_id, operation,
-                status, result_json, created_at, updated_at
-            ) VALUES (%s, %s, %s, %s, %s, 'running', '{}', %s, %s)
-            """,
-            (request_id, user_id, agent_id, session_id, operation, now, now),
-        )
-        if inserted == 1:
-            return None
-        return await self.get_write_outcome(request_id)
-
-    async def finish_write_outcome(
-        self,
-        request_id: str,
-        status: str,
-        result: Dict[str, Any],
-    ) -> bool:
-        updated = await self._execute(
-            """
-            UPDATE write_outcomes
-            SET status = %s, result_json = %s, updated_at = %s
-            WHERE request_id = %s
-            """,
-            (
-                status,
-                json.dumps(result, ensure_ascii=False, default=str),
-                datetime.now().isoformat(),
-                request_id,
-            ),
-        )
-        return updated == 1
-
-    async def get_write_outcome(self, request_id: str) -> Optional[Dict[str, Any]]:
-        row = await self._fetchone(
-            "SELECT * FROM write_outcomes WHERE request_id = %s",
-            (request_id,),
-        )
-        if row is None:
-            return None
-        try:
-            row["result"] = json.loads(row.pop("result_json") or "{}")
-        except (TypeError, json.JSONDecodeError):
-            row["result"] = {}
-        return row
-
     # ================================================================
     # System Metrics（分钟粒度落盘）
     # ================================================================

hy_memory/data/cache_sqlite.py

@@ -87,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/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）。
@@ -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/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,
-    )

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/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_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_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_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_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" },