From 729e699eab2f2a8b169282984e042cd48a0d3d4e Mon Sep 17 00:00:00 2001 From: alvinfei Date: Thu, 22 Oct 2026 20:31:22 +0800 Subject: [PATCH] refactor(sdk): keep legacy migration out of client APIs Co-authored-by: Cursor --- CHANGELOG.md | 29 +- docs/usage.md | 13 +- hy_memory/client.py | 497 +++--------------- memhub/MEMORY.md | 4 +- memhub/apps/core/src/one_memory/api.py | 7 +- .../core/src/one_memory/hy_sdk_migration.py | 302 +++++++++-- .../apps/core/src/one_memory/hy_sdk_types.py | 25 +- .../src/one_memory/sdk_migration_inventory.py | 14 +- .../tests/test_architecture_layer_gate.py | 16 +- .../tests/test_sdk_data_plane_migration.py | 130 ++++- memhub/docs/ARCHITECTURE.md | 2 +- .../contracts/sdk-data-plane-migration-v1.md | 7 +- tests/test_block_l1_raw_in_read_filter.py | 26 +- tests/test_list_layer_filter_client.py | 54 +- tests/test_raw_public_api.py | 282 ++-------- 15 files changed, 613 insertions(+), 795 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 017f804..46c958e 100644 --- a/CHANGELOG.md +++ b/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, - and read modules; prompt/reconcile/response monkey-patches and private - VDB/Coding access were deleted. Kuzu Graph projection/admin remains the one - documented product exception. + public `HyMemoryClient` methods at runtime; prompt/reconcile/response + monkey-patches and private VDB/Coding runtime access were deleted. Kuzu Graph + projection/admin and stopped-writer App migration are the documented product + exceptions. - Extended public SDK `get()` to resolve Chat, Coding Experience, and Working - Fact IDs, including owner/session fields and explicit ambiguity handling; - added constructor-time public embedding-service injection. + Fact IDs, including raw records, owner/session fields, and explicit ambiguity + 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,11 @@ 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 - quiesced local data-plane snapshot/restore contracts for the MemHub - authority migration. + idempotent durable write outcomes, and quiesced local data-plane + snapshot/restore contracts for the MemHub authority migration. Exact legacy + backfill is now an App-owned privileged offline operation rather than a + public `HyMemoryClient` API. - 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 +85,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` - memories, including memories written in lite mode. The default remains - hidden for backward compatibility. +- Unified exact reads and lists across memory layers: `get(memory_id)` returns + the matched layer, while `list_memories(layers=["l1_raw"])` explicitly lists + 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. diff --git a/docs/usage.md b/docs/usage.md index 9468a34..dbd77c4 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -273,8 +273,8 @@ catalog 也可以在没有基础 L0 时形成只读的合成画像。 ```python memory = client.get("memory_id") -# lite 模式写入的是 L1_RAW；按 ID 读取时必须显式选择返回原文。 -raw_memory = client.get("memory_id", include_raw=True) +# raw ID 返回 L1_RAW，fact ID 返回对应事实层。 +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`，因此不会改变现有调用方的原文访问 -边界。只有显式传入 `include_raw=True` 才返回该层；这也是 lite 模式按 ID 读取 -写入内容及其 `meta` 的方式。MemHub 的 HTTP 读取接口不开放此参数。 +`get()` / `async_get()` 是统一的 ID 精确读取：返回结果的 `layer`、`kind` 和 +`variant` 表明实际类型。授权由调用 SDK 前的产品边界完成，不由 SDK 根据 layer +猜测。`list_memories()` 默认仍不包含 `L1_RAW`；显式传 +`layers=["l1_raw"]` 时按同一列表合同返回 raw。 普通 `list_memories()` 返回 `vdb` 桶，并在 Graph 有结果时附带 `graph`；传入 `workspace_id` 时返回生产力记忆的 `scene="productivity"` 和 `experiences`。 diff --git a/hy_memory/client.py b/hy_memory/client.py index d52966d..3ecb63f 100644 --- a/hy_memory/client.py +++ b/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: @@ -5282,6 +4853,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 +4925,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 +4953,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 +4977,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 +5056,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 +5106,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 +5232,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 +5311,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 +5321,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, ) diff --git a/memhub/MEMORY.md b/memhub/MEMORY.md index 4ffcb40..f0075d5 100644 --- a/memhub/MEMORY.md +++ b/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 只在调用前执行 diff --git a/memhub/apps/core/src/one_memory/api.py b/memhub/apps/core/src/one_memory/api.py index 8d7a6ad..0a442aa 100644 --- a/memhub/apps/core/src/one_memory/api.py +++ b/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) diff --git a/memhub/apps/core/src/one_memory/hy_sdk_migration.py b/memhub/apps/core/src/one_memory/hy_sdk_migration.py index 869ee3b..be98469 100644 --- a/memhub/apps/core/src/one_memory/hy_sdk_migration.py +++ b/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,10 @@ from one_memory.hy_sdk_types import ( class MigrationClient(Protocol): + _loop_thread: Any + _vector_store: Any + _embed_service: Any + def list_memories( self, *, @@ -36,32 +53,19 @@ 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 async def list_memories( self, @@ -85,7 +89,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,27 +111,113 @@ class HySdkMigration: ) -> dict[str, object]: raw = await _sdk_call( "hy_sdk_memory_import_failed", - self.client.import_memory_exact, - content, - memory_id=memory_id, + self._loop_thread.run, + self._write_legacy_memory( + content=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, + ), + ) + 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") + 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=source, - layer=layer, - status=status, + 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), - request_id=request_id, + custom=custom, ) - 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") + return { + "success": True, + "memory_id": memory_id, + "memory_ids": [memory_id], + "request_id": request_id, + } - async def list_raw( + async def read_raw_inventory( self, *, user_id: str, @@ -140,18 +230,57 @@ class HySdkMigration: ) -> HySdkRawPage: raw = await _sdk_call( "hy_sdk_raw_list_failed", - self.client.list_raw, - user_id=user_id, - agent_id=agent_id, - session_id=session_id, - source=source, - limit=limit, - offset=offset, - order=order, + self._loop_thread.run, + self._read_raw_inventory( + user_id=user_id, + agent_id=agent_id, + session_id=session_id, + source=source, + 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,18 +295,77 @@ class HySdkMigration: ) -> dict[str, object]: raw = await _sdk_call( "hy_sdk_raw_import_failed", - self.client.import_raw, - content, - raw_id=raw_id, + self._loop_thread.run, + 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=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=source, + 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, - metadata=dict(metadata), - request_id=request_id, + custom=custom, ) - return _successful_result(raw, "hy_sdk_raw_import_failed") + 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( @@ -338,6 +526,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") diff --git a/memhub/apps/core/src/one_memory/hy_sdk_types.py b/memhub/apps/core/src/one_memory/hy_sdk_types.py index 13b00c3..ce7ba3a 100644 --- a/memhub/apps/core/src/one_memory/hy_sdk_types.py +++ b/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") - total = raw.get("total") + bucket = raw.get("vdb") + 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, diff --git a/memhub/apps/core/src/one_memory/sdk_migration_inventory.py b/memhub/apps/core/src/one_memory/sdk_migration_inventory.py index 43e370b..ee3a9ce 100644 --- a/memhub/apps/core/src/one_memory/sdk_migration_inventory.py +++ b/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, diff --git a/memhub/apps/core/tests/test_architecture_layer_gate.py b/memhub/apps/core/tests/test_architecture_layer_gate.py index 0ed06eb..09bf93f 100644 --- a/memhub/apps/core/tests/test_architecture_layer_gate.py +++ b/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", "_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 " - f"HyMemoryClient API instead: {offenders}" + "SDK private implementation access is forbidden outside the isolated " + 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( diff --git a/memhub/apps/core/tests/test_sdk_data_plane_migration.py b/memhub/apps/core/tests/test_sdk_data_plane_migration.py index 3c124d0..06d0e67 100644 --- a/memhub/apps/core/tests/test_sdk_data_plane_migration.py +++ b/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, @@ -432,7 +434,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 +442,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 +451,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 +756,125 @@ 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 _PrivilegedMigrationClient: + def __init__(self): + self._loop_thread = _MigrationLoop() + self._vector_store = _MigrationVectorStore() + self._embed_service = _MigrationEmbedder() + + 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 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"} diff --git a/memhub/docs/ARCHITECTURE.md b/memhub/docs/ARCHITECTURE.md index c0b3c03..66be195 100644 --- a/memhub/docs/ARCHITECTURE.md +++ b/memhub/docs/ARCHITECTURE.md @@ -129,7 +129,7 @@ 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。 diff --git a/memhub/docs/contracts/sdk-data-plane-migration-v1.md b/memhub/docs/contracts/sdk-data-plane-migration-v1.md index eb89919..9b36f5d 100644 --- a/memhub/docs/contracts/sdk-data-plane-migration-v1.md +++ b/memhub/docs/contracts/sdk-data-plane-migration-v1.md @@ -172,8 +172,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 - of a two-stage normal write path. +- Legacy raw and semantic backfill is an App-owned privileged offline + 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 diff --git a/tests/test_block_l1_raw_in_read_filter.py b/tests/test_block_l1_raw_in_read_filter.py index 9c0a06b..d6ba6f5 100644 --- a/tests/test_block_l1_raw_in_read_filter.py +++ b/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` 处默认挡住。 -Python SDK 调用方可显式传 `include_raw=True` 读取；HTTP 接口不传该参数， -继续保持原有隐藏行为。 +按 ID 直取是精确读取：raw ID 返回 L1_RAW，fact ID 返回对应事实。权限由 +调用 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: - """按 ID 直取默认隐藏 raw，只有 SDK 显式选择时才返回。""" +class TestGetByIdReturnsExactLayer: + """按 ID 直取返回命中的真实层级。""" @staticmethod def _client(node): @@ -159,29 +158,20 @@ class TestGetByIdRawOptIn: 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 - - 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)) + result = run(raw_get(client, "raw-1")) assert result["memory_id"] == "raw-1" assert result["content"] == "原始对话" assert result["layer"] == MemoryLayer.L1_RAW.value assert result["meta"] == [{"source": "test"}] - def test_public_get_exposes_keyword_only_opt_in(self): + 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 param.default is False - assert param.kind is inspect.Parameter.KEYWORD_ONLY + assert "include_raw" not in inspect.signature(HyMemoryClient.get).parameters diff --git a/tests/test_list_layer_filter_client.py b/tests/test_list_layer_filter_client.py index ce8569e..43e3d1d 100644 --- a/tests/test_list_layer_filter_client.py +++ b/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): - """显式只要 L1_RAW 也拿不到：raw 对外读路径一律不可见（契约变更）。 - - 改动前：默认排除只是缺省策略，调用方点名 L1_RAW 即以调用方为准。 - 改动后：向量库侧 allow_raw_layer 默认 False，排除条件随请求下发， - raw 不出库；调用方无论传什么 layers 都拿不到原始对话。 - 仅内部管线（writer 拼对话历史、clone 全量复制）显式放行。 - """ + def test_explicit_raw_returns_raw(self, faiss_store): + """默认列表不含 raw；调用方点名 L1_RAW 时按统一 list 合同返回。""" 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 got["total"] == 0 + assert [m["content"] for m in got["memories"]] == ["raw"] + 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): - """raw 与其他层混传时只剔除 raw，其余层照常返回。""" + def test_raw_mixed_with_other_layers_returns_both(self, faiss_store): 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([ diff --git a/tests/test_raw_public_api.py b/tests/test_raw_public_api.py index cf2e12b..3619bfb 100644 --- a/tests/test_raw_public_api.py +++ b/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 - - -class OutcomeCache: - def __init__(self): - self.rows = {} - - async def claim_write_outcome( - self, request_id, user_id, agent_id, session_id, operation - ): - if request_id in self.rows: - return self.rows[request_id] - self.rows[request_id] = { - "request_id": request_id, - "user_id": user_id, - "agent_id": agent_id, - "session_id": session_id, - "operation": operation, - "status": "running", - "result": {}, - } - return None - - async def finish_write_outcome(self, request_id, status, result): - self.rows[request_id].update(status=status, result=dict(result)) - return True - - async def get_write_outcome(self, request_id): - return self.rows.get(request_id) +from hy_memory.models.memory import ( + MemoryLayer, + MemoryNode, + MemoryStatus, + SourceType, +) class RawStore: - def __init__(self): - self.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) - ] + def __init__(self, nodes: list[MemoryNode]): + self.nodes = nodes 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 - -@pytest.mark.asyncio -async def test_import_raw_preserves_identity_without_extraction(): - client = _client() - result = await HyMemoryClient.async_import_raw.__wrapped__( + result = await HyMemoryClient._search_explicit_raw_layer.__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"],