fix(migration): stage SDK data before cutover

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

CHANGELOG.md

@@ -41,13 +41,13 @@ Hy Memory and MemHub are versioned and released independently.
 ## Unreleased
 
 - Removed MemHub's private `hy_memory_bridge` implementation. MemHub now calls
-  public `HyMemoryClient` methods through explicit runtime, write, migration,
+  public `HyMemoryClient` methods at runtime; prompt/reconcile/response
-  and read modules; prompt/reconcile/response monkey-patches and private
+  monkey-patches and private VDB/Coding runtime access were deleted. Kuzu Graph
-  VDB/Coding access were deleted. Kuzu Graph projection/admin remains the one
+  projection/admin and stopped-writer App migration are the documented product
-  documented product exception.
+  exceptions.
 - Extended public SDK `get()` to resolve Chat, Coding Experience, and Working
-  Fact IDs, including owner/session fields and explicit ambiguity handling;
+  Fact IDs, including raw records, owner/session fields, and explicit ambiguity
-  added constructor-time public embedding-service injection.
+  handling; added constructor-time public embedding-service injection.
 - Added a MemHub-managed native Hermes MemoryProvider with automatic prefetch,
   persistent per-session memory-ID deduplication, authorized Broker reads, and
   recall-context stripping to prevent memory recursion.
@@ -58,13 +58,12 @@ Hy Memory and MemHub are versioned and released independently.
 - `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
+  idempotent durable write outcomes, and quiesced local data-plane
-  quiesced local data-plane snapshot/restore contracts for the MemHub
+  snapshot/restore contracts for the MemHub authority migration. Exact legacy
-  authority migration.
+  backfill is now an App-owned privileged offline operation rather than a
+  public `HyMemoryClient` API. Backfill runs against a complete staging copy;
+  cutover atomically swaps SDK roots and restores the previous root on failure.
 - 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
@@ -87,9 +86,10 @@ Hy Memory and MemHub are versioned and released independently.
   retry scheduling for their memory tasks.
 - 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`
+- Unified exact reads and lists across memory layers: `get(memory_id)` returns
-  memories, including memories written in lite mode. The default remains
+  the matched layer, while `list_memories(layers=["l1_raw"])` explicitly lists
-  hidden for backward compatibility.
+  raw records. Separate raw get/list methods and the `include_raw` switch were
+  removed.
 - Removed MemHub's task-scoped HyPermission store wrappers and global
   model-content consent switch. The trusted embedded Hy Memory SDK now uses
   its native local stores directly.

docs/usage.md

