release: publish Hy Memory 2.0.5.dev4

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

CHANGELOG.md

@@ -2,6 +2,40 @@
 
 Hy Memory and MemHub are versioned and released independently.
 
+## Hy Memory 2.0.5.dev4 — 2026-10-21
+
+- Development preview; not the final `2.0.5` release.
+- Adds idempotent `process_raw` / `async_process_raw` APIs that extract and
+  reconcile ordered existing L1 raw IDs without embedding or inserting raw
+  again.
+- Returns durable raw/memory ID receipts for MemHub queue recovery.
+
+## Hy Memory 2.0.5.dev3 — 2026-10-21
+
+- Development preview; not the final `2.0.5` release.
+- Published as
+  [`hy-memory==2.0.5.dev3`](https://pypi.org/project/hy-memory/2.0.5.dev3/).
+- Lets the existing `search` API explicitly query SDK raw with
+  `layers=["l1_raw"]`, including existing source filters.
+
+## Hy Memory 2.0.5.dev2 — 2026-10-21
+
+- Development preview; not the final `2.0.5` release.
+- Published as
+  [`hy-memory==2.0.5.dev2`](https://pypi.org/project/hy-memory/2.0.5.dev2/).
+- Adds exact, idempotent semantic-memory import for legacy MemHub cutover
+  without extraction, reconciliation or identity changes.
+
+## Hy Memory 2.0.5.dev1 — 2026-10-21
+
+- Development preview for the MemHub SDK-authority migration; this is not the
+  final `2.0.5` release.
+- Published as
+  [`hy-memory==2.0.5.dev1`](https://pypi.org/project/hy-memory/2.0.5.dev1/).
+- Adds lifecycle status filtering, durable idempotent write outcomes, exact raw
+  APIs, offline local data-plane snapshot/restore, and fail-closed physical
+  delete receipts.
+
 ## Hy Memory 2.0.3 — 2026-10-17
 
 - Added an optional runtime HTTP transport factory so embedded products can
@@ -14,6 +48,12 @@ Hy Memory and MemHub are versioned and released independently.
 
 ## Unreleased
 
+- `search(..., layers=["l1_raw"])` now explicitly searches SDK raw records in
+  pro/ultra mode and keeps existing user/agent/session/source filters; default
+  search continues to hide raw.
+- Added `import_memory_exact` for idempotent migration of an existing semantic
+  memory with its canonical ID, lifecycle and supersession links, without
+  extraction or reconciliation.
 - Added SDK-authoritative lifecycle listing with explicit status filters,
   idempotent durable write outcomes, exact raw import/get/list APIs, and
   quiesced local data-plane snapshot/restore contracts for the MemHub

hy_memory/client.py

@@ -2631,6 +2631,7 @@ class HyMemoryClient:
         branch: Optional[str] = None,
         request_id: Optional[str] = None,
         meta: Optional[Dict[str, Any]] = None,
+        _existing_raw_ids: Optional[List[str]] = None,
     ) -> Dict[str, Any]:
         """写入记忆（异步）
 
@@ -2799,6 +2800,7 @@ class HyMemoryClient:
             # 落库 memory_ops / WRITE_SUMMARY / pipeline_logs 的单一真相源。
             request_id=request_id,
             client_meta=meta,
+            existing_raw_ids=list(_existing_raw_ids or []),
         )
 
         # mode → writer:
@@ -2847,7 +2849,10 @@ class HyMemoryClient:
             logger.info(f"[add] done: memory_id={result.memory_id} elapsed={result.elapsed_ms:.0f}ms")
 
             # 记录历史（非阻塞，失败仅 warning）
-            if self._history_store is not None:
+            if (
+                self._history_store is not None
+                and not _existing_raw_ids
+            ):
                 try:
                     isolation_key = self._build_isolation_key(uid, agent_id)
                     # 从 content 或 messages 中提取用于记录的文本
@@ -2936,6 +2941,78 @@ class HyMemoryClient:
             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 process_raw(
+        self,
+        raw_ids: List[str],
+        *,
+        user_id: str,
+        agent_id: str = "default_agent",
+        session_id: str = "default_session",
+        source: str = "default",
+        scene: Optional[str] = None,
+        workspace_id: Optional[str] = None,
+        branch: Optional[str] = None,
+        request_id: Optional[str] = None,
+    ) -> Dict[str, Any]:
+        """Extract/reconcile existing L1 raw records without inserting raw again."""
+        return self._loop_thread.run(
+            self.async_process_raw(
+                raw_ids,
+                user_id=user_id,
+                agent_id=agent_id,
+                session_id=session_id,
+                source=source,
+                scene=scene,
+                workspace_id=workspace_id,
+                branch=branch,
+                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))
@@ -3223,6 +3300,299 @@ class HyMemoryClient:
             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_process_raw(
+        self,
+        raw_ids: List[str],
+        *,
+        user_id: str,
+        agent_id: str = "default_agent",
+        session_id: str = "default_session",
+        source: str = "default",
+        scene: Optional[str] = None,
+        workspace_id: Optional[str] = None,
+        branch: Optional[str] = None,
+        request_id: Optional[str] = None,
+    ) -> Dict[str, Any]:
+        """Process exact raw records through extraction/reconciliation in caller order."""
+        from .models.memory import MemoryLayer
+        from .utils.source_filter import normalize_source
+
+        if (
+            not isinstance(raw_ids, list)
+            or not raw_ids
+            or len(raw_ids) > 100
+            or any(not isinstance(raw_id, str) or not raw_id for raw_id in raw_ids)
+            or len(set(raw_ids)) != len(raw_ids)
+        ):
+            raise ValueError("raw_ids must contain 1-100 unique non-empty ids")
+        if not user_id:
+            raise ValueError("user_id is required")
+        normalized_source = normalize_source(source)
+        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="process_raw",
+        )
+        if replay is not None:
+            return replay
+        pipeline_invoked = False
+        try:
+            nodes = []
+            for raw_id in raw_ids:
+                node = await self._vector_store.get_by_id(raw_id)
+                if node is None or node.layer != MemoryLayer.L1_RAW:
+                    raise ValueError(f"raw record is unavailable: {raw_id}")
+                if (
+                    node.user_id != user_id
+                    or node.agent_id != agent_id
+                    or node.session_id != session_id
+                    or node.source != normalized_source
+                ):
+                    raise ValueError("raw record is outside the requested write scope")
+                nodes.append(node)
+            source_orders = [(node.custom or {}).get("source_order") for node in nodes]
+            numeric_orders = [value for value in source_orders if type(value) is int]
+            if numeric_orders and (
+                len(numeric_orders) != len(nodes)
+                or numeric_orders != sorted(numeric_orders)
+                or len(set(numeric_orders)) != len(numeric_orders)
+            ):
+                raise ValueError("raw_ids are not in stable source order")
+            roles = [str((node.custom or {}).get("role") or "user") for node in nodes]
+            if any(role not in {"user", "assistant", "system", "tool"} for role in roles):
+                raise ValueError("raw record has an unsupported message role")
+            messages = [
+                {
+                    "role": role,
+                    "content": node.content,
+                    **(
+                        {"name": str((node.custom or {})["name"])}
+                        if (node.custom or {}).get("name")
+                        else {}
+                    ),
+                    **(
+                        {"tool_call_id": str((node.custom or {})["tool_call_id"])}
+                        if (node.custom or {}).get("tool_call_id")
+                        else {}
+                    ),
+                    **(
+                        {"tool_name": str((node.custom or {})["tool_name"])}
+                        if (node.custom or {}).get("tool_name")
+                        else {}
+                    ),
+                }
+                for node, role in zip(nodes, roles)
+            ]
+            pipeline_invoked = True
+            result = await self._async_add_impl(
+                messages,
+                user_id=user_id,
+                agent_id=agent_id,
+                session_id=session_id,
+                source=normalized_source,
+                _existing_raw_ids=list(raw_ids),
+                memory_at=max(
+                    (node.memory_at for node in nodes if node.memory_at is not None),
+                    default=None,
+                ),
+                scene=scene,
+                workspace_id=workspace_id,
+                branch=branch,
+                request_id=request_id,
+            )
+            memory_ids = list(
+                dict.fromkeys(
+                    result.get("memory_ids")
+                    or [
+                        *(result.get("experience_ids") or []),
+                        *(result.get("fact_ids") or []),
+                    ]
+                )
+            )
+            result = {
+                **result,
+                "raw_ids": list(raw_ids),
+                "memory_ids": memory_ids,
+                "request_id": request_id,
+            }
+        except BaseException:
+            if pipeline_invoked:
+                await self._mark_write_outcome_unknown(request_id)
+            else:
+                await self._finish_write_outcome(
+                    request_id,
+                    "failed",
+                    {
+                        "success": False,
+                        "request_id": request_id,
+                        "raw_ids": list(raw_ids),
+                        "memory_ids": [],
+                        "error_code": "process_raw_invalid",
+                    },
+                )
+            raise
+        await self._finish_write_outcome(
+            request_id,
+            "succeeded" if result.get("success") is True else "failed",
+            result,
+        )
+        return result
+
+    @_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
@@ -3275,6 +3645,66 @@ class HyMemoryClient:
             "offset": offset,
         }
 
+    async def _search_explicit_raw_layer(
+        self,
+        query: str,
+        *,
+        user_ids: List[str],
+        agent_ids: List[str],
+        session_ids: List[str],
+        include_sources: List[str],
+        exclude_sources: List[str],
+        limit: int,
+        min_score: float,
+        created_after: Optional[float],
+        request_id: str,
+        t0: float,
+        query_embedding: Optional[List[float]],
+    ) -> Dict[str, Any]:
+        """Implement `search(..., layers=['l1_raw'])` without a second API."""
+        from .models.memory import MemoryLayer, MemoryStatus
+
+        embedding = query_embedding or await self._embed_service.embed(query)
+        fetch = min(max(limit * (10 if session_ids else 2), limit), 10_000)
+        hits = await self._vector_store.search(
+            query_embedding=embedding,
+            user_ids=user_ids,
+            agent_ids=agent_ids,
+            include_sources=include_sources,
+            exclude_sources=exclude_sources,
+            layers=[MemoryLayer.L1_RAW],
+            limit=fetch,
+            score_threshold=min_score,
+            status_filter=[MemoryStatus.ACTIVE],
+            only_latest=False,
+            created_after=created_after,
+        )
+        sessions = set(session_ids)
+        rows: List[Dict[str, Any]] = []
+        for hit in hits:
+            node = hit.get("node")
+            if node is None or node.layer != MemoryLayer.L1_RAW:
+                continue
+            if sessions and node.session_id not in sessions:
+                continue
+            row = self._raw_node_to_public(node)
+            rows.append(
+                {
+                    "memory_id": row.pop("raw_id"),
+                    **row,
+                    "layer": MemoryLayer.L1_RAW.value,
+                    "status": MemoryStatus.ACTIVE.value,
+                    "score": round(float(hit.get("score", 0.0)), 4),
+                }
+            )
+            if len(rows) >= limit:
+                break
+        return {
+            "request_id": request_id,
+            "memories": {"schema": [], "proactive": [], "normal": rows},
+            "elapsed_ms": round((time.perf_counter() - t0) * 1000, 2),
+        }
+
     @staticmethod
     def _public_search_views(
         result_memories: List[Dict[str, Any]],
@@ -3935,6 +4365,22 @@ class HyMemoryClient:
         effective_search_layers = [MemoryLayer.L1_RAW] if lite_mode else layers
         effective_schema_limit = 0 if lite_mode else schema_limit
         effective_intention_limit = 0 if lite_mode else intention_limit
+        normalized_search_layers = self._normalize_layers(effective_search_layers)
+        if normalized_search_layers == [MemoryLayer.L1_RAW]:
+            return await self._search_explicit_raw_layer(
+                effective_target_query,
+                user_ids=effective_user_ids,
+                agent_ids=effective_agent_ids,
+                session_ids=effective_session_ids,
+                include_sources=effective_include_sources,
+                exclude_sources=effective_exclude_sources,
+                limit=limit,
+                min_score=min_score,
+                created_after=effective_created_after,
+                request_id=request_id,
+                t0=t0,
+                query_embedding=query_embedding,
+            )
 
         read_req = ReadRequest(
             query=effective_target_query,
@@ -3955,7 +4401,7 @@ class HyMemoryClient:
             layers=(
                 [
                     layer.value
-                    for layer in self._normalize_layers(effective_search_layers)
+                    for layer in normalized_search_layers
                 ]
                 if effective_search_layers else None
             ),
@@ -3980,7 +4426,7 @@ class HyMemoryClient:
         _wanted_layers = (
             {
                 layer.value
-                for layer in self._normalize_layers(effective_search_layers)
+                for layer in normalized_search_layers
             }
             if effective_search_layers else None
         )

hy_memory/pipelines/base.py

@@ -231,6 +231,10 @@ class WriteRequest:
     # 已有记忆 (V1 合并检测用)
     existing_memories: Optional[List[Dict[str, Any]]] = None
 
+    # Public client process_raw() only: reuse these exact L1 nodes as input.
+    # Ordinary add() never populates this field.
+    existing_raw_ids: List[str] = field(default_factory=list)
+
     # 记忆时间戳（不传则用当前时间 datetime.now()）
     memory_at: Optional[datetime] = None
 

hy_memory/pipelines/writer.py

@@ -280,6 +280,27 @@ class MemoryWriter(WritePipeline):
             custom["source_raw_memory_ids"] = source_ids
         return custom
 
+    async def _attach_existing_raw_ids(
+        self,
+        vector_store,
+        memory_ids: List[str],
+        raw_ids: List[str],
+    ) -> None:
+        """Attach every reused raw record to canonical outputs."""
+        for memory_id in memory_ids:
+            node = await vector_store.get_by_id(memory_id)
+            if node is None:
+                continue
+            custom = dict(node.custom or {})
+            source_ids = list(custom.get("source_raw_memory_ids") or [])
+            legacy_source = str(getattr(node, "source_raw_memory_id", "") or "")
+            for raw_id in (legacy_source, *raw_ids):
+                if raw_id and raw_id not in source_ids:
+                    source_ids.append(raw_id)
+            if source_ids:
+                custom["source_raw_memory_ids"] = source_ids
+                await vector_store.update_payload(memory_id, {"custom": custom})
+
     @staticmethod
     def _attach_client_meta(custom: Dict[str, Any], request: Optional[WriteRequest]) -> None:
         """给新建节点挂上调用方 meta（单份 list）。
@@ -2596,13 +2617,23 @@ class MemoryWriter(WritePipeline):
             from ..metrics import MetricsCollector
             MetricsCollector.get().sys1_start()
 
-            # 2. 向量化 + 持久化（L1_RAW）
+            existing_raw_ids = request.existing_raw_ids
+            if (
+                not isinstance(existing_raw_ids, list)
+                or any(not isinstance(raw_id, str) or not raw_id for raw_id in existing_raw_ids)
+            ):
+                raise ValueError("_existing_raw_ids must be a list of non-empty ids")
+
+            # 2. 向量化（处理已有 raw 时复用已存节点，不再次 embed）
             _t_embed = datetime.now()
-            with tracer.span("embed") as s:
-                embedding, embed_meta = await self.embed_service.embed_queued_with_meta(
-                    request.content
-                )
-                s.set_output({"dims": len(embedding), "content_len": len(request.content)})
+            if existing_raw_ids:
+                embedding, embed_meta = [], {"reused_raw": True}
+            else:
+                with tracer.span("embed") as s:
+                    embedding, embed_meta = await self.embed_service.embed_queued_with_meta(
+                        request.content
+                    )
+                    s.set_output({"dims": len(embedding), "content_len": len(request.content)})
             _embed_ms = (datetime.now() - _t_embed).total_seconds() * 1000
             response.extra["embedding"] = embedding
             await self._emit_pipeline_step(
@@ -2625,27 +2656,43 @@ class MemoryWriter(WritePipeline):
             with tracer.span("qdrant_upsert") as s:
                 try:
                     vector_store = await self._get_vector_store()
-                    layer_enum = MemoryLayer.from_string(suggested_layer)
-                    _raw_custom: Dict[str, Any] = {}
-                    self._attach_client_meta(_raw_custom, request)
-                    mem_node = MemoryNode(
-                        user_id=request.user_id,
-                        agent_id=request.agent_id or "default_agent",
-                        session_id=request.session_id or "default_session",
-                        source=request.source,
-                        layer=layer_enum,
-                        content=request.content,
-                        source_type=SourceType.EXPLICIT,
-                        status=MemoryStatus.ACTIVE,
-                        is_latest=True,
-                        embedding=embedding,
-                        memory_at=request.memory_at,
-                        custom=_raw_custom,
-                    )
-                    memory_id = await vector_store.upsert(mem_node)
+                    if existing_raw_ids:
+                        memory_id = existing_raw_ids[-1]
+                        mem_node = await vector_store.get_by_id(memory_id)
+                        if (
+                            mem_node is None
+                            or mem_node.layer != MemoryLayer.L1_RAW
+                            or mem_node.user_id != request.user_id
+                            or mem_node.agent_id != (request.agent_id or "default_agent")
+                            or mem_node.session_id != (request.session_id or "default_session")
+                            or mem_node.source != request.source
+                        ):
+                            raise ValueError("existing raw record is outside the write scope")
+                    else:
+                        layer_enum = MemoryLayer.from_string(suggested_layer)
+                        _raw_custom: Dict[str, Any] = {}
+                        self._attach_client_meta(_raw_custom, request)
+                        mem_node = MemoryNode(
+                            user_id=request.user_id,
+                            agent_id=request.agent_id or "default_agent",
+                            session_id=request.session_id or "default_session",
+                            source=request.source,
+                            layer=layer_enum,
+                            content=request.content,
+                            source_type=SourceType.EXPLICIT,
+                            status=MemoryStatus.ACTIVE,
+                            is_latest=True,
+                            embedding=embedding,
+                            memory_at=request.memory_at,
+                            custom=_raw_custom,
+                        )
+                        memory_id = await vector_store.upsert(mem_node)
                     response.memory_id = memory_id
                     s.set_output({"memory_id": memory_id, "layer": suggested_layer})
-                    logger.debug(f"[write] persisted: id={memory_id} layer={suggested_layer}")
+                    logger.debug(
+                        f"[write] {'reused' if existing_raw_ids else 'persisted'}: "
+                        f"id={memory_id} layer={suggested_layer}"
+                    )
                 except Exception as persist_err:
                     _l1_error = str(persist_err)
                     s.set_error(_l1_error)
@@ -2709,6 +2756,12 @@ class MemoryWriter(WritePipeline):
                         recent_history_count=_recent_history_count,
                     )
                     total_tokens += response.extra.get("_agent_tokens", 0)
+                if existing_raw_ids and response.extra.get("agent_stored_ids"):
+                    await self._attach_existing_raw_ids(
+                        vector_store,
+                        list(response.extra["agent_stored_ids"]),
+                        existing_raw_ids,
+                    )
                 if response.extra.get("agent_status") == "failed":
                     await self._emit_pipeline_step(
                         request,
@@ -2718,12 +2771,13 @@ class MemoryWriter(WritePipeline):
                             "error": response.extra.get("agent_error", ""),
                         },
                     )
-                    await self._rollback_failed_l1(
-                        request=request,
-                        response=response,
-                        vector_store=vector_store,
-                        memory_id=memory_id,
-                    )
+                    if not existing_raw_ids:
+                        await self._rollback_failed_l1(
+                            request=request,
+                            response=response,
+                            vector_store=vector_store,
+                            memory_id=memory_id,
+                        )
                     response.tokens_used = total_tokens
                     _failed_workflow_ms = (
                         datetime.now() - _t_workflow

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "hy-memory"
-version = "2.0.4"
+version = "2.0.5.dev4"
 description = "HY Memory - Industrial-grade dual-system cognitive memory framework"
 readme = "README.md"
 requires-python = ">=3.9"

tests/test_raw_public_api.py

@@ -5,7 +5,15 @@ from unittest.mock import AsyncMock
 import pytest
 
 from hy_memory.client import HyMemoryClient
-from hy_memory.models.memory import MemoryLayer
+from hy_memory.config import MemoryConfig
+from hy_memory.models.memory import (
+    MemoryLayer,
+    MemoryNode,
+    MemoryStatus,
+    SourceType,
+)
+from hy_memory.pipelines.base import WriteRequest
+from hy_memory.pipelines.writer import MemoryWriter
 
 
 class OutcomeCache:
@@ -40,15 +48,24 @@ class RawStore:
     def __init__(self):
         self.nodes = {}
         self.upserts = 0
+        self.supports_fulltext = False
 
     async def get_by_id(self, node_id):
         return self.nodes.get(node_id)
 
+    async def initialize(self):
+        return None
+
     async def upsert(self, node):
         self.upserts += 1
         self.nodes[node.node_id] = node
         return node.node_id
 
+    async def update_payload(self, node_id, payload):
+        node = self.nodes[node_id]
+        for key, value in payload.items():
+            setattr(node, key, value)
+
     async def list_by_user(
         self,
         *,
@@ -65,6 +82,32 @@ class RawStore:
             and (not include_sources or node.source in include_sources)
         ]
 
+    async def search(
+        self,
+        *,
+        user_ids,
+        agent_ids,
+        include_sources,
+        exclude_sources,
+        layers,
+        limit,
+        **_kwargs,
+    ):
+        rows = []
+        for node in self.nodes.values():
+            if node.user_id not in user_ids:
+                continue
+            if agent_ids and node.agent_id not in agent_ids:
+                continue
+            if include_sources and node.source not in include_sources:
+                continue
+            if exclude_sources and node.source in exclude_sources:
+                continue
+            if node.layer not in layers:
+                continue
+            rows.append({"node": node, "score": 0.9})
+        return rows[:limit]
+
 
 def _client():
     client = HyMemoryClient.__new__(HyMemoryClient)
@@ -117,6 +160,101 @@ async def test_import_raw_request_replay_does_not_write_twice():
     assert client._vector_store.upserts == 1
 
 
+@pytest.mark.asyncio
+async def test_process_raw_is_idempotent_and_does_not_insert_raw_again():
+    client = _client()
+    for index, role in enumerate(("user", "assistant"), start=1):
+        await HyMemoryClient.async_import_raw.__wrapped__(
+            client,
+            f"turn-{index}",
+            user_id="u1",
+            agent_id="a1",
+            session_id="s1",
+            source="cursor",
+            raw_id=f"raw-{index}",
+            request_id=f"import-{index}",
+            metadata={"role": role, "source_order": index},
+        )
+    client._async_add_impl = AsyncMock(
+        return_value={"success": True, "memory_ids": ["memory-1"]}
+    )
+
+    first = await HyMemoryClient.async_process_raw.__wrapped__(
+        client,
+        ["raw-1", "raw-2"],
+        user_id="u1",
+        agent_id="a1",
+        session_id="s1",
+        source="cursor",
+        request_id="process-1",
+    )
+    second = await HyMemoryClient.async_process_raw.__wrapped__(
+        client,
+        ["raw-1", "raw-2"],
+        user_id="u1",
+        agent_id="a1",
+        session_id="s1",
+        source="cursor",
+        request_id="process-1",
+    )
+
+    assert first["raw_ids"] == ["raw-1", "raw-2"]
+    assert first["memory_ids"] == ["memory-1"]
+    assert second == {**first, "replayed": True}
+    assert client._vector_store.upserts == 2
+    client._async_add_impl.assert_awaited_once()
+    assert client._async_add_impl.await_args.kwargs["_existing_raw_ids"] == [
+        "raw-1",
+        "raw-2",
+    ]
+
+
+@pytest.mark.asyncio
+async def test_writer_reuses_existing_raw_without_embedding_or_upsert():
+    store = RawStore()
+    raw = MemoryNode(
+        node_id="raw-1",
+        user_id="u1",
+        agent_id="a1",
+        session_id="s1",
+        source="cursor",
+        layer=MemoryLayer.L1_RAW,
+        content="verbatim",
+        source_type=SourceType.EXPLICIT,
+        status=MemoryStatus.ACTIVE,
+        is_latest=True,
+        embedding=[0.1, 0.2],
+    )
+    store.nodes[raw.node_id] = raw
+    embed = AsyncMock()
+    writer = MemoryWriter(
+        MemoryConfig(),
+        embed_service=embed,
+        vector_store=store,
+        cache=None,
+    )
+
+    response = await writer.write(
+        WriteRequest(
+            content="[user]: verbatim",
+            user_id="u1",
+            agent_id="a1",
+            session_id="s1",
+            source="cursor",
+            request_id="process-1",
+            existing_raw_ids=["raw-1"],
+            extra={
+                "agent_mode": "disabled",
+            },
+        )
+    )
+
+    assert response.success is True
+    assert response.memory_id == "raw-1"
+    assert store.upserts == 0
+    embed.embed_queued_with_meta.assert_not_awaited()
+
+
 @pytest.mark.asyncio
 async def test_existing_raw_id_must_match_exact_content_and_owner():
     client = _client()
@@ -160,3 +298,108 @@ async def test_raw_get_and_list_use_explicit_api():
     )
     assert page["total"] == 1
     assert [item["raw_id"] for item in page["raw"]] == ["raw-2"]
+
+
+@pytest.mark.asyncio
+async def test_exact_memory_import_preserves_id_lifecycle_and_chain():
+    client = _client()
+    result = await HyMemoryClient.async_import_memory_exact.__wrapped__(
+        client,
+        "older preference",
+        memory_id="memory-1",
+        user_id="u1",
+        agent_id="a1",
+        session_id="s1",
+        source="cursor",
+        layer="l2_fact",
+        status="superseded",
+        owner="user",
+        tags=["preference"],
+        superseded_by=["memory-2"],
+        is_latest=False,
+        request_id="request-memory-1",
+    )
+
+    assert result["memory_ids"] == ["memory-1"]
+    node = client._vector_store.nodes["memory-1"]
+    assert node.content == "older preference"
+    assert node.layer == MemoryLayer.L2_FACT
+    assert node.status == MemoryStatus.SUPERSEDED
+    assert node.superseded_by == ["memory-2"]
+    assert node.is_latest is False
+    assert node.custom["input_kind"] == "memory_exact_import"
+
+
+@pytest.mark.asyncio
+async def test_exact_memory_import_refuses_same_id_with_different_body():
+    client = _client()
+    await HyMemoryClient.async_import_memory_exact.__wrapped__(
+        client,
+        "first",
+        memory_id="memory-1",
+        user_id="u1",
+        request_id="request-memory-1",
+    )
+
+    with pytest.raises(ValueError, match="different memory content"):
+        await HyMemoryClient.async_import_memory_exact.__wrapped__(
+            client,
+            "changed",
+            memory_id="memory-1",
+            user_id="u1",
+            request_id="request-memory-2",
+        )
+
+
+@pytest.mark.asyncio
+async def test_exact_memory_import_rejects_raw_layer():
+    client = _client()
+    with pytest.raises(ValueError, match="use import_raw"):
+        await HyMemoryClient.async_import_memory_exact.__wrapped__(
+            client,
+            "raw",
+            memory_id="memory-1",
+            user_id="u1",
+            layer="l1_raw",
+            request_id="request-memory-1",
+        )
+
+
+@pytest.mark.asyncio
+async def test_existing_search_contract_can_explicitly_search_raw_with_source_filter():
+    client = _client()
+    for raw_id, session_id, source in (
+        ("raw-1", "s1", "cursor"),
+        ("raw-2", "s2", "claude"),
+    ):
+        await HyMemoryClient.async_import_raw.__wrapped__(
+            client,
+            f"content {raw_id}",
+            user_id="u1",
+            agent_id="a1",
+            session_id=session_id,
+            source=source,
+            raw_id=raw_id,
+            request_id=f"request-{raw_id}",
+        )
+
+    result = await client._search_explicit_raw_layer(
+        "content",
+        user_ids=["u1"],
+        agent_ids=["a1"],
+        session_ids=["s2"],
+        include_sources=["claude"],
+        exclude_sources=["cursor"],
+        limit=5,
+        min_score=0.0,
+        created_after=None,
+        request_id="search-1",
+        t0=0.0,
+        query_embedding=[0.1, 0.2],
+    )
+
+    assert [item["memory_id"] for item in result["memories"]["normal"]] == [
+        "raw-2"
+    ]
+    assert result["memories"]["normal"][0]["layer"] == "l1_raw"
+    assert result["memories"]["normal"][0]["source"] == "claude"

uv.lock

@@ -1715,7 +1715,7 @@ wheels = [
 
 [[package]]
 name = "hy-memory"
-version = "2.0.4"
+version = "2.0.5.dev4"
 source = { editable = "." }
 dependencies = [
     { name = "chromadb" },