@@ -273,8 +273,8 @@ catalog 也可以在没有基础 L0 时形成只读的合成画像。
 
 ```python
 memory = client.get("memory_id")
-# lite 模式写入的是 L1_RAW；按 ID 读取时必须显式选择返回原文。
+# raw ID 返回 L1_RAW，fact ID 返回对应事实层。
-raw_memory = client.get("memory_id", include_raw=True)
+raw_memory = client.get("raw_id")
 result = client.update("memory_id", "更新后的内容")
 result = client.delete("memory_id")
 result = client.delete_all(user_id="user_001")
@@ -284,7 +284,7 @@ page = client.list_memories(
     limit=100,
     offset=0,
     order="desc",
-    layers=None,
+    layers=["l1_raw"],  # 显式列出 raw；不传仍沿用默认语义层列表
     with_evolution=False,
 )
 
@@ -294,9 +294,10 @@ result = client.clone_user("user_001", "user_002", agent_id=None)
 对应异步方法为 `async_get()`、`async_update()`、`async_delete()`、
 `async_delete_all()`、`async_list_memories()`、`async_clone_user()`。
 
-`get()` / `async_get()` 默认继续隐藏 `L1_RAW`，因此不会改变现有调用方的原文访问
+`get()` / `async_get()` 是统一的 ID 精确读取：返回结果的 `layer`、`kind` 和
-边界。只有显式传入 `include_raw=True` 才返回该层；这也是 lite 模式按 ID 读取
+`variant` 表明实际类型。授权由调用 SDK 前的产品边界完成，不由 SDK 根据 layer
-写入内容及其 `meta` 的方式。MemHub 的 HTTP 读取接口不开放此参数。
+猜测。`list_memories()` 默认仍不包含 `L1_RAW`；显式传
+`layers=["l1_raw"]` 时按同一列表合同返回 raw。
 
 普通 `list_memories()` 返回 `vdb` 桶，并在 Graph 有结果时附带 `graph`；传入
 `workspace_id` 时返回生产力记忆的 `scene="productivity"` 和 `experiences`。

hy_memory/client.py

@@ -1886,7 +1886,6 @@ class HyMemoryClient:
         self,
         memory_id: str,
         *,
-        include_raw: bool = False,
         kind: Optional[MemoryKind] = None,
         variant: Optional[MemoryVariant] = None,
     ) -> Optional[Dict[str, Any]]:
@@ -1895,8 +1894,6 @@ class HyMemoryClient:
 
         Args:
             memory_id: 记忆 ID
-            include_raw: 是否允许返回 L1_RAW 原始记忆。默认 False，保持原有隐藏行为；
-                         lite 模式的记忆均为 L1_RAW，需显式设为 True 才能按 ID 读取。
             kind: 可选 namespace（"chat" / "experience" / "fact"）。
             variant: 可选生产力变体（"working" / "coding"）。
 
@@ -1906,7 +1903,6 @@ class HyMemoryClient:
         return self._loop_thread.run(
             self.async_get(
                 memory_id,
-                include_raw=include_raw,
                 kind=kind,
                 variant=variant,
             )
@@ -1978,6 +1974,9 @@ class HyMemoryClient:
         branch: Optional[str] = None,
         user_ids: Optional[List[str]] = None,
         agent_ids: Optional[List[str]] = None,
+        session_ids: Optional[List[str]] = None,
+        include_sources: Optional[List[str]] = None,
+        exclude_sources: Optional[List[str]] = None,
         layers: Optional[List[Any]] = None,
         statuses: Optional[List[Any]] = None,
         with_evolution: bool = False,
@@ -2000,6 +1999,9 @@ class HyMemoryClient:
             user_ids: 可选，多用户批量列举。非空时优先于 user_id，返回并集
                       （分页在合并排序后统一施加）。
             agent_ids: 可选，多 agent 批量列举。非空时优先于 agent_id。
+            session_ids: 可选，只返回指定 session。
+            include_sources: 可选，只返回指定来源。
+            exclude_sources: 可选，排除指定来源；与 include_sources 冲突时排除优先。
             layers:   可选，只返回指定层级（字符串或 MemoryLayer）。不传时返回
                       L1_RAW 以外的全部层级，与改动前一致。
             statuses: 可选，只返回指定生命周期（字符串或 MemoryStatus）。不传时
@@ -2015,6 +2017,9 @@ class HyMemoryClient:
             self.async_list_memories(
                 user_id=user_id, agent_id=agent_id,
                 user_ids=user_ids, agent_ids=agent_ids,
+                session_ids=session_ids,
+                include_sources=include_sources,
+                exclude_sources=exclude_sources,
                 limit=limit, offset=offset, order=order,
                 workspace_id=workspace_id, branch=branch,
                 layers=layers, statuses=statuses, with_evolution=with_evolution,
@@ -2975,106 +2980,6 @@ class HyMemoryClient:
             include_runtime_diag=True,
         )
 
-    def import_raw(
-        self,
-        content: str,
-        *,
-        user_id: str,
-        agent_id: str = "default_agent",
-        session_id: str = "default_session",
-        source: str = "default",
-        memory_at: Optional[datetime] = None,
-        metadata: Optional[Dict[str, Any]] = None,
-        raw_id: Optional[str] = None,
-        request_id: Optional[str] = None,
-    ) -> Dict[str, Any]:
-        """Persist one exact raw record without extraction or reconciliation."""
-        return self._loop_thread.run(
-            self.async_import_raw(
-                content,
-                user_id=user_id,
-                agent_id=agent_id,
-                session_id=session_id,
-                source=source,
-                memory_at=memory_at,
-                metadata=metadata,
-                raw_id=raw_id,
-                request_id=request_id,
-            ),
-            include_runtime_diag=True,
-        )
-
-    def import_memory_exact(
-        self,
-        content: str,
-        *,
-        memory_id: str,
-        user_id: str,
-        agent_id: str = "default_agent",
-        session_id: str = "default_session",
-        source: str = "default",
-        layer: Any = "l2_fact",
-        status: Any = "active",
-        owner: Optional[str] = None,
-        tags: Optional[List[str]] = None,
-        supersedes: Optional[List[str]] = None,
-        superseded_by: Optional[List[str]] = None,
-        is_latest: bool = True,
-        memory_at: Optional[datetime] = None,
-        metadata: Optional[Dict[str, Any]] = None,
-        request_id: Optional[str] = None,
-    ) -> Dict[str, Any]:
-        """Import one existing semantic memory without extraction/reconcile."""
-        return self._loop_thread.run(
-            self.async_import_memory_exact(
-                content,
-                memory_id=memory_id,
-                user_id=user_id,
-                agent_id=agent_id,
-                session_id=session_id,
-                source=source,
-                layer=layer,
-                status=status,
-                owner=owner,
-                tags=tags,
-                supersedes=supersedes,
-                superseded_by=superseded_by,
-                is_latest=is_latest,
-                memory_at=memory_at,
-                metadata=metadata,
-                request_id=request_id,
-            ),
-            include_runtime_diag=True,
-        )
-
-    def get_raw(self, raw_id: str) -> Optional[Dict[str, Any]]:
-        """Read one SDK raw record through the explicit raw API."""
-        return self._loop_thread.run(self.async_get_raw(raw_id))
-
-    def list_raw(
-        self,
-        *,
-        user_id: str,
-        agent_id: Optional[str] = None,
-        session_id: Optional[str] = None,
-        source: Optional[str] = None,
-        limit: int = 100,
-        offset: int = 0,
-        order: str = "desc",
-    ) -> Dict[str, Any]:
-        """List raw records for an authorized SDK owner/scope."""
-        return self._loop_thread.run(
-            self.async_list_raw(
-                user_id=user_id,
-                agent_id=agent_id,
-                session_id=session_id,
-                source=source,
-                limit=limit,
-                offset=offset,
-                order=order,
-            )
-        )
-
     @_ensure_internal_loop
     async def async_add_extracted(
         self,
@@ -3227,316 +3132,6 @@ class HyMemoryClient:
         }
 
     @_ensure_internal_loop
-    async def async_import_raw(
-        self,
-        content: str,
-        *,
-        user_id: str,
-        agent_id: str = "default_agent",
-        session_id: str = "default_session",
-        source: str = "default",
-        memory_at: Optional[datetime] = None,
-        metadata: Optional[Dict[str, Any]] = None,
-        raw_id: Optional[str] = None,
-        request_id: Optional[str] = None,
-    ) -> Dict[str, Any]:
-        """Persist exact source content as L1_RAW without running a model."""
-        from .utils.source_filter import normalize_source
-
-        if not isinstance(content, str) or not content:
-            raise ValueError("content is required")
-        if not user_id:
-            raise ValueError("user_id is required")
-        request_id = request_id or str(uuid.uuid4())
-        replay = await self._claim_write_outcome(
-            request_id=request_id,
-            user_id=user_id,
-            agent_id=agent_id,
-            session_id=session_id,
-            operation="import_raw",
-        )
-        if replay is not None:
-            return replay
-        try:
-            result = await self._async_import_raw_impl(
-                content,
-                user_id=user_id,
-                agent_id=agent_id,
-                session_id=session_id,
-                source=normalize_source(source),
-                memory_at=memory_at,
-                metadata=metadata,
-                raw_id=raw_id,
-                request_id=request_id,
-            )
-        except BaseException:
-            await self._mark_write_outcome_unknown(request_id)
-            raise
-        await self._finish_write_outcome(request_id, "succeeded", result)
-        return result
-
-    async def _async_import_raw_impl(
-        self,
-        content: str,
-        *,
-        user_id: str,
-        agent_id: str,
-        session_id: str,
-        source: str,
-        memory_at: Optional[datetime],
-        metadata: Optional[Dict[str, Any]],
-        raw_id: Optional[str],
-        request_id: str,
-    ) -> Dict[str, Any]:
-        from .models.memory import MemoryLayer, MemoryNode, MemoryStatus, SourceType
-
-        node_id = raw_id or str(uuid.uuid4())
-        existing = await self._vector_store.get_by_id(node_id)
-        if existing is not None:
-            identity = (
-                existing.layer == MemoryLayer.L1_RAW,
-                existing.content == content,
-                existing.user_id == user_id,
-                existing.agent_id == agent_id,
-                existing.session_id == session_id,
-                existing.source == source,
-            )
-            if not all(identity):
-                raise ValueError("raw_id is already bound to different source content")
-            return {
-                "success": True,
-                "raw_id": node_id,
-                "request_id": request_id,
-                "replayed": True,
-            }
-
-        custom = dict(metadata or {})
-        custom["request_id"] = request_id
-        custom["input_kind"] = "raw_exact_import"
-        embedding = await self._embed_service.embed(content)
-        node = MemoryNode(
-            node_id=node_id,
-            user_id=user_id,
-            agent_id=agent_id,
-            session_id=session_id,
-            source=source,
-            layer=MemoryLayer.L1_RAW,
-            content=content,
-            source_type=SourceType.EXPLICIT,
-            status=MemoryStatus.ACTIVE,
-            is_latest=True,
-            embedding=embedding,
-            memory_at=memory_at,
-            custom=custom,
-        )
-        stored_id = await self._vector_store.upsert(node)
-        if stored_id != node_id:
-            raise RuntimeError("raw import changed the requested identity")
-        return {"success": True, "raw_id": node_id, "request_id": request_id}
-
-    @_ensure_internal_loop
-    async def async_import_memory_exact(
-        self,
-        content: str,
-        *,
-        memory_id: str,
-        user_id: str,
-        agent_id: str = "default_agent",
-        session_id: str = "default_session",
-        source: str = "default",
-        layer: Any = "l2_fact",
-        status: Any = "active",
-        owner: Optional[str] = None,
-        tags: Optional[List[str]] = None,
-        supersedes: Optional[List[str]] = None,
-        superseded_by: Optional[List[str]] = None,
-        is_latest: bool = True,
-        memory_at: Optional[datetime] = None,
-        metadata: Optional[Dict[str, Any]] = None,
-        request_id: Optional[str] = None,
-    ) -> Dict[str, Any]:
-        """Import a canonical memory node exactly, with no model invocation."""
-        from .models.memory import MemoryLayer, MemoryStatus
-        from .utils.source_filter import normalize_source
-
-        if not content or not memory_id or not user_id:
-            raise ValueError("content, memory_id and user_id are required")
-        normalized_layer = (
-            layer if isinstance(layer, MemoryLayer) else MemoryLayer.from_string(str(layer))
-        )
-        if normalized_layer == MemoryLayer.L1_RAW:
-            raise ValueError("use import_raw for L1_RAW")
-        normalized_status = (
-            status if isinstance(status, MemoryStatus) else MemoryStatus(str(status).lower())
-        )
-        request_id = request_id or str(uuid.uuid4())
-        replay = await self._claim_write_outcome(
-            request_id=request_id,
-            user_id=user_id,
-            agent_id=agent_id,
-            session_id=session_id,
-            operation="import_memory_exact",
-        )
-        if replay is not None:
-            return replay
-        try:
-            result = await self._async_import_memory_exact_impl(
-                content,
-                memory_id=memory_id,
-                user_id=user_id,
-                agent_id=agent_id,
-                session_id=session_id,
-                source=normalize_source(source),
-                layer=normalized_layer,
-                status=normalized_status,
-                owner=owner,
-                tags=tags or [],
-                supersedes=supersedes or [],
-                superseded_by=superseded_by or [],
-                is_latest=is_latest,
-                memory_at=memory_at,
-                metadata=metadata,
-                request_id=request_id,
-            )
-        except BaseException:
-            await self._mark_write_outcome_unknown(request_id)
-            raise
-        await self._finish_write_outcome(request_id, "succeeded", result)
-        return result
-
-    async def _async_import_memory_exact_impl(
-        self,
-        content: str,
-        *,
-        memory_id: str,
-        user_id: str,
-        agent_id: str,
-        session_id: str,
-        source: str,
-        layer,
-        status,
-        owner: Optional[str],
-        tags: List[str],
-        supersedes: List[str],
-        superseded_by: List[str],
-        is_latest: bool,
-        memory_at: Optional[datetime],
-        metadata: Optional[Dict[str, Any]],
-        request_id: str,
-    ) -> Dict[str, Any]:
-        from .models.memory import MemoryNode, SourceType
-
-        existing = await self._vector_store.get_by_id(memory_id)
-        if existing is not None:
-            identity = (
-                existing.content == content,
-                existing.user_id == user_id,
-                existing.agent_id == agent_id,
-                existing.session_id == session_id,
-                existing.source == source,
-                existing.layer == layer,
-                existing.status == status,
-                existing.owner == owner,
-                list(existing.tags or []) == tags,
-                list(existing.supersedes or []) == supersedes,
-                list(existing.superseded_by or []) == superseded_by,
-                existing.is_latest == is_latest,
-            )
-            if not all(identity):
-                raise ValueError("memory_id is already bound to different memory content")
-            return {
-                "success": True,
-                "memory_id": memory_id,
-                "memory_ids": [memory_id],
-                "request_id": request_id,
-                "replayed": True,
-            }
-
-        custom = dict(metadata or {})
-        custom["request_id"] = request_id
-        custom["input_kind"] = "memory_exact_import"
-        embedding = await self._embed_service.embed(content)
-        node = MemoryNode(
-            node_id=memory_id,
-            user_id=user_id,
-            agent_id=agent_id,
-            session_id=session_id,
-            source=source,
-            layer=layer,
-            content=content,
-            source_type=SourceType.EXPLICIT,
-            status=status,
-            owner=owner,
-            tags=list(tags),
-            supersedes=list(supersedes),
-            superseded_by=list(superseded_by),
-            is_latest=is_latest,
-            embedding=embedding,
-            memory_at=memory_at,
-            custom=custom,
-        )
-        stored_id = await self._vector_store.upsert(node)
-        if stored_id != memory_id:
-            raise RuntimeError("exact memory import changed the requested identity")
-        return {
-            "success": True,
-            "memory_id": memory_id,
-            "memory_ids": [memory_id],
-            "request_id": request_id,
-        }
-
-    @_ensure_internal_loop
-    async def async_get_raw(self, raw_id: str) -> Optional[Dict[str, Any]]:
-        from .models.memory import MemoryLayer
-
-        node = await self._vector_store.get_by_id(raw_id)
-        if node is None or node.layer != MemoryLayer.L1_RAW:
-            return None
-        return self._raw_node_to_public(node)
-
-    @_ensure_internal_loop
-    async def async_list_raw(
-        self,
-        *,
-        user_id: str,
-        agent_id: Optional[str] = None,
-        session_id: Optional[str] = None,
-        source: Optional[str] = None,
-        limit: int = 100,
-        offset: int = 0,
-        order: str = "desc",
-    ) -> Dict[str, Any]:
-        from .models.memory import MemoryLayer, MemoryStatus
-
-        if not user_id:
-            raise ValueError("user_id is required")
-        if limit < 1 or offset < 0:
-            raise ValueError("limit must be positive and offset must be non-negative")
-        nodes = await self._vector_store.list_by_user(
-            user_id=user_id,
-            agent_id=agent_id,
-            status_filter=[MemoryStatus.ACTIVE],
-            layers=[MemoryLayer.L1_RAW],
-            include_sources=[source] if source else None,
-            limit=100_000,
-            only_latest=False,
-            with_vectors=False,
-            allow_raw_layer=True,
-        )
-        if session_id:
-            nodes = [node for node in nodes if node.session_id == session_id]
-        nodes = self._sort_memory_nodes(nodes, order=order)
-        total = len(nodes)
-        return {
-            "raw": [
-                self._raw_node_to_public(node)
-                for node in nodes[offset: offset + limit]
-            ],
-            "total": total,
-            "limit": limit,
-            "offset": offset,
-        }
-
     async def _search_explicit_raw_layer(
         self,
         query: str,
@@ -4500,7 +4095,6 @@ class HyMemoryClient:
         self,
         memory_id: str,
         *,
-        include_raw: bool = False,
         kind: Optional[MemoryKind] = None,
         variant: Optional[MemoryVariant] = None,
     ) -> Optional[Dict[str, Any]]:
@@ -4513,15 +4107,12 @@ class HyMemoryClient:
 
         Args:
             memory_id: 记忆 ID
-            include_raw: 是否允许返回 L1_RAW 原始记忆，默认 False。
             kind: 可选 namespace（"chat" / "experience" / "fact"）。
             variant: 可选生产力变体（"working" / "coding"）。
 
         Returns:
             统一 memory dict，或 None。
         """
-        from .models.memory import MemoryLayer
-
         if kind not in (None, "chat", "experience", "fact"):
             raise ValueError(
                 "kind must be one of: chat, experience, fact"
@@ -4583,26 +4174,6 @@ class HyMemoryClient:
         vdb_node = by_name.get("chat:vdb")
         graph_node = by_name.get("chat:graph")
 
-        # L1_RAW 默认不对外可见：按 ID 直取走主键查询、绕过向量库的 layer
-        # filter，故在此统一挡掉。SDK 调用方可通过 include_raw=True 显式读取；
-        # HTTP 接口不传该参数，继续保持隐藏。内部写链路不经过此方法。
-        #
-        # 置于 gather 之后、埋点之前：命中 raw 仍照常落 GET 埋点（保留可观测性，
-        # 便于发现有调用方拿 raw id 试探），只是不把内容返回给调用方。
-        # 对调用方的表现与"该 id 不存在"完全一致，不额外暴露"存在但被挡"的信息。
-        if (
-            not include_raw
-            and vdb_node is not None
-            and vdb_node.layer == MemoryLayer.L1_RAW
-        ):
-            vdb_node = None
-        if (
-            not include_raw
-            and graph_node is not None
-            and getattr(graph_node, "layer", None) == MemoryLayer.L1_RAW
-        ):
-            graph_node = None
-
         matches = []
         chat_node = vdb_node if vdb_node is not None else graph_node
         if chat_node is not None:
@@ -4682,7 +4253,8 @@ class HyMemoryClient:
                 "confidence": vdb_node.confidence,
                 "custom": vdb_node.custom or {},
                 "meta": list((vdb_node.custom or {}).get("meta") or []),
-                "source": "vdb",
+                "source": vdb_node.source,
+                "retrieval_source": "vdb",
                 "request_id": request_id,
                 "elapsed_ms": elapsed,
             }
@@ -4709,7 +4281,8 @@ class HyMemoryClient:
                 "owner": getattr(graph_node, "owner", None),
                 "confidence": graph_node.confidence,
                 "custom": getattr(graph_node, "custom", {}) or {},
-                "source": "graph",
+                "source": getattr(graph_node, "source", "") or "",
+                "retrieval_source": "graph",
                 "request_id": request_id,
                 "elapsed_ms": elapsed,
             }
@@ -5282,6 +4855,7 @@ class HyMemoryClient:
             "user_id": node.user_id,
             "agent_id": node.agent_id,
             "session_id": getattr(node, "session_id", None),
+            "source": getattr(node, "source", None),
             "speculate": getattr(node, "speculate", None),
             # 不透出 source_raw_memory_id：见 async_search 同处说明。
             "tags": node.tags or [],
@@ -5353,6 +4927,9 @@ class HyMemoryClient:
         order: str,
         user_ids: Optional[List[str]] = None,
         agent_ids: Optional[List[str]] = None,
+        session_ids: Optional[List[str]] = None,
+        include_sources: Optional[List[str]] = None,
+        exclude_sources: Optional[List[str]] = None,
         layers: Optional[List] = None,
         statuses: Optional[List] = None,
         with_evolution: bool = False,
@@ -5378,6 +4955,9 @@ class HyMemoryClient:
             status_filter=effective_statuses,
             layers=layers,
             with_vectors=False,
+            allow_raw_layer=bool(
+                layers is not None and MemoryLayer.L1_RAW in layers
+            ),
         )
 
         # L0 是 user-level，不应被 agent 过滤。默认列表或显式包含 L0 时，额外按
@@ -5399,6 +4979,22 @@ class HyMemoryClient:
             ])
             all_nodes.extend(hit["node"] for hit in l0_hits if hit.get("node") is not None)
 
+        if session_ids:
+            wanted_sessions = set(session_ids)
+            all_nodes = [
+                node for node in all_nodes if node.session_id in wanted_sessions
+            ]
+        if include_sources:
+            wanted_sources = set(include_sources)
+            all_nodes = [
+                node for node in all_nodes if node.source in wanted_sources
+            ]
+        if exclude_sources:
+            blocked_sources = set(exclude_sources)
+            all_nodes = [
+                node for node in all_nodes if node.source not in blocked_sources
+            ]
+
         all_nodes = self._sort_memory_nodes(all_nodes, order=order)
         total = len(all_nodes)
         page_nodes = all_nodes[offset: offset + limit]
@@ -5462,6 +5058,9 @@ class HyMemoryClient:
         order: str,
         user_ids: Optional[List[str]] = None,
         agent_ids: Optional[List[str]] = None,
+        session_ids: Optional[List[str]] = None,
+        include_sources: Optional[List[str]] = None,
+        exclude_sources: Optional[List[str]] = None,
         layers: Optional[List] = None,
         statuses: Optional[List] = None,
     ) -> Optional[Dict[str, Any]]:
@@ -5509,6 +5108,21 @@ class HyMemoryClient:
                 node for node in graph_nodes
                 if getattr(node, "status", None) in effective_statuses
             ]
+            if session_ids:
+                wanted_sessions = set(session_ids)
+                graph_nodes = [
+                    node for node in graph_nodes if node.session_id in wanted_sessions
+                ]
+            if include_sources:
+                wanted_sources = set(include_sources)
+                graph_nodes = [
+                    node for node in graph_nodes if node.source in wanted_sources
+                ]
+            if exclude_sources:
+                blocked_sources = set(exclude_sources)
+                graph_nodes = [
+                    node for node in graph_nodes if node.source not in blocked_sources
+                ]
 
             graph_nodes = self._sort_memory_nodes(graph_nodes, order=order)
             total = len(graph_nodes)
@@ -5620,6 +5234,9 @@ class HyMemoryClient:
         branch: Optional[str] = None,
         user_ids: Optional[List[str]] = None,
         agent_ids: Optional[List[str]] = None,
+        session_ids: Optional[List[str]] = None,
+        include_sources: Optional[List[str]] = None,
+        exclude_sources: Optional[List[str]] = None,
         layers: Optional[List[Any]] = None,
         statuses: Optional[List[Any]] = None,
         with_evolution: bool = False,
@@ -5696,6 +5313,9 @@ class HyMemoryClient:
         vdb = await self._list_vdb_bucket(
             user_id=user_id, agent_id=agent_id,
             user_ids=user_ids, agent_ids=agent_ids,
+            session_ids=session_ids,
+            include_sources=include_sources,
+            exclude_sources=exclude_sources,
             limit=limit, offset=offset, order=order,
             layers=effective_layers, statuses=effective_statuses,
             with_evolution=with_evolution,
@@ -5703,6 +5323,9 @@ class HyMemoryClient:
         graph = await self._list_graph_bucket(
             user_id=user_id, agent_id=agent_id,
             user_ids=user_ids, agent_ids=agent_ids,
+            session_ids=session_ids,
+            include_sources=include_sources,
+            exclude_sources=exclude_sources,
             limit=limit, offset=offset, order=order,
             layers=effective_layers, statuses=effective_statuses,
         )

memhub/MEMORY.md

@@ -9,7 +9,9 @@
 ## Architecture Principles
 
 1. Memory 读写与存储 backend **只经 SDK**（config / env）。
-2. MemHub 对 memory 的原子操作 **只经 SDK**；缺接口先补 SDK。
+2. MemHub 运行时对 memory 的原子操作 **只经 SDK 公共 Client**；缺正常业务
+   接口先补 SDK。Runtime writers 停止后的 App 启动迁移可直接操作底层存储，
+   但不得把迁移专用方法加入公共 Client。
 3. MemHub **不存 memory 实体**；**可以存 `memory_id`** 做组织/回执。原文进入 **SDK raw**。
 
 生产入口通过共享 Runtime 租约取得公开 `HyMemoryClient`；MemHub 只在调用前执行

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

@@ -3240,13 +3240,14 @@ def create_app(
         try:
             client = await _sdk_client_handle(app).client()
             value = await asyncio.to_thread(
-                client.list_raw,
+                client.list_memories,
                 user_id="local",
-                source=source,
+                layers=["l1_raw"],
+                include_sources=[source] if source else None,
                 offset=offset,
                 limit=limit,
             )
-            page = HySdkRawPage.from_sdk(value, limit=limit, offset=offset)
+            page = HySdkRawPage.from_memory_list(value, limit=limit, offset=offset)
         except Exception as error:
             raise _sdk_http_error(error, "hy_memory_raw_list_failed") from None
         return SdkRawPage(items=page.raw, total=page.total, offset=offset, limit=limit)

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

@@ -431,7 +431,12 @@ def sdk_migrate(options: Sequence[str], stream: TextIO) -> int:
         MigrationBundleRejected,
         inspect_sdk_migration_bundle,
     )
-    from one_memory.sdk_migration_runner import BackfillResult, run_sdk_data_plane_backfill
+    from one_memory.sdk_migration_runner import (
+        BackfillResult,
+        prepare_sdk_migration_staging,
+        run_sdk_data_plane_backfill,
+        sdk_migration_staging,
+    )
     from one_memory.secrets import KeyringSecretStore, SecretStoreUnavailableError
 
     database = settings.data_dir / "one-memory.sqlite3"
@@ -482,13 +487,27 @@ def sdk_migrate(options: Sequence[str], stream: TextIO) -> int:
             separators=(",", ":"),
         )
         backend_fingerprint = hashlib.sha256(backend_material.encode()).hexdigest()
+        staging = sdk_migration_staging(
+            data_dir=settings.data_dir,
+            migration_id=migration_id,
+            namespace=HY_MEMORY_NAMESPACE,
+        )
+        try:
+            staging_data_dir = prepare_sdk_migration_staging(
+                paths=staging,
+                database_path=database,
+            )
+        except Exception:
+            code = "sdk_migration_staging_failed"
+            staging_data_dir = None
 
         async def run_backfill() -> BackfillResult:
+            assert staging_data_dir is not None
             lease = await asyncio.to_thread(
                 acquire_client,
                 profile,
                 api_key=secret,
-                data_dir=settings.data_dir,
+                data_dir=staging_data_dir,
             )
             try:
                 return await run_sdk_data_plane_backfill(
@@ -506,10 +525,18 @@ def sdk_migrate(options: Sequence[str], stream: TextIO) -> int:
                 await asyncio.to_thread(lease.close)
 
         try:
-            result = asyncio.run(run_backfill())
+            result = (
+                asyncio.run(run_backfill())
+                if staging_data_dir is not None
+                else None
+            )
         except Exception:
             code = "sdk_migration_backfill_failed"
         else:
+            if result is None:
+                document = {"schema_version": 1, "state": "failed", "code": code}
+                _write_restore(stream, as_json, document, f"SDK backfill 失败：{code}")
+                return 1
             document = {
                 "schema_version": 1,
                 "state": result.state,
@@ -585,20 +612,39 @@ def sdk_cutover(options: Sequence[str], stream: TextIO) -> int:
         _write_restore(stream, as_json, document, "预览通过；加 --yes 后提交 SDK authority。")
         return 4
 
-    from one_memory.hy_runtime_config import HY_MEMORY_VERSION
+    from one_memory.hy_runtime_config import HY_MEMORY_NAMESPACE, HY_MEMORY_VERSION
     from one_memory.sdk_data_plane_migration import DataPlaneMigrationError
     from one_memory.sdk_migration_bundle import MigrationBundleRejected
-    from one_memory.sdk_migration_runner import commit_reconciled_cutover
+    from one_memory.sdk_migration_runner import (
+        activate_sdk_migration_staging,
+        commit_reconciled_cutover,
+        resume_legacy_cleanup,
+        sdk_migration_staging,
+    )
 
     try:
-        marker = commit_reconciled_cutover(
+        database_path = settings.data_dir / "one-memory.sqlite3"
+        staging = sdk_migration_staging(
             data_dir=settings.data_dir,
-            database_path=settings.data_dir / "one-memory.sqlite3",
+            migration_id=migration_id,
+            namespace=HY_MEMORY_NAMESPACE,
+        )
+        marker = activate_sdk_migration_staging(
+            paths=staging,
+            database_path=database_path,
+            commit=lambda: commit_reconciled_cutover(
+                data_dir=settings.data_dir,
+                database_path=database_path,
                 migration_id=migration_id,
                 bundle_id=bundle_id,
                 sdk_version=HY_MEMORY_VERSION,
+            ),
         )
-    except (DataPlaneMigrationError, MigrationBundleRejected) as error:
+        resume_legacy_cleanup(
+            data_dir=settings.data_dir,
+            database_path=database_path,
+        )
+    except (DataPlaneMigrationError, MigrationBundleRejected, OSError) as error:
         code = getattr(error, "code", str(error))
         document = {
             "schema_version": 1,
@@ -640,10 +686,12 @@ def sdk_migration_status(options: Sequence[str], stream: TextIO) -> int:
         import sqlite3
 
         from one_memory.database_access import main_database
+        from one_memory.hy_runtime_config import HY_MEMORY_NAMESPACE
         from one_memory.sdk_data_plane_migration import (
             MemoryAuthority,
             SdkDataPlaneMigrationStore,
         )
+        from one_memory.sdk_migration_runner import sdk_migration_staging
         from one_memory.target_schema import legacy_tables_present
 
         migration_store = SdkDataPlaneMigrationStore(database_path)
@@ -653,10 +701,25 @@ def sdk_migration_status(options: Sequence[str], stream: TextIO) -> int:
         generation = hashlib.sha256(f"{stat.st_size}:{stat.st_mtime_ns}".encode()).hexdigest()[:16]
         if authority is MemoryAuthority.SDK:
             legacy_tables = legacy_tables_present(database_path)
+            staging_cleanup = False
+            if marker is not None:
+                staging = sdk_migration_staging(
+                    data_dir=settings.data_dir,
+                    migration_id=marker.migration_id,
+                    namespace=HY_MEMORY_NAMESPACE,
+                )
+                staging_cleanup = (
+                    staging.protective_root.exists()
+                    or staging.staging_home.exists()
+                )
             document = {
                 "schema_version": 1,
                 "authority": "sdk",
-                "state": "cleanup_required" if legacy_tables else "complete",
+                "state": (
+                    "cleanup_required"
+                    if legacy_tables or staging_cleanup
+                    else "complete"
+                ),
                 "migration_id": marker.migration_id if marker else None,
                 "generation": generation,
                 "legacy_tables": list(legacy_tables),

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

@@ -1,4 +1,10 @@
-"""Migration-only access to the HY SDK's public inventory and import APIs."""
+"""Privileged offline access to the HY SDK data plane during App migration.
+
+Runtime product code must use public ``HyMemoryClient`` methods.  This module
+is the deliberately isolated exception: App startup has stopped all writers,
+so the one-time legacy migration may access SDK stores directly without
+turning migration mechanics into public Client APIs.
+"""
 
 from __future__ import annotations
 
@@ -6,7 +12,7 @@ import asyncio
 from collections.abc import Callable, Mapping
 from datetime import datetime
 from pathlib import Path
-from typing import Protocol
+from typing import Any, Protocol
 
 from hy_memory import (  # type: ignore[import-untyped]
     DataPlaneRestore,
@@ -14,6 +20,13 @@ from hy_memory import (  # type: ignore[import-untyped]
     restore_local_data_plane,
     snapshot_local_data_plane,
 )
+from hy_memory.models.memory import (  # type: ignore[import-untyped]
+    MemoryLayer,
+    MemoryNode,
+    MemoryStatus,
+    SourceType,
+)
+from hy_memory.utils.source_filter import normalize_source  # type: ignore[import-untyped]
 
 from one_memory.hy_sdk_types import (
     HySdkMemoryPage,
@@ -25,6 +38,11 @@ from one_memory.hy_sdk_types import (
 
 
 class MigrationClient(Protocol):
+    _loop_thread: Any
+    _vector_store: Any
+    _embed_service: Any
+    _graph_store: Any
+
     def list_memories(
         self,
         *,
@@ -36,32 +54,20 @@ class MigrationClient(Protocol):
         order: str = "desc",
     ) -> Mapping[str, object]: ...
 
-    def import_memory_exact(self, content: str, **kwargs: object) -> Mapping[str, object]: ...
-
-    def list_raw(
-        self,
-        *,
-        user_id: str,
-        agent_id: str | None = None,
-        session_id: str | None = None,
-        source: str | None = None,
-        limit: int = 100,
-        offset: int = 0,
-        order: str = "desc",
-    ) -> Mapping[str, object]: ...
-
-    def import_raw(self, content: str, **kwargs: object) -> Mapping[str, object]: ...
-
     def delete(self, memory_id: str) -> Mapping[str, object]: ...
 
     def get_write_outcome(self, request_id: str) -> Mapping[str, object] | None: ...
 
 
 class HySdkMigration:
-    """Typed async adapter over the SDK's exact migration primitives."""
+    """Typed offline adapter over one initialized SDK data plane."""
 
     def __init__(self, client: MigrationClient) -> None:
         self.client = client
+        self._loop_thread = client._loop_thread
+        self._vector_store = client._vector_store
+        self._embed_service = client._embed_service
+        self._graph_store = client._graph_store
 
     async def list_memories(
         self,
@@ -85,7 +91,7 @@ class HySdkMigration:
         )
         return _memory_page(raw, limit=limit, offset=offset)
 
-    async def import_memory_exact(
+    async def write_legacy_memory(
         self,
         *,
         content: str,
@@ -107,8 +113,9 @@ class HySdkMigration:
     ) -> dict[str, object]:
         raw = await _sdk_call(
             "hy_sdk_memory_import_failed",
-            self.client.import_memory_exact,
+            self._loop_thread.run,
-            content,
+            self._write_legacy_memory(
+                content=content,
                 memory_id=memory_id,
                 user_id=user_id,
                 agent_id=agent_id,
@@ -117,17 +124,108 @@ class HySdkMigration:
                 layer=layer,
                 status=status,
                 owner=owner,
+                tags=tags,
+                supersedes=supersedes,
+                superseded_by=superseded_by,
+                is_latest=is_latest,
+                memory_at=memory_at,
+                metadata=metadata,
+                request_id=request_id,
+            ),
+        )
+        return _successful_result(raw, "hy_sdk_memory_import_failed")
+
+    async def _write_legacy_memory(
+        self,
+        *,
+        content: str,
+        memory_id: str,
+        user_id: str,
+        agent_id: str,
+        session_id: str,
+        source: str,
+        layer: str,
+        status: str,
+        owner: str | None,
+        tags: tuple[str, ...],
+        supersedes: tuple[str, ...],
+        superseded_by: tuple[str, ...],
+        is_latest: bool,
+        memory_at: datetime | None,
+        metadata: Mapping[str, object],
+        request_id: str,
+    ) -> dict[str, object]:
+        if not content or not memory_id or not user_id:
+            raise ValueError("content, memory_id and user_id are required")
+        normalized_layer = MemoryLayer.from_string(layer)
+        if normalized_layer is MemoryLayer.L1_RAW:
+            raise ValueError("raw content must use the raw migration lane")
+        normalized_status = MemoryStatus(status.lower())
+        normalized_source = normalize_source(source)
+        existing = await self._vector_store.get_by_id(memory_id)
+        if existing is not None:
+            identity = (
+                existing.content == content,
+                existing.user_id == user_id,
+                existing.agent_id == agent_id,
+                existing.session_id == session_id,
+                existing.source == normalized_source,
+                existing.layer == normalized_layer,
+                existing.status == normalized_status,
+                existing.owner == owner,
+                list(existing.tags or []) == list(tags),
+                list(existing.supersedes or []) == list(supersedes),
+                list(existing.superseded_by or []) == list(superseded_by),
+                existing.is_latest == is_latest,
+            )
+            if not all(identity):
+                raise ValueError("memory_id is already bound to different memory content")
+            await self._project_graph(existing)
+            return {
+                "success": True,
+                "memory_id": memory_id,
+                "memory_ids": [memory_id],
+                "request_id": request_id,
+                "replayed": True,
+            }
+        custom = dict(metadata)
+        custom["migration_request_id"] = request_id
+        embedding = await self._embed_service.embed(content)
+        node = MemoryNode(
+            node_id=memory_id,
+            user_id=user_id,
+            agent_id=agent_id,
+            session_id=session_id,
+            source=normalized_source,
+            layer=normalized_layer,
+            content=content,
+            source_type=SourceType.EXPLICIT,
+            status=normalized_status,
+            owner=owner,
             tags=list(tags),
             supersedes=list(supersedes),
             superseded_by=list(superseded_by),
             is_latest=is_latest,
+            embedding=embedding,
             memory_at=memory_at,
-            metadata=dict(metadata),
+            custom=custom,
-            request_id=request_id,
         )
-        return _successful_result(raw, "hy_sdk_memory_import_failed")
+        stored_id = await self._vector_store.upsert(node)
+        if stored_id != memory_id:
+            raise RuntimeError("exact migration changed the requested memory identity")
+        await self._project_graph(node)
+        return {
+            "success": True,
+            "memory_id": memory_id,
+            "memory_ids": [memory_id],
+            "request_id": request_id,
+        }
 
-    async def list_raw(
+    async def _project_graph(self, node: MemoryNode) -> None:
+        if self._graph_store is not None:
+            await self._graph_store.upsert_memory_node(node)
+
+    async def read_raw_inventory(
         self,
         *,
         user_id: str,
@@ -140,7 +238,8 @@ class HySdkMigration:
     ) -> HySdkRawPage:
         raw = await _sdk_call(
             "hy_sdk_raw_list_failed",
-            self.client.list_raw,
+            self._loop_thread.run,
+            self._read_raw_inventory(
                 user_id=user_id,
                 agent_id=agent_id,
                 session_id=session_id,
@@ -148,10 +247,48 @@ class HySdkMigration:
                 limit=limit,
                 offset=offset,
                 order=order,
+            ),
         )
         return _raw_page(raw, limit=limit, offset=offset)
 
-    async def import_raw(
+    async def _read_raw_inventory(
+        self,
+        *,
+        user_id: str,
+        agent_id: str | None,
+        session_id: str | None,
+        source: str | None,
+        limit: int,
+        offset: int,
+        order: str,
+    ) -> dict[str, object]:
+        if not user_id:
+            raise ValueError("user_id is required")
+        nodes = await self._vector_store.list_by_user(
+            user_id=user_id,
+            agent_id=agent_id,
+            status_filter=[MemoryStatus.ACTIVE],
+            layers=[MemoryLayer.L1_RAW],
+            include_sources=[source] if source else None,
+            limit=100_000,
+            only_latest=False,
+            with_vectors=False,
+            allow_raw_layer=True,
+        )
+        if session_id:
+            nodes = [node for node in nodes if node.session_id == session_id]
+        nodes.sort(
+            key=_node_timestamp,
+            reverse=order.lower() != "asc",
+        )
+        return {
+            "raw": [_raw_node(node) for node in nodes[offset : offset + limit]],
+            "total": len(nodes),
+            "limit": limit,
+            "offset": offset,
+        }
+
+    async def write_legacy_raw(
         self,
         *,
         content: str,
@@ -166,19 +303,78 @@ class HySdkMigration:
     ) -> dict[str, object]:
         raw = await _sdk_call(
             "hy_sdk_raw_import_failed",
-            self.client.import_raw,
+            self._loop_thread.run,
-            content,
+            self._write_legacy_raw(
+                content=content,
                 raw_id=raw_id,
                 user_id=user_id,
                 agent_id=agent_id,
                 session_id=session_id,
                 source=source,
                 memory_at=memory_at,
-            metadata=dict(metadata),
+                metadata=metadata,
                 request_id=request_id,
+            ),
         )
         return _successful_result(raw, "hy_sdk_raw_import_failed")
 
+    async def _write_legacy_raw(
+        self,
+        *,
+        content: str,
+        raw_id: str,
+        user_id: str,
+        agent_id: str,
+        session_id: str,
+        source: str,
+        memory_at: datetime | None,
+        metadata: Mapping[str, object],
+        request_id: str,
+    ) -> dict[str, object]:
+        if not content or not raw_id or not user_id:
+            raise ValueError("content, raw_id and user_id are required")
+        normalized_source = normalize_source(source)
+        existing = await self._vector_store.get_by_id(raw_id)
+        if existing is not None:
+            identity = (
+                existing.layer is MemoryLayer.L1_RAW,
+                existing.content == content,
+                existing.user_id == user_id,
+                existing.agent_id == agent_id,
+                existing.session_id == session_id,
+                existing.source == normalized_source,
+            )
+            if not all(identity):
+                raise ValueError("raw_id is already bound to different source content")
+            return {
+                "success": True,
+                "raw_id": raw_id,
+                "request_id": request_id,
+                "replayed": True,
+            }
+        custom = dict(metadata)
+        custom["migration_request_id"] = request_id
+        embedding = await self._embed_service.embed(content)
+        node = MemoryNode(
+            node_id=raw_id,
+            user_id=user_id,
+            agent_id=agent_id,
+            session_id=session_id,
+            source=normalized_source,
+            layer=MemoryLayer.L1_RAW,
+            content=content,
+            source_type=SourceType.EXPLICIT,
+            status=MemoryStatus.ACTIVE,
+            is_latest=True,
+            embedding=embedding,
+            memory_at=memory_at,
+            custom=custom,
+        )
+        stored_id = await self._vector_store.upsert(node)
+        if stored_id != raw_id:
+            raise RuntimeError("exact migration changed the requested raw identity")
+        return {"success": True, "raw_id": raw_id, "request_id": request_id}
+
     async def delete(self, memory_id: str) -> dict[str, object]:
         raw = await _sdk_call(
             "hy_sdk_memory_delete_failed",
@@ -338,6 +534,28 @@ def _raw_page(raw: object, *, limit: int, offset: int) -> HySdkRawPage:
     )
 
 
+def _node_timestamp(node: Any) -> float:
+    value = getattr(node, "memory_at", None) or getattr(node, "gmt_created", None)
+    return value.timestamp() if isinstance(value, datetime) else 0.0
+
+
+def _raw_node(node: Any) -> dict[str, object]:
+    def timestamp(value: object) -> int | None:
+        return int(value.timestamp()) if isinstance(value, datetime) else None
+
+    return {
+        "raw_id": node.node_id,
+        "content": node.content,
+        "user_id": node.user_id,
+        "agent_id": node.agent_id,
+        "session_id": node.session_id,
+        "source": node.source,
+        "memory_at": timestamp(node.memory_at),
+        "gmt_created": timestamp(node.gmt_created),
+        "custom": node.custom or {},
+    }
+
+
 def _raw_record(raw: object) -> HySdkRawRecord:
     if not isinstance(raw, Mapping):
         raise HySdkMigrationError("hy_sdk_raw_record_shape_invalid")

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

@@ -22,7 +22,7 @@ class HySdkRuntimeError(HySdkError):
 
 
 class HySdkMigrationError(HySdkError):
-    """A public SDK migration operation failed or returned an invalid shape."""
+    """The privileged offline SDK data-plane migration failed."""
 
 
 class HySdkReadError(HySdkError):
@@ -200,11 +200,21 @@ class HySdkRawPage(BaseModel):
     offset: int = Field(ge=0)
 
     @classmethod
-    def from_sdk(cls, raw: object, *, limit: int, offset: int) -> HySdkRawPage:
+    def from_memory_list(
+        cls,
+        raw: object,
+        *,
+        limit: int,
+        offset: int,
+    ) -> HySdkRawPage:
+        """Parse ``list_memories(layers=['l1_raw'])`` into the App raw DTO."""
         if not isinstance(raw, Mapping):
             raise HySdkReadError("hy_memory_raw_list_shape_unknown")
-        rows = raw.get("raw")
+        bucket = raw.get("vdb")
-        total = raw.get("total")
+        if not isinstance(bucket, Mapping):
+            raise HySdkReadError("hy_memory_raw_list_shape_unknown")
+        rows = bucket.get("memories")
+        total = bucket.get("total")
         if (
             not isinstance(rows, list)
             or not isinstance(total, int)
@@ -212,8 +222,13 @@ class HySdkRawPage(BaseModel):
             or total < 0
         ):
             raise HySdkReadError("hy_memory_raw_list_shape_unknown")
+        converted = []
+        for row in rows:
+            if not isinstance(row, Mapping) or row.get("layer") != "l1_raw":
+                raise HySdkReadError("hy_memory_raw_list_shape_unknown")
+            converted.append({**row, "raw_id": row.get("memory_id")})
         return cls(
-            raw=tuple(HySdkRawRecord.from_sdk(row) for row in rows),
+            raw=tuple(HySdkRawRecord.from_sdk(row) for row in converted),
             total=total,
             limit=limit,
             offset=offset,

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

@@ -115,7 +115,7 @@ class SdkMigrationClient(Protocol):
         order: str = "desc",
     ) -> Awaitable[HySdkMemoryPage]: ...
 
-    def import_memory_exact(
+    def write_legacy_memory(
         self,
         *,
         content: str,
@@ -133,7 +133,7 @@ class SdkMigrationClient(Protocol):
         request_id: str,
     ) -> Awaitable[dict[str, object]]: ...
 
-    def list_raw(
+    def read_raw_inventory(
         self,
         *,
         user_id: str,
@@ -145,7 +145,7 @@ class SdkMigrationClient(Protocol):
         order: str = "desc",
     ) -> Awaitable[HySdkRawPage]: ...
 
-    def import_raw(
+    def write_legacy_raw(
         self,
         *,
         content: str,
@@ -472,7 +472,7 @@ async def execute_memory_migration_plan(
             request_id = hashlib.sha256(
                 f"{migration_id}:{decision.item_key}:import_exact".encode()
             ).hexdigest()
-            await client.import_memory_exact(
+            await client.write_legacy_memory(
                 content=item.content,
                 memory_id=item.canonical_memory_id,
                 user_id=item.user_id,
@@ -514,7 +514,7 @@ async def read_sdk_raw_inventory(
     records: dict[str, SdkRawInventory] = {}
     offset = 0
     while True:
-        page = await client.list_raw(
+        page = await client.read_raw_inventory(
             user_id=user_id,
             limit=page_size,
             offset=offset,
@@ -657,9 +657,9 @@ async def execute_raw_migration_plan(
             raise RuntimeError(f"missing legacy raw inventory for {decision.item_key}")
         if decision.action == "import_exact":
             request_id = hashlib.sha256(
-                f"{migration_id}:{decision.item_key}:import_raw".encode()
+                f"{migration_id}:{decision.item_key}:write_legacy_raw".encode()
             ).hexdigest()
-            await client.import_raw(
+            await client.write_legacy_raw(
                 content=item.content,
                 raw_id=decision.raw_id,
                 user_id=item.user_id,

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

@@ -4,7 +4,10 @@ from __future__ import annotations
 
 import hashlib
 import json
+import shutil
 import sqlite3
+from collections.abc import Callable
+from contextlib import suppress
 from dataclasses import dataclass
 from pathlib import Path
 
@@ -47,6 +50,139 @@ class BackfillResult:
     raw_items: int
 
 
+@dataclass(frozen=True)
+class SdkMigrationStaging:
+    migration_id: str
+    staging_home: Path
+    live_root: Path
+    staging_root: Path
+    protective_root: Path
+    state_file: Path
+
+
+def sdk_migration_staging(
+    *,
+    data_dir: Path,
+    migration_id: str,
+    namespace: str,
+) -> SdkMigrationStaging:
+    """Return same-filesystem paths without trusting the migration ID as a path."""
+    token = hashlib.sha256(migration_id.encode()).hexdigest()[:16]
+    home = data_dir.resolve()
+    staging_home = home / ".sdk-migration" / token
+    live_root = home / "hy-memory" / namespace
+    return SdkMigrationStaging(
+        migration_id=migration_id,
+        staging_home=staging_home,
+        live_root=live_root,
+        staging_root=staging_home / "hy-memory" / namespace,
+        protective_root=live_root.parent / f".{namespace}.pre-migration-{token}",
+        state_file=staging_home / "swap-state.json",
+    )
+
+
+def _write_staging_state(paths: SdkMigrationStaging, phase: str) -> None:
+    paths.staging_home.mkdir(parents=True, exist_ok=True)
+    temporary = paths.state_file.with_suffix(".tmp")
+    temporary.write_text(
+        json.dumps(
+            {"schema_version": 1, "migration_id": paths.migration_id, "phase": phase},
+            sort_keys=True,
+        ),
+        encoding="utf-8",
+    )
+    temporary.replace(paths.state_file)
+
+
+def recover_sdk_migration_swap(
+    *,
+    paths: SdkMigrationStaging,
+    database_path: Path,
+) -> None:
+    """Recover an interrupted directory swap before opening any SDK store."""
+    if not paths.protective_root.exists():
+        return
+    authority = SdkDataPlaneMigrationStore(database_path).authority()
+    if authority.value == "sdk":
+        if not paths.live_root.is_dir():
+            raise DataPlaneMigrationError("sdk_migration_committed_store_missing")
+        shutil.rmtree(paths.protective_root)
+        shutil.rmtree(paths.staging_home, ignore_errors=True)
+        return
+    if paths.live_root.exists():
+        if paths.staging_root.exists():
+            raise DataPlaneMigrationError("sdk_migration_swap_state_ambiguous")
+        paths.staging_root.parent.mkdir(parents=True, exist_ok=True)
+        paths.live_root.replace(paths.staging_root)
+    paths.protective_root.replace(paths.live_root)
+    _write_staging_state(paths, "recovered")
+
+
+def prepare_sdk_migration_staging(
+    *,
+    paths: SdkMigrationStaging,
+    database_path: Path,
+) -> Path:
+    """Build a private copy; incomplete retries restart from the unchanged live root."""
+    recover_sdk_migration_swap(paths=paths, database_path=database_path)
+    migration_store = SdkDataPlaneMigrationStore(database_path)
+    try:
+        progress = migration_store.progress(paths.migration_id)
+        reuse_ready = progress.state == "ready" and progress.phase == "reconciled"
+    except Exception:
+        reuse_ready = False
+    if paths.staging_root.exists():
+        if paths.staging_root.is_symlink() or not paths.staging_root.is_dir():
+            raise DataPlaneMigrationError("sdk_migration_staging_invalid")
+        if reuse_ready:
+            return paths.staging_home
+        shutil.rmtree(paths.staging_root)
+    paths.live_root.mkdir(parents=True, exist_ok=True)
+    from one_memory.hy_sdk_migration import snapshot_data_plane
+
+    snapshot_data_plane(
+        source_root=paths.live_root,
+        target_root=paths.staging_root,
+        writers_stopped=True,
+    )
+    _write_staging_state(paths, "prepared")
+    return paths.staging_home
+
+
+def activate_sdk_migration_staging(
+    *,
+    paths: SdkMigrationStaging,
+    database_path: Path,
+    commit: Callable[[], CutoverMarker],
+) -> CutoverMarker:
+    """Atomically expose staged SDK files and roll them back if cutover fails."""
+    recover_sdk_migration_swap(paths=paths, database_path=database_path)
+    if not paths.staging_root.is_dir():
+        raise DataPlaneMigrationError("sdk_migration_staging_missing")
+    paths.live_root.mkdir(parents=True, exist_ok=True)
+    _write_staging_state(paths, "activating")
+    paths.live_root.replace(paths.protective_root)
+    try:
+        paths.staging_root.replace(paths.live_root)
+        _write_staging_state(paths, "staging_live")
+        marker = commit()
+    except BaseException:
+        if paths.live_root.exists():
+            paths.staging_root.parent.mkdir(parents=True, exist_ok=True)
+            paths.live_root.replace(paths.staging_root)
+        if paths.protective_root.exists():
+            paths.protective_root.replace(paths.live_root)
+        _write_staging_state(paths, "rolled_back")
+        raise
+    with suppress(OSError):
+        _write_staging_state(paths, "committed")
+    # Authority is already committed; leftover paths are detected by
+    # sdk-migration-status and removed by sdk-cleanup on the next startup.
+    shutil.rmtree(paths.protective_root, ignore_errors=True)
+    shutil.rmtree(paths.staging_home, ignore_errors=True)
+    return marker
+
+
 def prepare_legacy_write_queue_for_cutover(database_path: Path) -> tuple[int, int]:
     """Freeze old jobs and move unstarted task rows into the surviving queue."""
     with main_database(database_path).read(row_factory=sqlite3.Row) as database:
@@ -217,7 +353,6 @@ def commit_reconciled_cutover(
         reconciliation_digest=digest,
     )
     inspect_sdk_migration_bundle(data_dir=data_dir, bundle_id=marker.backup_bundle_id)
-    drop_legacy_tables(database_path)
     return marker
 
 
@@ -232,6 +367,16 @@ def resume_legacy_cleanup(*, data_dir: Path, database_path: Path) -> tuple[str,
             data_dir=data_dir,
             bundle_id=marker.backup_bundle_id,
         )
+        from one_memory.hy_runtime_config import HY_MEMORY_NAMESPACE
+
+        recover_sdk_migration_swap(
+            paths=sdk_migration_staging(
+                data_dir=data_dir,
+                migration_id=marker.migration_id,
+                namespace=HY_MEMORY_NAMESPACE,
+            ),
+            database_path=database_path,
+        )
     return drop_legacy_tables(database_path)
 
 

memhub/apps/core/tests/test_architecture_layer_gate.py

@@ -107,6 +107,12 @@ SDK_PRIVATE_ACCESS_EXCEPTIONS = {
     f"{PACKAGE_LABEL}/hy_graph_projection.py": frozenset(
         {"_graph_store", "_loop_thread", "_run"}
     ),
+    # Startup migration runs while Runtime writers are stopped. It is the one
+    # non-runtime path allowed to manipulate SDK storage directly so that
+    # migration-only mechanics never become public HyMemoryClient methods.
+    f"{PACKAGE_LABEL}/hy_sdk_migration.py": frozenset(
+        {"_embed_service", "_graph_store", "_loop_thread", "_vector_store"}
+    ),
 }
 
 
@@ -241,7 +247,7 @@ def test_no_new_module_writes_legacy_memory_authority() -> None:
 
 
 def test_no_sdk_private_implementation_access() -> None:
-    """No debt allowlist: MemHub calls public client methods only."""
+    """Runtime uses public APIs; only Graph and stopped-writer migration are privileged."""
     offenders: list[str] = []
     for path in _all_module_files():
         tree = ast.parse(path.read_text(encoding="utf-8"))
@@ -266,16 +272,18 @@ def test_no_sdk_private_implementation_access() -> None:
             if unexpected:
                 offenders.append(f"{relative}: {sorted(unexpected)}")
     assert not offenders, (
-        "SDK private implementation access is forbidden; add a public "
+        "SDK private implementation access is forbidden outside the isolated "
-        f"HyMemoryClient API instead: {offenders}"
+        f"Graph and offline-migration boundaries: {offenders}"
     )
 
 
 def test_sdk_modules_import_only_the_public_package_root() -> None:
-    """Implementation submodules are not a public product contract."""
+    """Runtime imports only public SDK symbols; offline migration is privileged."""
     offenders: list[str] = []
     for path in _all_module_files():
         relative = _relative(path)
+        if relative == f"{PACKAGE_LABEL}/hy_sdk_migration.py":
+            continue
         tree = ast.parse(path.read_text(encoding="utf-8"))
         for node in ast.walk(tree):
             if isinstance(node, ast.Import) and any(

memhub/apps/core/tests/test_cli_restore.py

@@ -302,7 +302,12 @@ def test_sdk_migrate_runs_offline_backfill_without_cutover(tmp_path, monkeypatch
         pass
 
     class Lease:
-        client = object()
+        client = SimpleNamespace(
+            _loop_thread=None,
+            _vector_store=None,
+            _embed_service=None,
+            _graph_store=None,
+        )
 
         def close(self):
             return None
@@ -362,6 +367,14 @@ def test_sdk_cutover_is_offline_confirmed_and_returns_marker(tmp_path, monkeypat
             completed_at="2026-10-21T00:00:00Z",
         ),
     )
+    monkeypatch.setattr(
+        "one_memory.sdk_migration_runner.activate_sdk_migration_staging",
+        lambda *, commit, **_kwargs: commit(),
+    )
+    monkeypatch.setattr(
+        "one_memory.sdk_migration_runner.resume_legacy_cleanup",
+        lambda **_kwargs: (),
+    )
     output = io.StringIO()
 
     code = cli.run(

memhub/apps/core/tests/test_sdk_data_plane_migration.py

@@ -4,8 +4,10 @@ import asyncio
 import json
 
 import pytest
+from hy_memory.models.memory import MemoryLayer, MemoryStatus
 
 from one_memory.database_access import close_main_databases, main_database
+from one_memory.hy_sdk_migration import HySdkMigration
 from one_memory.hy_sdk_types import (
     HySdkMemoryPage,
     HySdkMemoryRecord,
@@ -40,10 +42,15 @@ from one_memory.sdk_migration_inventory import (
     record_raw_migration_plan,
 )
 from one_memory.sdk_migration_runner import (
+    activate_sdk_migration_staging,
     commit_reconciled_cutover,
     prepare_legacy_write_queue_for_cutover,
+    prepare_sdk_migration_staging,
+    recover_sdk_migration_swap,
+    resume_legacy_cleanup,
     run_sdk_data_plane_backfill,
     sdk_cutover_blockers,
+    sdk_migration_staging,
 )
 from one_memory.target_schema import legacy_tables_present
 
@@ -215,6 +222,8 @@ def test_startup_cutover_freezes_pending_job_input_and_preserves_the_queue(store
         sdk_version="2.0.5.dev3",
     )
     assert marker.migration_id == "m1"
+    assert legacy_tables_present(store.database_path)
+    resume_legacy_cleanup(data_dir=data_dir, database_path=store.database_path)
     assert legacy_tables_present(store.database_path) == ()
     assert store.authority() is MemoryAuthority.SDK
     preserved = DurableWriteQueue(store.database_path, sdk_authority=True)
@@ -432,7 +441,7 @@ class _MigrationClient:
             offset=offset,
         )
 
-    async def import_memory_exact(self, **kwargs):
+    async def write_legacy_memory(self, **kwargs):
         self.imported.append(kwargs)
         return {"success": True, "memory_id": kwargs["memory_id"]}
 
@@ -440,7 +449,7 @@ class _MigrationClient:
         self.deleted.append(memory_id)
         return {"success": True, "deleted_count": 1}
 
-    async def list_raw(self, *, limit, offset, **_kwargs):
+    async def read_raw_inventory(self, *, limit, offset, **_kwargs):
         rows = self.raw[offset : offset + limit]
         return HySdkRawPage(
             raw=rows,
@@ -449,7 +458,7 @@ class _MigrationClient:
             offset=offset,
         )
 
-    async def import_raw(self, **kwargs):
+    async def write_legacy_raw(self, **kwargs):
         self.imported_raw.append(kwargs)
         return {"success": True, "raw_id": kwargs["raw_id"]}
 
@@ -754,3 +763,217 @@ def test_offline_backfill_blocks_cutover_when_an_item_is_quarantined(store):
     assert result.phase == "review_required"
     assert result.counts == {"quarantine": 1}
     assert store.authority() is MemoryAuthority.LEGACY
+
+
+class _MigrationLoop:
+    def run(self, coroutine):
+        return asyncio.run(coroutine)
+
+
+class _MigrationVectorStore:
+    def __init__(self):
+        self.nodes = {}
+
+    async def get_by_id(self, memory_id):
+        return self.nodes.get(memory_id)
+
+    async def upsert(self, node):
+        self.nodes[node.node_id] = node
+        return node.node_id
+
+    async def list_by_user(self, *, user_id, layers, **_kwargs):
+        return [
+            node
+            for node in self.nodes.values()
+            if node.user_id == user_id and node.layer in layers
+        ]
+
+
+class _MigrationEmbedder:
+    async def embed(self, _content):
+        return [0.1, 0.2]
+
+
+class _MigrationGraphStore:
+    def __init__(self):
+        self.nodes = {}
+
+    async def upsert_memory_node(self, node):
+        self.nodes[node.node_id] = node
+        return node.node_id
+
+
+class _PrivilegedMigrationClient:
+    def __init__(self):
+        self._loop_thread = _MigrationLoop()
+        self._vector_store = _MigrationVectorStore()
+        self._embed_service = _MigrationEmbedder()
+        self._graph_store = _MigrationGraphStore()
+
+    def list_memories(self, **_kwargs):
+        return {"vdb": {"memories": [], "total": 0}, "request_id": "list"}
+
+    def delete(self, _memory_id):
+        return {"success": True, "deleted_count": 1}
+
+    def get_write_outcome(self, _request_id):
+        return None
+
+
+def test_offline_migration_writes_sdk_stores_without_public_client_import_methods():
+    client = _PrivilegedMigrationClient()
+    migration = HySdkMigration(client)
+
+    raw_result = asyncio.run(
+        migration.write_legacy_raw(
+            content="verbatim",
+            raw_id="raw-1",
+            user_id="local",
+            agent_id="one-memory",
+            session_id="session",
+            source="cursor",
+            memory_at=None,
+            metadata={"role": "user"},
+            request_id="migrate-raw-1",
+        )
+    )
+    memory_result = asyncio.run(
+        migration.write_legacy_memory(
+            content="prefers tea",
+            memory_id="memory-1",
+            user_id="local",
+            agent_id="one-memory",
+            session_id="session",
+            source="legacy-governance",
+            layer="l2_fact",
+            status="active",
+            owner="user",
+            tags=("preference",),
+            supersedes=(),
+            superseded_by=(),
+            is_latest=True,
+            memory_at=None,
+            metadata={},
+            request_id="migrate-memory-1",
+        )
+    )
+    raw_page = asyncio.run(migration.read_raw_inventory(user_id="local"))
+
+    assert raw_result["raw_id"] == "raw-1"
+    assert memory_result["memory_id"] == "memory-1"
+    assert [item.raw_id for item in raw_page.raw] == ["raw-1"]
+    assert client._vector_store.nodes["raw-1"].layer is MemoryLayer.L1_RAW
+    assert client._vector_store.nodes["memory-1"].status is MemoryStatus.ACTIVE
+    assert client._graph_store.nodes["memory-1"].content == "prefers tea"
+    assert not hasattr(client, "import_raw")
+    assert not hasattr(client, "import_memory_exact")
+
+
+def test_raw_http_dto_uses_the_unified_memory_list_shape():
+    page = HySdkRawPage.from_memory_list(
+        {
+            "vdb": {
+                "memories": [
+                    {
+                        "memory_id": "raw-1",
+                        "content": "verbatim",
+                        "layer": "l1_raw",
+                        "status": "active",
+                        "user_id": "local",
+                        "agent_id": "one-memory",
+                        "session_id": "session",
+                        "source": "cursor",
+                        "custom": {"role": "user"},
+                    }
+                ],
+                "total": 1,
+                "limit": 100,
+                "offset": 0,
+            }
+        },
+        limit=100,
+        offset=0,
+    )
+    assert page.raw[0].raw_id == "raw-1"
+    assert page.raw[0].custom == {"role": "user"}
+
+
+def _staging_fixture(tmp_path, store):
+    data_dir = tmp_path / "home"
+    live = data_dir / "hy-memory" / "governed-v2"
+    live.mkdir(parents=True)
+    (live / "sdk-only.txt").write_text("old", encoding="utf-8")
+    paths = sdk_migration_staging(
+        data_dir=data_dir,
+        migration_id="migration-1",
+        namespace="governed-v2",
+    )
+    prepare_sdk_migration_staging(
+        paths=paths,
+        database_path=store.database_path,
+    )
+    return paths
+
+
+def test_staging_preserves_sdk_only_data_and_swaps_after_commit(tmp_path, store):
+    paths = _staging_fixture(tmp_path, store)
+    staged = paths.staging_root / "sdk-only.txt"
+    assert staged.read_text(encoding="utf-8") == "old"
+    staged.write_text("migrated", encoding="utf-8")
+    marker = object()
+
+    result = activate_sdk_migration_staging(
+        paths=paths,
+        database_path=store.database_path,
+        commit=lambda: marker,
+    )
+
+    assert result is marker
+    assert (paths.live_root / "sdk-only.txt").read_text(encoding="utf-8") == "migrated"
+    assert not paths.protective_root.exists()
+    assert not paths.staging_home.exists()
+
+
+def test_incomplete_staging_retry_restarts_from_unchanged_live_root(tmp_path, store):
+    paths = _staging_fixture(tmp_path, store)
+    (paths.staging_root / "sdk-only.txt").write_text("partial", encoding="utf-8")
+
+    prepare_sdk_migration_staging(
+        paths=paths,
+        database_path=store.database_path,
+    )
+
+    assert (paths.staging_root / "sdk-only.txt").read_text(encoding="utf-8") == "old"
+
+
+def test_failed_cutover_restores_live_store_and_keeps_staging(tmp_path, store):
+    paths = _staging_fixture(tmp_path, store)
+    (paths.staging_root / "sdk-only.txt").write_text("migrated", encoding="utf-8")
+
+    def fail():
+        raise DataPlaneMigrationError("cutover_failed")
+
+    with pytest.raises(DataPlaneMigrationError, match="cutover_failed"):
+        activate_sdk_migration_staging(
+            paths=paths,
+            database_path=store.database_path,
+            commit=fail,
+        )
+
+    assert (paths.live_root / "sdk-only.txt").read_text(encoding="utf-8") == "old"
+    assert (paths.staging_root / "sdk-only.txt").read_text(encoding="utf-8") == "migrated"
+
+
+def test_restart_recovers_swap_when_authority_was_not_committed(tmp_path, store):
+    paths = _staging_fixture(tmp_path, store)
+    (paths.staging_root / "sdk-only.txt").write_text("migrated", encoding="utf-8")
+    paths.live_root.replace(paths.protective_root)
+    paths.staging_root.replace(paths.live_root)
+
+    recover_sdk_migration_swap(
+        paths=paths,
+        database_path=store.database_path,
+    )
+
+    assert (paths.live_root / "sdk-only.txt").read_text(encoding="utf-8") == "old"
+    assert (paths.staging_root / "sdk-only.txt").read_text(encoding="utf-8") == "migrated"

memhub/apps/desktop/src/main.ts

@@ -116,7 +116,10 @@ import {
   startStableRuntimeRepair,
 } from "./stable-runtime";
 import { STARTUP_SCREEN_URL, startupScreenHtml } from "./startup-screen";
-import { runStartupSdkMigration } from "./startup-sdk-migration";
+import {
+  requireCompletedStartupSdkMigration,
+  runStartupSdkMigration,
+} from "./startup-sdk-migration";
 import { resolveTrialProfile } from "./trial-profile";
 import { createWindowOptions } from "./window-options";
 
@@ -1097,6 +1100,7 @@ async function initialize(isQuitting: () => boolean): Promise<void> {
       ...(migration.code === undefined ? {} : { error_code: migration.code }),
       details: { state: migration.state },
     });
+    requireCompletedStartupSdkMigration(migration);
   };
 
   try {

memhub/apps/desktop/src/startup-sdk-migration.test.ts

@@ -1,5 +1,8 @@
 import { expect, it, vi } from "vitest";
-import { runStartupSdkMigration } from "./startup-sdk-migration";
+import {
+  requireCompletedStartupSdkMigration,
+  runStartupSdkMigration,
+} from "./startup-sdk-migration";
 
 const core = { command: "/core/one-memory", args: [] };
 
@@ -89,6 +92,9 @@ it("reports an unreadable migration status as deferred", async () => {
     state: "deferred",
     code: "sdk_migration_status_unavailable",
   });
+  expect(() => requireCompletedStartupSdkMigration(result)).toThrow(
+    "sdk_migration_status_unavailable",
+  );
 });
 
 it("resumes legacy table cleanup under the maintenance fence", async () => {

memhub/apps/desktop/src/startup-sdk-migration.ts

@@ -9,6 +9,14 @@ export interface StartupMigrationResult {
   code?: string;
 }
 
+export function requireCompletedStartupSdkMigration(
+  result: StartupMigrationResult,
+): void {
+  if (result.state === "deferred") {
+    throw new Error(result.code ?? "sdk_migration_incomplete");
+  }
+}
+
 interface CommandResult {
   exitCode: number;
   document: Record<string, unknown>;

memhub/docs/ARCHITECTURE.md

@@ -129,12 +129,15 @@ directly for low-latency Cypher projection and store-width administration.
 This exception does not extend to VDB, Coding, prompt, reconcile, or lifecycle.
 
 - Storage backends：仅 SDK config/env（Chroma/FAISS/Qdrant、Kuzu/Neo4j、cache/history sqlite|mysql、coding sqlite 等）。
-- Queue：pending `memory_write_jobs` 只表示尚未调用 SDK 的待处理请求，因此没有 SDK raw/memory ID。worker 单次调用 SDK `add()`；SDK 内部完成 raw 写入与 extraction/reconciliation，成功后 MemHub 仅将返回的 raw ID / memory IDs 记作 job outcome。`import_raw()` 只用于历史数据的 exact migration，不拆分正常写路径。
+- Queue：pending `memory_write_jobs` 只表示尚未调用 SDK 的待处理请求，因此没有 SDK raw/memory ID。worker 单次调用 SDK `add()`；SDK 内部完成 raw 写入与 extraction/reconciliation，成功后 MemHub 仅将返回的 raw ID / memory IDs 记作 job outcome。旧用户启动迁移在 Runtime writers 停止期间直接操作 SDK 存储，不向公开 `HyMemoryClient` 增加迁移专用方法，也不拆分正常写路径。
 - Recall：Host Bridge / Core 先做宿主安装凭据与 tool grant，再调 **SDK search/get**。当前产品没有逐 memory 的 A-harness→B-harness 可见性开关，默认允许跨 harness 搜索；未来若增加来源限制，直接透传 SDK `source` / `include_sources` / `exclude_sources`，不在 MemHub 建 memory ACL。
 - Context wire：cutover marker 提交前只使用 legacy Context v1；提交后 Host Bridge 只使用 SDK-native Context v2。两种 schema 原样校验、原样返回，不把 SDK item 补造成 v1 的 `revision`、`evidence_ids`、`paths` 或 `adoption`，已选择的 v2 路径失败时也不回退 legacy 权威。
 - Ask wire：marker 前保留 Governance Ask v1；marker 后 Web / Host Bridge 只调用 SDK Ask v2。v2 的引用只含 SDK 原生 `memory_id/content/layer/status/source/gmt_created`，生成过程中重复 `get` 校验同一 SDK 快照，不补造 revision、Evidence 或 scopes，SDK 失败不回退 v1。
 - Matter：未形成已发布产品合同，不迁移占位数据；所有 `matter*` 表在升级清理时物理删除，fresh install 不创建，Runtime 不注册 Matter API。
 - 旧用户迁移：由 App 在可见的启动 loading 阶段统一执行；此时 Core 尚未对外提供产品读写。启动流程创建/校验完整 bundle、续跑 ledger、完成 SDK backfill 与 reconciliation，满足全部门禁后才提交 marker 并进入主界面。普通用户不需要另跑 CLI，也不按旧 App 内 SDK 版本分叉迁移。
+- SDK backfill 只写同文件系统的私有 staging 副本；验证后先原子切换 SDK
+  root，再提交 authority marker。切换或 marker 提交失败必须恢复原 live root；
+  App 不得在 migration `deferred` 状态继续启动 Runtime。
 - Matter 若未来重新开发，需以新的 SDK-native 合同重新设计，不继承当前占位 schema。
 
 Pinned SDK 版本见发布锁文件 / `MEMORY.md`（随发布更新）。

memhub/docs/contracts/sdk-data-plane-migration-v1.md

@@ -163,6 +163,13 @@ Before cleanup, every surviving reference is rewritten to its canonical SDK
 ID. Alias, item-ledger, and detailed cutover tables are then dropped; the
 authority singleton and verified bundle remain.
 
+Backfill never mutates the live SDK root. Startup first copies the complete
+local SDK data plane, including SDK-only records, to a same-filesystem staging
+home. Exact legacy writes and reconciliation run there. Cutover renames the
+live root to a protective path, renames staging into place, and only then
+commits SDK authority. A failed commit restores the previous root; restart
+recovery resolves any interrupted rename before opening SDK stores.
+
 ## 8. Queue and unknown outcomes
 
 Queue rows and leases remain MemHub management state.
@@ -172,8 +179,11 @@ Queue rows and leases remain MemHub management state.
 - Its worker calls SDK `add()` once. The SDK owns raw persistence plus
   extraction/reconciliation; a successful job stores only the returned raw ID,
   canonical memory IDs and terminal outcome.
-- Exact `import_raw()` is a historical migration primitive, not the first half
+- Legacy raw and semantic backfill is an App-owned privileged offline
-  of a two-stage normal write path.
+  operation while Runtime writers are stopped. It writes SDK stores directly;
+  no migration-only import/get/list method is added to public
+  `HyMemoryClient`, and this is not the first half of a two-stage normal write
+  path.
 
 - `hy_invoked=0` may be safely retried after normal lease recovery.
 - `memory_write_jobs.hy_invoked=1`, `state=unknown`, or any non-terminal SDK

tests/test_block_l1_raw_in_read_filter.py

@@ -12,9 +12,8 @@ L1_RAW 对外读路径不可见：闸门焊在向量库唯一查询入口。
 hybrid_search_native / list_by_user 四个查询方法的共同必经处，
 上层无法绕开，新增接口自动受保护。
 
-按 ID 直取走主键查询、不经过 filter，故在 `client.async_get` 处默认挡住。
+按 ID 直取是精确读取：raw ID 返回 L1_RAW，fact ID 返回对应事实。权限由
-Python SDK 调用方可显式传 `include_raw=True` 读取；HTTP 接口不传该参数，
+调用 SDK 前的产品边界检查，不由 SDK 根据 layer 猜测。
-继续保持原有隐藏行为。
 
 内部管线需读原始对话时显式 allow_raw_layer=True 放行：
 - `writer._get_recent_dialogue_history`：提取前拼多轮对话历史喂 LLM
@@ -125,8 +124,8 @@ class TestListNoLongerFiltersInProcess:
             "list 又出现进程内过滤 raw 的写法，应交由 vdb 侧 filter 完成"
 
 
-class TestGetByIdRawOptIn:
+class TestGetByIdReturnsExactLayer:
-    """按 ID 直取默认隐藏 raw，只有 SDK 显式选择时才返回。"""
+    """按 ID 直取返回命中的真实层级。"""
 
     @staticmethod
     def _client(node):
@@ -153,35 +152,29 @@ class TestGetByIdRawOptIn:
             user_id="u1",
             agent_id="a1",
             session_id="s1",
+            source="cursor",
             content="原始对话",
             layer=MemoryLayer.L1_RAW,
             status=MemoryStatus.ACTIVE,
             custom={"meta": [{"source": "test"}]},
         )
 
-    def test_default_still_hides_raw(self):
+    def test_raw_id_returns_lite_memory(self):
         from hy_memory.client import HyMemoryClient
 
         client = self._client(self._raw_node())
         raw_get = HyMemoryClient.async_get.__wrapped__
-        assert run(raw_get(client, "raw-1")) is None
+        result = run(raw_get(client, "raw-1"))
-
-    def test_explicit_include_raw_returns_lite_memory(self):
-        from hy_memory.client import HyMemoryClient
-
-        client = self._client(self._raw_node())
-        raw_get = HyMemoryClient.async_get.__wrapped__
-        result = run(raw_get(client, "raw-1", include_raw=True))
 
         assert result["memory_id"] == "raw-1"
         assert result["content"] == "原始对话"
         assert result["layer"] == MemoryLayer.L1_RAW.value
+        assert result["source"] == "cursor"
+        assert result["retrieval_source"] == "vdb"
         assert result["meta"] == [{"source": "test"}]
 
-    def test_public_get_exposes_keyword_only_opt_in(self):
+    def test_public_get_has_no_raw_visibility_switch(self):
         import inspect
         from hy_memory.client import HyMemoryClient
 
-        param = inspect.signature(HyMemoryClient.get).parameters["include_raw"]
+        assert "include_raw" not in inspect.signature(HyMemoryClient.get).parameters
-        assert param.default is False
-        assert param.kind is inspect.Parameter.KEYWORD_ONLY

tests/test_list_layer_filter_client.py

@@ -36,13 +36,16 @@ def _node(
     content="x",
     layer=MemoryLayer.L2_FACT,
     status=MemoryStatus.ACTIVE,
+    session_id="s1",
+    source="default",
 ):
     import random
     n = MemoryNode(
         node_id=str(uuid.uuid4()),
         user_id=user_id,
         agent_id="a1",
-        session_id="s1",
+        session_id=session_id,
+        source=source,
         content=content,
         layer=layer,
         status=status,
@@ -159,30 +162,53 @@ class TestVdbBucketLayerFilter:
         ))
         assert {m["content"] for m in got["memories"]} == {"fact", "summary"}
 
-    def test_explicit_raw_is_blocked(self, faiss_store):
+    def test_explicit_raw_returns_raw(self, faiss_store):
-        """显式只要 L1_RAW 也拿不到：raw 对外读路径一律不可见（契约变更）。
+        """默认列表不含 raw；调用方点名 L1_RAW 时按统一 list 合同返回。"""
-
-        改动前：默认排除只是缺省策略，调用方点名 L1_RAW 即以调用方为准。
-        改动后：向量库侧 allow_raw_layer 默认 False，排除条件随请求下发，
-        raw 不出库；调用方无论传什么 layers 都拿不到原始对话。
-        仅内部管线（writer 拼对话历史、clone 全量复制）显式放行。
-        """
         self._seed(faiss_store)
         got = run(_client_with(faiss_store)._list_vdb_bucket(
             user_id="u1", agent_id=None, limit=100, offset=0, order="desc",
             layers=[MemoryLayer.L1_RAW],
         ))
-        assert got["memories"] == []
+        assert [m["content"] for m in got["memories"]] == ["raw"]
-        assert got["total"] == 0
+        assert got["memories"][0]["layer"] == MemoryLayer.L1_RAW.value
+        assert got["total"] == 1
 
-    def test_raw_mixed_with_other_layers_only_drops_raw(self, faiss_store):
+    def test_raw_mixed_with_other_layers_returns_both(self, faiss_store):
-        """raw 与其他层混传时只剔除 raw，其余层照常返回。"""
         self._seed(faiss_store)
         got = run(_client_with(faiss_store)._list_vdb_bucket(
             user_id="u1", agent_id=None, limit=100, offset=0, order="desc",
             layers=[MemoryLayer.L1_RAW, MemoryLayer.L2_FACT],
         ))
-        assert [m["content"] for m in got["memories"]] == ["fact"]
+        assert {m["content"] for m in got["memories"]} == {"raw", "fact"}
+
+    def test_raw_uses_generic_session_and_source_filters(self, faiss_store):
+        run(faiss_store.upsert_batch([
+            _node(
+                content="cursor raw",
+                layer=MemoryLayer.L1_RAW,
+                session_id="s1",
+                source="cursor",
+            ),
+            _node(
+                content="claude raw",
+                layer=MemoryLayer.L1_RAW,
+                session_id="s2",
+                source="claude",
+            ),
+        ]))
+        got = run(_client_with(faiss_store)._list_vdb_bucket(
+            user_id="u1",
+            agent_id=None,
+            limit=100,
+            offset=0,
+            order="desc",
+            layers=[MemoryLayer.L1_RAW],
+            session_ids=["s2"],
+            include_sources=["claude"],
+            exclude_sources=["cursor"],
+        ))
+        assert [item["content"] for item in got["memories"]] == ["claude raw"]
+        assert got["memories"][0]["source"] == "claude"
 
     def test_status_filter_defaults_to_active_and_can_read_superseded(self, faiss_store):
         run(faiss_store.upsert_batch([

tests/test_raw_public_api.py

@@ -1,69 +1,22 @@
 from __future__ import annotations
 
+import inspect
 from unittest.mock import AsyncMock
 
 import pytest
 
 from hy_memory.client import HyMemoryClient
-from hy_memory.models.memory import MemoryLayer, MemoryStatus
+from hy_memory.models.memory import (
-
+    MemoryLayer,
-
+    MemoryNode,
-class OutcomeCache:
+    MemoryStatus,
-    def __init__(self):
+    SourceType,
-        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):
+    def __init__(self, nodes: list[MemoryNode]):
-        self.nodes = {}
+        self.nodes = nodes
-        self.upserts = 0
-
-    async def get_by_id(self, node_id):
-        return self.nodes.get(node_id)
-
-    async def upsert(self, node):
-        self.upserts += 1
-        self.nodes[node.node_id] = node
-        return node.node_id
-
-    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,
@@ -77,7 +30,7 @@ class RawStore:
         **_kwargs,
     ):
         rows = []
-        for node in self.nodes.values():
+        for node in self.nodes:
             if node.user_id not in user_ids:
                 continue
             if agent_ids and node.agent_id not in agent_ids:
@@ -92,186 +45,51 @@ class RawStore:
         return rows[:limit]
 
 
-def _client():
+def raw_node(raw_id: str, session_id: str, source: str) -> MemoryNode:
+    return MemoryNode(
+        node_id=raw_id,
+        user_id="u1",
+        agent_id="a1",
+        session_id=session_id,
+        source=source,
+        layer=MemoryLayer.L1_RAW,
+        content=f"content {raw_id}",
+        source_type=SourceType.EXPLICIT,
+        status=MemoryStatus.ACTIVE,
+        is_latest=True,
+        embedding=[0.1, 0.2],
+    )
+
+
+def test_migration_only_methods_are_not_public_client_api():
+    for name in (
+        "get_raw",
+        "async_get_raw",
+        "list_raw",
+        "async_list_raw",
+        "import_raw",
+        "async_import_raw",
+        "import_memory_exact",
+        "async_import_memory_exact",
+    ):
+        assert not hasattr(HyMemoryClient, name)
+    assert "include_raw" not in inspect.signature(HyMemoryClient.get).parameters
+
+
+@pytest.mark.asyncio
+async def test_existing_search_contract_can_explicitly_search_raw():
     client = HyMemoryClient.__new__(HyMemoryClient)
-    client._cache = OutcomeCache()
+    client._vector_store = RawStore(
-    client._vector_store = RawStore()
+        [
+            raw_node("raw-1", "s1", "cursor"),
+            raw_node("raw-2", "s2", "claude"),
+        ]
+    )
     client._embed_service = AsyncMock()
     client._embed_service.embed.return_value = [0.1, 0.2]
-    return client
 
-
+    result = await HyMemoryClient._search_explicit_raw_layer.__wrapped__(
-@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_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"],