From 5b342f89f14aaa48d48830695ffb0468ad1aa1d8 Mon Sep 17 00:00:00 2001
From: dingliang <2650876010@qq.com>
Date: Wed, 8 Jul 2026 16:13:46 +0800
Subject: [PATCH 01/13] feat(api): align OpenMem v1 SDK with cloud API

---
 src/memos/api/client.py  | 263 ++++++++++++++++++++++---
 tests/api/test_client.py | 406 +++++++++++++++++++++++++++++++++++++++
 2 files changed, 647 insertions(+), 22 deletions(-)
 create mode 100644 tests/api/test_client.py

diff --git a/src/memos/api/client.py b/src/memos/api/client.py
index 818ce5e0d..1ace6ec8f 100644
--- a/src/memos/api/client.py
+++ b/src/memos/api/client.py
@@ -65,6 +65,32 @@ def _validate_required_params(self, **params):
             if not param_value:
                 raise ValueError(f"{param_name} is required")
 
+    def _validate_profile_subject(self, user_id: str | None, agent_id: str | None) -> None:
+        if bool(user_id) == bool(agent_id):
+            raise ValueError("exactly one of user_id or agent_id is required")
+
+    def _post_json_dict(
+        self, endpoint: str, payload: dict[str, Any], operation: str
+    ) -> dict[str, Any] | None:
+        url = f"{self.base_url}/{endpoint}"
+        for retry in range(MAX_RETRY_COUNT):
+            try:
+                response = requests.post(
+                    url, data=json.dumps(payload), headers=self.headers, timeout=30
+                )
+                response.raise_for_status()
+                return response.json()
+            except Exception as e:
+                logger.error(
+                    "Failed to %s (retry %s/%s): %s",
+                    operation,
+                    retry + 1,
+                    MAX_RETRY_COUNT,
+                    e,
+                )
+                if retry == MAX_RETRY_COUNT - 1:
+                    raise
+
     def get_message(
         self,
         user_id: str,
@@ -102,7 +128,7 @@ def get_message(
     def add_message(
         self,
         messages: list[dict[str, Any]],
-        user_id: str,
+        user_id: str | list[str],
         conversation_id: str,
         info: dict[str, Any] | None = None,
         source: str | None = None,
@@ -112,6 +138,7 @@ def add_message(
         tags: list[str] | None = None,
         allow_public: bool = False,
         allow_knowledgebase_ids: list[str] | None = None,
+        allow_memory_view: list[str] | None = None,
     ) -> MemOSAddResponse | None:
         """Add message"""
         # Validate required parameters
@@ -130,8 +157,9 @@ def add_message(
             "agent_id": agent_id,
             "allow_public": allow_public,
             "allow_knowledgebase_ids": allow_knowledgebase_ids,
+            "allow_memory_view": allow_memory_view,
             "tags": tags,
-            "asyncMode": async_mode,
+            "async_mode": async_mode,
         }
         for retry in range(MAX_RETRY_COUNT):
             try:
@@ -151,7 +179,8 @@ def search_memory(
         self,
         query: str,
         user_id: str,
-        conversation_id: str,
+        conversation_id: str | None = None,
+        agent_id: str | None = None,
         memory_limit_number: int = 6,
         include_preference: bool = True,
         knowledgebase_ids: list[str] | None = None,
@@ -160,6 +189,11 @@ def search_memory(
         include_tool_memory: bool = False,
         preference_limit_number: int = 6,
         tool_memory_limit_number: int = 6,
+        relativity: float | None = None,
+        include_skill: bool = False,
+        skill_limit_number: int = 6,
+        include_memory_view: list[str] | None = None,
+        context_format: str = "memory",
     ) -> MemOSSearchResponse | None:
         """Search memories"""
         # Validate required parameters
@@ -170,12 +204,18 @@ def search_memory(
             "query": query,
             "user_id": user_id,
             "conversation_id": conversation_id,
+            "agent_id": agent_id,
             "memory_limit_number": memory_limit_number,
             "include_preference": include_preference,
             "knowledgebase_ids": knowledgebase_ids,
             "filter": filter,
             "preference_limit_number": preference_limit_number,
             "tool_memory_limit_number": tool_memory_limit_number,
+            "relativity": relativity,
+            "include_skill": include_skill,
+            "skill_limit_number": skill_limit_number,
+            "include_memory_view": include_memory_view,
+            "context_format": context_format,
             "source": source,
             "include_tool_memory": include_tool_memory,
         }
@@ -195,16 +235,28 @@ def search_memory(
                     raise
 
     def get_memory(
-        self, user_id: str, include_preference: bool = True, page: int = 1, size: int = 10
+        self,
+        user_id: str | None = None,
+        include_preference: bool = True,
+        page: int = 1,
+        size: int = 10,
+        agent_id: str | None = None,
+        include_tool_memory: bool = True,
+        include_memory_view: list[str] | None = None,
+        filter: dict[str, Any] | None = None,
     ) -> MemOSGetMemoryResponse | None:
         """get memories"""
         # Validate required parameters
-        self._validate_required_params(include_preference=include_preference, user_id=user_id)
+        self._validate_profile_subject(user_id, agent_id)
 
         url = f"{self.base_url}/get/memory"
         payload = {
             "include_preference": include_preference,
             "user_id": user_id,
+            "agent_id": agent_id,
+            "include_tool_memory": include_tool_memory,
+            "include_memory_view": include_memory_view,
+            "filter": filter,
             "page": page,
             "size": size,
         }
@@ -313,7 +365,7 @@ def add_knowledgebase_file_json(
                     raise
 
     def add_knowledgebase_file_form(
-        self, knowledgebase_id: str, files: list[str]
+        self, knowledgebase_id: str, files: list[str], type: str | None = None
     ) -> MemOSAddKnowledgebaseFileResponse | None:
         """
         add knowledgebase-file from form
@@ -321,12 +373,12 @@ def add_knowledgebase_file_form(
         # Validate required parameters
         self._validate_required_params(knowledgebase_id=knowledgebase_id, files=files)
 
-        def build_file_form_param(file_path):
+        def build_file_form_param(file_path: str):
             """
             form-Automatically generate the structure required for the `files` parameter in requests based on the local file path
             """
             if not os.path.isfile(file_path):
-                logger.warning(f"File {file_path} does not exist")
+                logger.warning("File %s does not exist", file_path)
                 return None
             filename = os.path.basename(file_path)
 
@@ -335,31 +387,47 @@ def build_file_form_param(file_path):
                 mime_type = "application/octet-stream"
             return ("file", (filename, open(file_path, "rb"), mime_type))
 
+        def build_file_form_params() -> list:
+            file_params = [
+                file_param
+                for file_path in files
+                if (file_param := build_file_form_param(file_path)) is not None
+            ]
+            if not file_params:
+                raise ValueError("files must contain at least one valid file path")
+            return file_params
+
         url = f"{self.base_url}/add/knowledgebase-file"
         payload = {
             "knowledgebase_id": knowledgebase_id,
         }
+        if type is not None:
+            payload["type"] = type
         headers = {
             "Authorization": f"Token {self.api_key}",
         }
         for retry in range(MAX_RETRY_COUNT):
+            file_params = []
             try:
+                file_params = build_file_form_params()
                 response = requests.post(
                     url,
                     params=payload,
                     headers=headers,
                     timeout=30,
-                    files=[build_file_form_param(file_path) for file_path in files],
+                    files=file_params,
                 )
                 response.raise_for_status()
                 response_data = response.json()
-                print(response_data)
 
                 return MemOSAddKnowledgebaseFileResponse(**response_data)
             except Exception as e:
                 logger.error(f"Failed to add knowledgebase-file form (retry {retry + 1}/3): {e}")
                 if retry == MAX_RETRY_COUNT - 1:
                     raise
+            finally:
+                for file_param in file_params:
+                    file_param[1][1].close()
 
     def delete_knowledgebase_file(
         self, file_ids: list[str]
@@ -390,17 +458,27 @@ def delete_knowledgebase_file(
                     raise
 
     def get_knowledgebase_file(
-        self, file_ids: list[str]
+        self,
+        file_ids: list[str] | None = None,
+        knowledgebase_id: str | None = None,
+        type: str | None = None,
+        page: int | None = None,
+        page_size: int | None = None,
     ) -> MemOSGetKnowledgebaseFileResponse | None:
         """
         get knowledgebase-file
         """
         # Validate required parameters
-        self._validate_required_params(file_ids=file_ids)
+        if bool(file_ids) == bool(knowledgebase_id):
+            raise ValueError("exactly one of file_ids or knowledgebase_id is required")
 
         url = f"{self.base_url}/get/knowledgebase-file"
         payload = {
             "file_ids": file_ids,
+            "knowledgebase_id": knowledgebase_id,
+            "type": type,
+            "page": page,
+            "page_size": page_size,
         }
 
         for retry in range(MAX_RETRY_COUNT):
@@ -486,17 +564,43 @@ def add_feedback(
                     raise
 
     def delete_memory(
-        self, user_ids: list[str], memory_ids: list[str]
+        self,
+        user_ids: list[str] | None = None,
+        memory_ids: list[str] | None = None,
+        *,
+        user_id: str | None = None,
+        agent_id: str | None = None,
+        filter: dict[str, Any] | None = None,
+        memory_type: str | None = None,
     ) -> MemOSDeleteMemoryResponse | None:
         """delete_memory memories"""
-        # Validate required parameters
-        self._validate_required_params(user_ids=user_ids, memory_ids=memory_ids)
+        if user_id is None and user_ids:
+            if len(user_ids) != 1 and not memory_ids:
+                raise ValueError("current API supports a single user_id, not multiple user_ids")
+            if not memory_ids:
+                user_id = user_ids[0]
+
+        delete_modes = [
+            bool(memory_ids),
+            bool(user_id),
+            bool(agent_id),
+            filter is not None,
+        ]
+        if sum(delete_modes) != 1:
+            raise ValueError("exactly one delete condition is required")
 
         url = f"{self.base_url}/delete/memory"
-        payload = {
-            "user_ids": user_ids,
-            "memory_ids": memory_ids,
-        }
+        payload: dict[str, Any] = {}
+        if memory_ids:
+            payload["memory_ids"] = memory_ids
+        if user_id:
+            payload["user_id"] = user_id
+        if agent_id:
+            payload["agent_id"] = agent_id
+        if filter is not None:
+            payload["filter"] = filter
+        if memory_type is not None:
+            payload["memory_type"] = memory_type
 
         for retry in range(MAX_RETRY_COUNT):
             try:
@@ -512,6 +616,111 @@ def delete_memory(
                 if retry == MAX_RETRY_COUNT - 1:
                     raise
 
+    def update_memory(
+        self,
+        memory_id: str,
+        content: str | None = None,
+        title: str | None = None,
+        status: str | None = None,
+    ) -> dict[str, Any] | None:
+        """Update an existing memory."""
+        self._validate_required_params(memory_id=memory_id)
+        if not content and not title and not status:
+            raise ValueError("content, title or status is required")
+
+        payload = {
+            "memory_id": memory_id,
+            "content": content,
+            "title": title,
+            "status": status,
+        }
+        return self._post_json_dict("update/memory", payload, "update memory")
+
+    def extract_memory(
+        self,
+        messages: list[dict[str, Any]],
+        extraction_types: list[str] | None = None,
+        model: str | None = None,
+    ) -> dict[str, Any] | None:
+        """Extract memory candidates from conversation messages."""
+        self._validate_required_params(messages=messages)
+
+        payload = {
+            "messages": messages,
+            "extraction_types": extraction_types,
+            "model": model,
+        }
+        return self._post_json_dict("extract/memory", payload, "extract memory")
+
+    def rerank(
+        self,
+        query: str,
+        documents: list[str],
+        model: str | None = None,
+        top_n: int | None = None,
+    ) -> dict[str, Any] | None:
+        """Rerank documents for a query."""
+        self._validate_required_params(query=query, documents=documents)
+        if top_n is not None and top_n <= 0:
+            raise ValueError("top_n must be greater than 0")
+
+        payload = {
+            "query": query,
+            "documents": documents,
+            "model": model,
+            "top_n": top_n,
+        }
+        return self._post_json_dict("rerank", payload, "rerank documents")
+
+    def bind_profile_template(self, bind_list: list[dict[str, Any]]) -> dict[str, Any] | None:
+        """Bind profile templates to user or agent subjects."""
+        self._validate_required_params(bind_list=bind_list)
+
+        payload = {
+            "bind_list": bind_list,
+        }
+        return self._post_json_dict("bind/profile_template", payload, "bind profile template")
+
+    def edit_profile(
+        self,
+        profile_template_id: str,
+        user_id: str | None = None,
+        agent_id: str | None = None,
+        metadata: dict[str, Any] | None = None,
+        remove_fields: list[str] | None = None,
+    ) -> dict[str, Any] | None:
+        """Edit a profile instance."""
+        self._validate_required_params(profile_template_id=profile_template_id)
+        self._validate_profile_subject(user_id, agent_id)
+        if metadata is None and not remove_fields:
+            raise ValueError("metadata or remove_fields is required")
+
+        payload = {
+            "user_id": user_id,
+            "agent_id": agent_id,
+            "profile_template_id": profile_template_id,
+            "metadata": metadata,
+            "remove_fields": remove_fields,
+        }
+        return self._post_json_dict("edit/profile", payload, "edit profile")
+
+    def delete_profile(
+        self,
+        profile_template_id: str,
+        user_id: str | None = None,
+        agent_id: str | None = None,
+    ) -> dict[str, Any] | None:
+        """Delete a profile instance."""
+        self._validate_required_params(profile_template_id=profile_template_id)
+        self._validate_profile_subject(user_id, agent_id)
+
+        payload = {
+            "user_id": user_id,
+            "agent_id": agent_id,
+            "profile_template_id": profile_template_id,
+        }
+        return self._post_json_dict("delete/profile", payload, "delete profile")
+
     def chat(
         self,
         user_id: str,
@@ -524,20 +733,25 @@ def chat(
         system_prompt: str | None = None,
         model_name: str | None = None,
         knowledgebase_ids: list[str] | None = None,
-        filter: dict[str:Any] | None = None,
-        add_message_on_answer: bool = False,
+        filter: dict[str, Any] | None = None,
+        add_message_on_answer: bool = True,
         app_id: str | None = None,
         agent_id: str | None = None,
         async_mode: bool = True,
         tags: list[str] | None = None,
-        info: dict[str:Any] | None = None,
+        info: dict[str, Any] | None = None,
         allow_public: bool = False,
+        allow_knowledgebase_ids: list[str] | None = None,
         max_tokens: int = 8192,
         temperature: float | None = None,
         top_p: float | None = None,
         include_preference: bool = True,
         preference_limit_number: int = 6,
         memory_limit_number: int = 6,
+        stream: bool = False,
+        include_tool_memory: bool = False,
+        tool_memory_limit_number: int = 6,
+        relativity: float | None = None,
     ) -> MemOSChatResponse | None:
         """chat"""
         # Validate required parameters
@@ -565,12 +779,17 @@ def chat(
             "tags": tags,
             "info": info,
             "allow_public": allow_public,
+            "allow_knowledgebase_ids": allow_knowledgebase_ids,
             "max_tokens": max_tokens,
             "temperature": temperature,
             "top_p": top_p,
             "include_preference": include_preference,
             "preference_limit_number": preference_limit_number,
             "memory_limit_number": memory_limit_number,
+            "stream": stream,
+            "include_tool_memory": include_tool_memory,
+            "tool_memory_limit_number": tool_memory_limit_number,
+            "relativity": relativity,
         }
 
         for retry in range(MAX_RETRY_COUNT):
diff --git a/tests/api/test_client.py b/tests/api/test_client.py
new file mode 100644
index 000000000..61724b0d4
--- /dev/null
+++ b/tests/api/test_client.py
@@ -0,0 +1,406 @@
+import json
+import sys
+import types
+
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+
+SRC_DIR = Path(__file__).resolve().parents[2] / "src" / "memos"
+
+
+def _install_memos_package_stub() -> None:
+    if "memos" not in sys.modules:
+        memos_pkg = types.ModuleType("memos")
+        memos_pkg.__path__ = [str(SRC_DIR)]
+        sys.modules["memos"] = memos_pkg
+
+    if "memos.api" not in sys.modules:
+        api_pkg = types.ModuleType("memos.api")
+        api_pkg.__path__ = [str(SRC_DIR / "api")]
+        sys.modules["memos.api"] = api_pkg
+        sys.modules["memos"].api = api_pkg
+
+
+def _load_client_module() -> Any:
+    _install_memos_package_stub()
+
+    import memos.api.client as client_module
+
+    return client_module
+
+
+class DummyResponse:
+    def __init__(self, payload: dict):
+        self.payload = payload
+
+    def raise_for_status(self) -> None:
+        return None
+
+    def json(self) -> dict:
+        return self.payload
+
+
+def _response_for(url: str) -> dict:
+    if url.endswith("/add/message"):
+        return {
+            "code": 200,
+            "message": "ok",
+            "data": {"success": True, "task_id": "task-1", "status": "completed"},
+        }
+    if url.endswith("/search/memory"):
+        return {"code": 200, "message": "ok", "data": {"memory_detail_list": []}}
+    if url.endswith("/get/memory"):
+        return {"code": 200, "message": "ok", "data": {"memory_detail_list": []}}
+    if url.endswith("/get/knowledgebase-file"):
+        return {"code": 200, "message": "ok", "data": {"file_detail_list": []}}
+    if url.endswith("/delete/memory"):
+        return {"code": 200, "message": "ok", "data": {"success": True}}
+    if url.endswith("/chat"):
+        return {"code": 200, "message": "ok", "data": {"response": "answer"}}
+    if url.endswith("/add/knowledgebase-file"):
+        return {"code": 200, "message": "ok", "data": []}
+    if url.endswith("/update/memory"):
+        return {"code": 200, "message": "ok", "data": {"success": True}}
+    if url.endswith("/extract/memory"):
+        return {
+            "code": 200,
+            "message": "ok",
+            "data": {
+                "success": True,
+                "memory_detail_list": [],
+                "preference_detail_list": [],
+            },
+        }
+    if url.endswith("/rerank"):
+        return {"code": 200, "message": "ok", "data": {"id": "rerank-1", "results": []}}
+    if url.endswith("/bind/profile_template"):
+        return {"code": 200, "message": "ok", "data": {"success": True}}
+    if url.endswith("/edit/profile"):
+        return {"code": 200, "message": "ok", "data": {"success": True}}
+    if url.endswith("/delete/profile"):
+        return {"code": 200, "message": "ok", "data": {"success": True}}
+    raise AssertionError(f"Unexpected URL: {url}")
+
+
+@pytest.fixture
+def client_module() -> Any:
+    return _load_client_module()
+
+
+@pytest.fixture
+def posted_requests(monkeypatch, client_module):
+    calls: list[dict] = []
+
+    def fake_post(url: str, **kwargs):
+        calls.append({"url": url, **kwargs})
+        return DummyResponse(_response_for(url))
+
+    monkeypatch.setattr(client_module.requests, "post", fake_post)
+    return calls
+
+
+@pytest.fixture
+def client(client_module) -> Any:
+    return client_module.MemOSClient(api_key="test-key", base_url="https://example.test/openmem/v1")
+
+
+def _json_payload(call: dict) -> dict:
+    return json.loads(call["data"])
+
+
+def test_add_message_uses_snake_case_async_mode_and_memory_view(
+    client: Any, posted_requests: list[dict]
+) -> None:
+    client.add_message(
+        messages=[{"role": "user", "content": "hello"}],
+        user_id="user-1",
+        conversation_id="conversation-1",
+        async_mode=False,
+        allow_memory_view=["kb-1"],
+    )
+
+    payload = _json_payload(posted_requests[0])
+
+    assert payload["async_mode"] is False
+    assert "asyncMode" not in payload
+    assert payload["allow_memory_view"] == ["kb-1"]
+
+
+def test_search_memory_sends_updated_existing_request_fields(
+    client: Any, posted_requests: list[dict]
+) -> None:
+    client.search_memory(
+        query="hello",
+        user_id="user-1",
+        agent_id="agent-1",
+        relativity=0.2,
+        include_skill=True,
+        skill_limit_number=4,
+        include_memory_view=["kb-1"],
+        context_format="json",
+    )
+
+    payload = _json_payload(posted_requests[0])
+
+    assert payload["conversation_id"] is None
+    assert payload["agent_id"] == "agent-1"
+    assert payload["relativity"] == 0.2
+    assert payload["include_skill"] is True
+    assert payload["skill_limit_number"] == 4
+    assert payload["include_memory_view"] == ["kb-1"]
+    assert payload["context_format"] == "json"
+
+
+def test_get_memory_can_scope_by_agent_and_include_updated_filters(
+    client: Any, posted_requests: list[dict]
+) -> None:
+    memory_filter = {"and": [{"memory_type": "LongTermMemory"}]}
+
+    client.get_memory(
+        user_id=None,
+        agent_id="agent-1",
+        include_tool_memory=False,
+        include_memory_view=["kb-1"],
+        filter=memory_filter,
+        page=2,
+        size=20,
+    )
+
+    payload = _json_payload(posted_requests[0])
+
+    assert payload["user_id"] is None
+    assert payload["agent_id"] == "agent-1"
+    assert payload["include_tool_memory"] is False
+    assert payload["include_memory_view"] == ["kb-1"]
+    assert payload["filter"] == memory_filter
+    assert payload["page"] == 2
+    assert payload["size"] == 20
+
+
+def test_get_memory_rejects_multiple_subjects(client: Any) -> None:
+    with pytest.raises(ValueError, match="exactly one of user_id or agent_id"):
+        client.get_memory(user_id="user-1", agent_id="agent-1")
+
+
+def test_get_knowledgebase_file_supports_listing_by_knowledgebase(
+    client: Any, posted_requests: list[dict]
+) -> None:
+    client.get_knowledgebase_file(
+        knowledgebase_id="kb-1",
+        type="doc",
+        page=2,
+        page_size=50,
+    )
+
+    payload = _json_payload(posted_requests[0])
+
+    assert payload == {
+        "file_ids": None,
+        "knowledgebase_id": "kb-1",
+        "type": "doc",
+        "page": 2,
+        "page_size": 50,
+    }
+
+
+def test_delete_memory_keeps_legacy_memory_id_call_but_sends_current_contract(
+    client: Any, posted_requests: list[dict]
+) -> None:
+    client.delete_memory(user_ids=["legacy-user"], memory_ids=["memory-1"])
+
+    payload = _json_payload(posted_requests[0])
+
+    assert payload == {"memory_ids": ["memory-1"]}
+
+
+def test_delete_memory_supports_quick_delete_by_user_id(
+    client: Any, posted_requests: list[dict]
+) -> None:
+    client.delete_memory(user_id="user-1")
+
+    payload = _json_payload(posted_requests[0])
+
+    assert payload == {"user_id": "user-1"}
+
+
+def test_chat_sends_updated_existing_request_fields(
+    client: Any, posted_requests: list[dict]
+) -> None:
+    client.chat(
+        user_id="user-1",
+        conversation_id="conversation-1",
+        query="hello",
+        stream=True,
+        allow_knowledgebase_ids=["kb-1"],
+        include_tool_memory=True,
+        tool_memory_limit_number=3,
+        relativity=0.1,
+    )
+
+    payload = _json_payload(posted_requests[0])
+
+    assert payload["stream"] is True
+    assert payload["allow_knowledgebase_ids"] == ["kb-1"]
+    assert payload["include_tool_memory"] is True
+    assert payload["tool_memory_limit_number"] == 3
+    assert payload["relativity"] == 0.1
+    assert payload["add_message_on_answer"] is True
+
+
+def test_add_knowledgebase_file_form_sends_type_and_closes_files(
+    client: Any, posted_requests: list[dict], tmp_path
+) -> None:
+    file_path = tmp_path / "note.txt"
+    file_path.write_text("hello", encoding="utf-8")
+
+    client.add_knowledgebase_file_form(
+        knowledgebase_id="kb-1",
+        files=[str(file_path)],
+        type="doc",
+    )
+
+    call = posted_requests[0]
+    uploaded_file = call["files"][0][1][1]
+
+    assert call["params"] == {"knowledgebase_id": "kb-1", "type": "doc"}
+    assert uploaded_file.closed
+
+
+def test_update_memory_sends_selected_fields(client: Any, posted_requests: list[dict]) -> None:
+    response = client.update_memory(
+        memory_id="memory-1",
+        content="new content",
+        title="new title",
+        status="activated",
+    )
+
+    payload = _json_payload(posted_requests[0])
+
+    assert posted_requests[0]["url"].endswith("/update/memory")
+    assert payload == {
+        "memory_id": "memory-1",
+        "content": "new content",
+        "title": "new title",
+        "status": "activated",
+    }
+    assert response["data"]["success"] is True
+
+
+def test_update_memory_requires_a_change(client: Any) -> None:
+    with pytest.raises(ValueError, match="content, title or status is required"):
+        client.update_memory(memory_id="memory-1")
+
+
+def test_extract_memory_sends_messages_and_options(
+    client: Any, posted_requests: list[dict]
+) -> None:
+    messages = [{"role": "user", "content": "I like tea", "chat_time": "2026-07-06"}]
+
+    client.extract_memory(
+        messages=messages,
+        extraction_types=["memory", "preference"],
+        model="extract-model",
+    )
+
+    payload = _json_payload(posted_requests[0])
+
+    assert posted_requests[0]["url"].endswith("/extract/memory")
+    assert payload == {
+        "messages": messages,
+        "extraction_types": ["memory", "preference"],
+        "model": "extract-model",
+    }
+
+
+def test_rerank_sends_query_documents_and_options(client: Any, posted_requests: list[dict]) -> None:
+    client.rerank(
+        query="memory query",
+        documents=["doc a", "doc b"],
+        model="rerank-model",
+        top_n=1,
+    )
+
+    payload = _json_payload(posted_requests[0])
+
+    assert posted_requests[0]["url"].endswith("/rerank")
+    assert payload == {
+        "query": "memory query",
+        "documents": ["doc a", "doc b"],
+        "model": "rerank-model",
+        "top_n": 1,
+    }
+
+
+def test_rerank_rejects_non_positive_top_n(client: Any) -> None:
+    with pytest.raises(ValueError, match="top_n must be greater than 0"):
+        client.rerank(query="memory query", documents=["doc a"], top_n=0)
+
+
+def test_bind_profile_template_sends_bind_list(client: Any, posted_requests: list[dict]) -> None:
+    bind_list = [{"profile_template_id": "profile-template-1", "user_id": "user-1"}]
+
+    client.bind_profile_template(bind_list=bind_list)
+
+    payload = _json_payload(posted_requests[0])
+
+    assert posted_requests[0]["url"].endswith("/bind/profile_template")
+    assert payload == {"bind_list": bind_list}
+
+
+def test_edit_profile_sends_metadata_and_remove_fields(
+    client: Any, posted_requests: list[dict]
+) -> None:
+    metadata = {"basic": {"city": "Hangzhou"}}
+
+    client.edit_profile(
+        profile_template_id="profile-template-1",
+        user_id="user-1",
+        metadata=metadata,
+        remove_fields=["basic.job"],
+    )
+
+    payload = _json_payload(posted_requests[0])
+
+    assert posted_requests[0]["url"].endswith("/edit/profile")
+    assert payload == {
+        "user_id": "user-1",
+        "agent_id": None,
+        "profile_template_id": "profile-template-1",
+        "metadata": metadata,
+        "remove_fields": ["basic.job"],
+    }
+
+
+def test_edit_profile_requires_metadata_or_remove_fields(client: Any) -> None:
+    with pytest.raises(ValueError, match="metadata or remove_fields is required"):
+        client.edit_profile(profile_template_id="profile-template-1", user_id="user-1")
+
+
+def test_delete_profile_sends_profile_template_and_subject(
+    client: Any, posted_requests: list[dict]
+) -> None:
+    client.delete_profile(profile_template_id="profile-template-1", agent_id="agent-1")
+
+    payload = _json_payload(posted_requests[0])
+
+    assert posted_requests[0]["url"].endswith("/delete/profile")
+    assert payload == {
+        "user_id": None,
+        "agent_id": "agent-1",
+        "profile_template_id": "profile-template-1",
+    }
+
+
+def test_profile_subject_requires_exactly_one_user_or_agent(client: Any) -> None:
+    with pytest.raises(ValueError, match="exactly one of user_id or agent_id is required"):
+        client.delete_profile(profile_template_id="profile-template-1")
+
+    with pytest.raises(ValueError, match="exactly one of user_id or agent_id is required"):
+        client.delete_profile(
+            profile_template_id="profile-template-1",
+            user_id="user-1",
+            agent_id="agent-1",
+        )

From f46c02ed7b9199e812edc03cd3f4fd7dae459dc9 Mon Sep 17 00:00:00 2001
From: dingliang <2650876010@qq.com>
Date: Tue, 14 Jul 2026 17:18:25 +0800
Subject: [PATCH 02/13] feat(api): align OpenMem v1 SDK with cloud API

---
 src/memos/api/client.py         |  89 +++++++---
 src/memos/api/product_models.py |  46 +++++-
 tests/api/test_client.py        | 282 +++++++++++++++++++++++++++++++-
 3 files changed, 387 insertions(+), 30 deletions(-)

diff --git a/src/memos/api/client.py b/src/memos/api/client.py
index 1ace6ec8f..b8055b5b6 100644
--- a/src/memos/api/client.py
+++ b/src/memos/api/client.py
@@ -2,7 +2,9 @@
 import mimetypes
 import os
 
+from collections.abc import Iterator
 from typing import Any
+from urllib.parse import quote
 
 import requests
 
@@ -95,13 +97,13 @@ def get_message(
         self,
         user_id: str,
         conversation_id: str | None = None,
-        conversation_limit_number: int = 6,
-        message_limit_number: int = 6,
+        conversation_limit_number: int | None = None,
+        message_limit_number: int | None = None,
         source: str | None = None,
     ) -> MemOSGetMessagesResponse | None:
         """Get message"""
         # Validate required parameters
-        self._validate_required_params(user_id=user_id)
+        self._validate_required_params(user_id=user_id, conversation_id=conversation_id)
 
         url = f"{self.base_url}/get/message"
         payload = {
@@ -128,12 +130,12 @@ def get_message(
     def add_message(
         self,
         messages: list[dict[str, Any]],
-        user_id: str | list[str],
-        conversation_id: str,
+        user_id: str | list[str] | None = None,
+        conversation_id: str | None = None,
         info: dict[str, Any] | None = None,
         source: str | None = None,
         app_id: str | None = None,
-        agent_id: str | None = None,
+        agent_id: str | list[str] | None = None,
         async_mode: bool = True,
         tags: list[str] | None = None,
         allow_public: bool = False,
@@ -142,9 +144,9 @@ def add_message(
     ) -> MemOSAddResponse | None:
         """Add message"""
         # Validate required parameters
-        self._validate_required_params(
-            messages=messages, user_id=user_id, conversation_id=conversation_id
-        )
+        self._validate_required_params(messages=messages)
+        if not user_id and not agent_id:
+            raise ValueError("user_id or agent_id is required")
 
         url = f"{self.base_url}/add/message"
         payload = {
@@ -178,7 +180,7 @@ def add_message(
     def search_memory(
         self,
         query: str,
-        user_id: str,
+        user_id: str | None = None,
         conversation_id: str | None = None,
         agent_id: str | None = None,
         memory_limit_number: int = 6,
@@ -197,7 +199,8 @@ def search_memory(
     ) -> MemOSSearchResponse | None:
         """Search memories"""
         # Validate required parameters
-        self._validate_required_params(query=query, user_id=user_id)
+        self._validate_required_params(query=query)
+        self._validate_profile_subject(user_id, agent_id)
 
         url = f"{self.base_url}/search/memory"
         payload = {
@@ -248,6 +251,8 @@ def get_memory(
         """get memories"""
         # Validate required parameters
         self._validate_profile_subject(user_id, agent_id)
+        if size > 50:
+            raise ValueError("size must be less than or equal to 50")
 
         url = f"{self.base_url}/get/memory"
         payload = {
@@ -275,17 +280,47 @@ def get_memory(
                 if retry == MAX_RETRY_COUNT - 1:
                     raise
 
+    @staticmethod
+    def _iter_sse_data(response: requests.Response) -> Iterator[str]:
+        """Yield decoded data payloads from a Server-Sent Events response."""
+        try:
+            for line in response.iter_lines(decode_unicode=True):
+                if isinstance(line, bytes):
+                    line = line.decode("utf-8")
+                if not line or not line.startswith("data:"):
+                    continue
+                yield line.removeprefix("data:").lstrip()
+        finally:
+            response.close()
+
+    def get_memory_by_id(self, memid: str) -> dict[str, Any] | None:
+        """Get one memory detail by its memory ID."""
+        self._validate_required_params(memid=memid)
+
+        url = f"{self.base_url}/get/memory/{quote(memid, safe='')}"
+        for retry in range(MAX_RETRY_COUNT):
+            try:
+                response = requests.get(url, headers=self.headers, timeout=30)
+                response.raise_for_status()
+                return response.json()
+            except Exception as e:
+                logger.error(
+                    "Failed to get memory by ID (retry %s/%s): %s",
+                    retry + 1,
+                    MAX_RETRY_COUNT,
+                    e,
+                )
+                if retry == MAX_RETRY_COUNT - 1:
+                    raise
+
     def create_knowledgebase(
-        self, knowledgebase_name: str, knowledgebase_description: str
+        self, knowledgebase_name: str, knowledgebase_description: str | None = None
     ) -> MemOSCreateKnowledgebaseResponse | None:
         """
         Create knowledgebase
         """
         # Validate required parameters
-        self._validate_required_params(
-            knowledgebase_name=knowledgebase_name,
-            knowledgebase_description=knowledgebase_description,
-        )
+        self._validate_required_params(knowledgebase_name=knowledgebase_name)
 
         url = f"{self.base_url}/create/knowledgebase"
         payload = {
@@ -524,8 +559,8 @@ def get_task_status(self, task_id: str) -> MemOSGetTaskStatusResponse | None:
     def add_feedback(
         self,
         user_id: str,
-        conversation_id: str,
-        feedback_content: str,
+        conversation_id: str | None = None,
+        feedback_content: str | None = None,
         agent_id: str | None = None,
         app_id: str | None = None,
         feedback_time: str | None = None,
@@ -534,9 +569,7 @@ def add_feedback(
     ) -> MemOSAddFeedBackResponse | None:
         """Add feedback"""
         # Validate required parameters
-        self._validate_required_params(
-            feedback_content=feedback_content, user_id=user_id, conversation_id=conversation_id
-        )
+        self._validate_required_params(feedback_content=feedback_content, user_id=user_id)
 
         url = f"{self.base_url}/add/feedback"
         payload = {
@@ -743,8 +776,8 @@ def chat(
         allow_public: bool = False,
         allow_knowledgebase_ids: list[str] | None = None,
         max_tokens: int = 8192,
-        temperature: float | None = None,
-        top_p: float | None = None,
+        temperature: float | None = 0.7,
+        top_p: float | None = 0.95,
         include_preference: bool = True,
         preference_limit_number: int = 6,
         memory_limit_number: int = 6,
@@ -752,7 +785,7 @@ def chat(
         include_tool_memory: bool = False,
         tool_memory_limit_number: int = 6,
         relativity: float | None = None,
-    ) -> MemOSChatResponse | None:
+    ) -> MemOSChatResponse | Iterator[str] | None:
         """chat"""
         # Validate required parameters
         self._validate_required_params(
@@ -795,9 +828,15 @@ def chat(
         for retry in range(MAX_RETRY_COUNT):
             try:
                 response = requests.post(
-                    url, data=json.dumps(payload), headers=self.headers, timeout=30
+                    url,
+                    data=json.dumps(payload),
+                    headers=self.headers,
+                    timeout=30,
+                    stream=stream,
                 )
                 response.raise_for_status()
+                if stream:
+                    return self._iter_sse_data(response)
                 response_data = response.json()
 
                 return MemOSChatResponse(**response_data)
diff --git a/src/memos/api/product_models.py b/src/memos/api/product_models.py
index ff9d94859..86f6fb80c 100644
--- a/src/memos/api/product_models.py
+++ b/src/memos/api/product_models.py
@@ -1048,6 +1048,15 @@ class SearchMemoryData(BaseModel):
         alias="tool_memory_detail_list",
         description="List of tool_memor details (usually None)",
     )
+    skill_detail_list: list[MemoryDetail] | None = Field(
+        None, alias="skill_detail_list", description="List of skill memory details"
+    )
+    profile_detail_list: list[MemoryDetail] | None = Field(
+        None, alias="profile_detail_list", description="List of profile memory details"
+    )
+    event_detail_list: list[MemoryDetail] | None = Field(
+        None, alias="event_detail_list", description="List of event memory details"
+    )
     preference_note: str = Field(
         None, alias="preference_note", description="String of preference_note"
     )
@@ -1059,6 +1068,9 @@ class GetKnowledgebaseFileData(BaseModel):
     file_detail_list: list[FileDetail] = Field(
         default_factory=list, alias="file_detail_list", description="List of files details"
     )
+    total: int | None = Field(None, description="Total number of matching files")
+    page: int | None = Field(None, description="Current page number")
+    page_size: int | None = Field(None, alias="page_size", description="Page size")
 
 
 class GetMemoryData(BaseModel):
@@ -1070,6 +1082,22 @@ class GetMemoryData(BaseModel):
     preference_detail_list: list[MessageDetail] | None = Field(
         None, alias="preference_detail_list", description="List of preference detail"
     )
+    tool_memory_detail_list: list[MemoryDetail] | None = Field(
+        None, alias="tool_memory_detail_list", description="List of tool memory details"
+    )
+    profile_detail_list: list[MemoryDetail] | None = Field(
+        None, alias="profile_detail_list", description="List of profile memory details"
+    )
+    event_detail_list: list[MemoryDetail] | None = Field(
+        None, alias="event_detail_list", description="List of event memory details"
+    )
+    skill_detail_list: list[MemoryDetail] | None = Field(
+        None, alias="skill_detail_list", description="List of skill memory details"
+    )
+    total: int | None = Field(None, description="Total number of memories")
+    size: int | None = Field(None, description="Page size")
+    current: int | None = Field(None, description="Current page number")
+    pages: int | None = Field(None, description="Total number of pages")
 
 
 class AddMessageData(BaseModel):
@@ -1098,6 +1126,16 @@ class GetTaskStatusMessageData(BaseModel):
     status: str = Field(..., description="Operation task status")
 
 
+class GetTaskStatusData(BaseModel):
+    """Current OpenMem task status response data."""
+
+    task_id: str = Field(..., description="Task identifier")
+    status: str = Field(..., description="Operation task status")
+    memory_views: dict[str, Any] | None = Field(
+        None, alias="memory_views", description="Memory view changes produced by the task"
+    )
+
+
 # ─── MemOS Response Models (Similar to OpenAI ChatCompletion) ──────────────────
 
 
@@ -1181,12 +1219,12 @@ class MemOSGetTaskStatusResponse(BaseModel):
 
     code: int = Field(..., description="Response status code")
     message: str = Field(..., description="Response message")
-    data: list[GetTaskStatusMessageData] = Field(..., description="Task status data")
+    data: GetTaskStatusData = Field(..., description="Task status data")
 
     @property
-    def messages(self) -> list[GetTaskStatusMessageData]:
-        """Convenient access to task status messages."""
-        return self.data
+    def messages(self) -> list[GetTaskStatusData]:
+        """Backward-compatible list access to task status data."""
+        return [self.data]
 
 
 class MemOSCreateKnowledgebaseResponse(BaseModel):
diff --git a/tests/api/test_client.py b/tests/api/test_client.py
index 61724b0d4..2e911e4f4 100644
--- a/tests/api/test_client.py
+++ b/tests/api/test_client.py
@@ -43,7 +43,30 @@ def json(self) -> dict:
         return self.payload
 
 
+class DummyStreamResponse:
+    def __init__(self, lines: list[str]):
+        self.lines = lines
+        self.closed = False
+        self.json_called = False
+
+    def raise_for_status(self) -> None:
+        return None
+
+    def json(self) -> dict:
+        self.json_called = True
+        raise AssertionError("streaming responses must not be parsed as JSON")
+
+    def iter_lines(self, decode_unicode: bool = False):
+        assert decode_unicode is True
+        yield from self.lines
+
+    def close(self) -> None:
+        self.closed = True
+
+
 def _response_for(url: str) -> dict:
+    if url.endswith("/get/message"):
+        return {"code": 200, "message": "ok", "data": {"message_detail_list": []}}
     if url.endswith("/add/message"):
         return {
             "code": 200,
@@ -54,10 +77,18 @@ def _response_for(url: str) -> dict:
         return {"code": 200, "message": "ok", "data": {"memory_detail_list": []}}
     if url.endswith("/get/memory"):
         return {"code": 200, "message": "ok", "data": {"memory_detail_list": []}}
+    if url.endswith("/create/knowledgebase"):
+        return {"code": 200, "message": "ok", "data": {"id": "kb-1"}}
     if url.endswith("/get/knowledgebase-file"):
         return {"code": 200, "message": "ok", "data": {"file_detail_list": []}}
     if url.endswith("/delete/memory"):
         return {"code": 200, "message": "ok", "data": {"success": True}}
+    if url.endswith("/add/feedback"):
+        return {
+            "code": 200,
+            "message": "ok",
+            "data": {"success": True, "task_id": "task-1", "status": "running"},
+        }
     if url.endswith("/chat"):
         return {"code": 200, "message": "ok", "data": {"response": "answer"}}
     if url.endswith("/add/knowledgebase-file"):
@@ -102,6 +133,24 @@ def fake_post(url: str, **kwargs):
     return calls
 
 
+@pytest.fixture
+def fetched_requests(monkeypatch, client_module):
+    calls: list[dict] = []
+
+    def fake_get(url: str, **kwargs):
+        calls.append({"url": url, **kwargs})
+        return DummyResponse(
+            {
+                "code": 200,
+                "message": "ok",
+                "data": {"id": "memory-1", "memory_type": "LongTermMemory"},
+            }
+        )
+
+    monkeypatch.setattr(client_module.requests, "get", fake_get)
+    return calls
+
+
 @pytest.fixture
 def client(client_module) -> Any:
     return client_module.MemOSClient(api_key="test-key", base_url="https://example.test/openmem/v1")
@@ -134,7 +183,7 @@ def test_search_memory_sends_updated_existing_request_fields(
 ) -> None:
     client.search_memory(
         query="hello",
-        user_id="user-1",
+        user_id=None,
         agent_id="agent-1",
         relativity=0.2,
         include_skill=True,
@@ -404,3 +453,234 @@ def test_profile_subject_requires_exactly_one_user_or_agent(client: Any) -> None
             user_id="user-1",
             agent_id="agent-1",
         )
+
+
+def test_task_status_response_parses_current_object_shape(client_module: Any) -> None:
+    response = client_module.MemOSGetTaskStatusResponse(
+        code=200,
+        message="ok",
+        data={
+            "task_id": "task-1",
+            "status": "running",
+            "memory_views": {"added": 1},
+        },
+    )
+
+    assert response.data.task_id == "task-1"
+    assert response.data.status == "running"
+    assert response.data.memory_views == {"added": 1}
+
+
+def test_search_response_keeps_all_current_memory_view_lists(client_module: Any) -> None:
+    response = client_module.MemOSSearchResponse(
+        code=200,
+        message="ok",
+        data={
+            "memory_detail_list": [],
+            "skill_detail_list": [{"id": "skill-1"}],
+            "profile_detail_list": [{"id": "profile-1"}],
+            "event_detail_list": [{"id": "event-1"}],
+        },
+    )
+
+    assert response.data.skill_detail_list[0].id == "skill-1"
+    assert response.data.profile_detail_list[0].id == "profile-1"
+    assert response.data.event_detail_list[0].id == "event-1"
+
+
+def test_get_memory_response_keeps_views_and_pagination(client_module: Any) -> None:
+    response = client_module.MemOSGetMemoryResponse(
+        code=200,
+        message="ok",
+        data={
+            "memory_detail_list": [],
+            "tool_memory_detail_list": [{"id": "tool-1"}],
+            "profile_detail_list": [{"id": "profile-1"}],
+            "event_detail_list": [{"id": "event-1"}],
+            "skill_detail_list": [{"id": "skill-1"}],
+            "total": 21,
+            "size": 10,
+            "current": 2,
+            "pages": 3,
+        },
+    )
+
+    assert response.data.tool_memory_detail_list[0].id == "tool-1"
+    assert response.data.profile_detail_list[0].id == "profile-1"
+    assert response.data.event_detail_list[0].id == "event-1"
+    assert response.data.skill_detail_list[0].id == "skill-1"
+    assert response.data.total == 21
+    assert response.data.size == 10
+    assert response.data.current == 2
+    assert response.data.pages == 3
+
+
+def test_get_knowledgebase_file_response_keeps_pagination(client_module: Any) -> None:
+    response = client_module.MemOSGetKnowledgebaseFileResponse(
+        code=200,
+        message="ok",
+        data={
+            "file_detail_list": [],
+            "total": 8,
+            "page": 2,
+            "page_size": 5,
+        },
+    )
+
+    assert response.data.total == 8
+    assert response.data.page == 2
+    assert response.data.page_size == 5
+
+
+def test_get_message_requires_conversation_id(client: Any, posted_requests: list[dict]) -> None:
+    with pytest.raises(ValueError, match="conversation_id is required"):
+        client.get_message(user_id="user-1")
+
+    assert posted_requests == []
+
+
+def test_get_message_uses_playground_default_limits(
+    client: Any, posted_requests: list[dict]
+) -> None:
+    client.get_message(user_id="user-1", conversation_id="conversation-1")
+
+    payload = _json_payload(posted_requests[0])
+
+    assert payload["conversation_limit_number"] is None
+    assert payload["message_limit_number"] is None
+
+
+def test_add_message_allows_agent_only_and_generated_conversation(
+    client: Any, posted_requests: list[dict]
+) -> None:
+    client.add_message(
+        messages=[{"role": "user", "content": "hello"}],
+        user_id=None,
+        agent_id="agent-1",
+        conversation_id=None,
+    )
+
+    payload = _json_payload(posted_requests[0])
+
+    assert payload["user_id"] is None
+    assert payload["agent_id"] == "agent-1"
+    assert payload["conversation_id"] is None
+
+
+def test_search_memory_allows_agent_only(client: Any, posted_requests: list[dict]) -> None:
+    client.search_memory(query="hello", user_id=None, agent_id="agent-1")
+
+    payload = _json_payload(posted_requests[0])
+
+    assert payload["user_id"] is None
+    assert payload["agent_id"] == "agent-1"
+
+
+def test_search_memory_rejects_multiple_subjects(client: Any, posted_requests: list[dict]) -> None:
+    with pytest.raises(ValueError, match="exactly one of user_id or agent_id"):
+        client.search_memory(query="hello", user_id="user-1", agent_id="agent-1")
+
+    assert posted_requests == []
+
+
+def test_create_knowledgebase_allows_empty_description(
+    client: Any, posted_requests: list[dict]
+) -> None:
+    client.create_knowledgebase(knowledgebase_name="Knowledge Base")
+
+    payload = _json_payload(posted_requests[0])
+
+    assert payload == {
+        "knowledgebase_name": "Knowledge Base",
+        "knowledgebase_description": None,
+    }
+
+
+def test_add_feedback_allows_generated_conversation(
+    client: Any, posted_requests: list[dict]
+) -> None:
+    client.add_feedback(user_id="user-1", feedback_content="helpful")
+
+    payload = _json_payload(posted_requests[0])
+
+    assert payload["conversation_id"] is None
+    assert payload["feedback_content"] == "helpful"
+
+
+def test_chat_uses_playground_sampling_defaults(client: Any, posted_requests: list[dict]) -> None:
+    client.chat(user_id="user-1", conversation_id="conversation-1", query="hello")
+
+    payload = _json_payload(posted_requests[0])
+
+    assert payload["temperature"] == 0.7
+    assert payload["top_p"] == 0.95
+
+
+def test_get_memory_rejects_size_above_playground_limit(
+    client: Any, posted_requests: list[dict]
+) -> None:
+    with pytest.raises(ValueError, match="size must be less than or equal to 50"):
+        client.get_memory(user_id="user-1", size=51)
+
+    assert posted_requests == []
+
+
+def test_get_memory_by_id_uses_detail_get_endpoint(
+    client: Any, fetched_requests: list[dict]
+) -> None:
+    response = client.get_memory_by_id("memory-1")
+
+    assert fetched_requests == [
+        {
+            "url": "https://example.test/openmem/v1/get/memory/memory-1",
+            "headers": client.headers,
+            "timeout": 30,
+        }
+    ]
+    assert response == {
+        "code": 200,
+        "message": "ok",
+        "data": {"id": "memory-1", "memory_type": "LongTermMemory"},
+    }
+
+
+def test_get_memory_by_id_requires_memid(client: Any, fetched_requests: list[dict]) -> None:
+    with pytest.raises(ValueError, match="memid is required"):
+        client.get_memory_by_id("")
+
+    assert fetched_requests == []
+
+
+def test_chat_stream_yields_sse_data_and_closes_response(monkeypatch, client_module: Any) -> None:
+    calls: list[dict] = []
+    stream_response = DummyStreamResponse(
+        [
+            "event: message",
+            'data: {"response":"first"}',
+            "",
+            "data: [DONE]",
+        ]
+    )
+
+    def fake_post(url: str, **kwargs):
+        calls.append({"url": url, **kwargs})
+        return stream_response
+
+    monkeypatch.setattr(client_module.requests, "post", fake_post)
+    client = client_module.MemOSClient(
+        api_key="test-key", base_url="https://example.test/openmem/v1"
+    )
+
+    chunks = list(
+        client.chat(
+            user_id="user-1",
+            conversation_id="conversation-1",
+            query="hello",
+            stream=True,
+        )
+    )
+
+    assert calls[0]["stream"] is True
+    assert chunks == ['{"response":"first"}', "[DONE]"]
+    assert stream_response.json_called is False
+    assert stream_response.closed is True

From 9d2c9b75d2f462e8e05f7750e2266d7eea0a7a48 Mon Sep 17 00:00:00 2001
From: de1ty <7804799+de1tydev@users.noreply.github.com>
Date: Tue, 14 Jul 2026 22:27:08 +0800
Subject: [PATCH 03/13] fix(memos-local): include json hint in user messages
 (#1756)

Co-authored-by: shinetata <149466187+shinetata@users.noreply.github.com>
---
 apps/memos-local-plugin/core/llm/client.ts    | 19 +++++++++++++++++--
 .../tests/unit/llm/client.test.ts             |  8 ++++++--
 2 files changed, 23 insertions(+), 4 deletions(-)

diff --git a/apps/memos-local-plugin/core/llm/client.ts b/apps/memos-local-plugin/core/llm/client.ts
index 41f31a439..6bedafa70 100644
--- a/apps/memos-local-plugin/core/llm/client.ts
+++ b/apps/memos-local-plugin/core/llm/client.ts
@@ -260,6 +260,21 @@ export function createLlmClientWithProvider(
     return [{ role: "system", content: systemInsert }, ...messages];
   }
 
+  function ensureJsonWordInUserMessage(messages: LlmMessage[]): LlmMessage[] {
+    const lastUserIdx = messages.map((m) => m.role).lastIndexOf("user");
+    if (lastUserIdx < 0) return [...messages, { role: "user", content: "Return valid json only." }];
+
+    const msg = messages[lastUserIdx];
+    if (/\bjson\b/i.test(msg.content)) return messages;
+
+    const out = messages.slice();
+    out[lastUserIdx] = {
+      ...msg,
+      content: `${msg.content}\n\nReturn valid json only.`,
+    };
+    return out;
+  }
+
   function buildCallInput(opts: LlmCallOptions | undefined, jsonMode: boolean): ProviderCallInput {
     return {
       temperature: opts?.temperature ?? config.temperature,
@@ -463,7 +478,7 @@ export function createLlmClientWithProvider(
   ): Promise<LlmCompletion> {
     const messages = normalizeMessages(input);
     const msgsWithJsonHint = opts?.jsonMode
-      ? inject(messages, buildJsonSystemHint())
+      ? ensureJsonWordInUserMessage(inject(messages, buildJsonSystemHint()))
       : messages;
     const call = buildCallInput(opts, opts?.jsonMode === true);
     const { completion } = await callWithFallback(msgsWithJsonHint, call, opts, opts?.op ?? "complete");
@@ -476,7 +491,7 @@ export function createLlmClientWithProvider(
   ): Promise<LlmJsonCompletion<T>> {
     const messages = normalizeMessages(input);
     const systemHint = buildJsonSystemHint(opts.schemaHint);
-    const msgs = inject(messages, systemHint);
+    const msgs = ensureJsonWordInUserMessage(inject(messages, systemHint));
     const call = buildCallInput(opts, true);
     const op = opts.op ?? "complete.json";
     const maxMalformedRetries = Math.max(0, opts.malformedRetries ?? 1);
diff --git a/apps/memos-local-plugin/tests/unit/llm/client.test.ts b/apps/memos-local-plugin/tests/unit/llm/client.test.ts
index 7e904a2c6..dee0de228 100644
--- a/apps/memos-local-plugin/tests/unit/llm/client.test.ts
+++ b/apps/memos-local-plugin/tests/unit/llm/client.test.ts
@@ -96,12 +96,14 @@ describe("llm/client", () => {
     expect(fake.lastMessages).toEqual([{ role: "user", content: "hi there" }]);
   });
 
-  it("injects a json system hint when jsonMode=true", async () => {
+  it("injects json hints into system and user messages when jsonMode=true", async () => {
     const fake = new FakeProvider("openai_compatible", () => ({ text: '{"ok":1}', durationMs: 1 }));
     const client = createLlmClientWithProvider(cfg(), fake);
     await client.complete("do it", { jsonMode: true });
     expect(fake.lastMessages?.[0]?.role).toBe("system");
     expect(fake.lastMessages?.[0]?.content).toMatch(/single valid JSON value/i);
+    expect(fake.lastMessages?.at(-1)?.role).toBe("user");
+    expect(fake.lastMessages?.at(-1)?.content).toMatch(/valid json only/i);
     expect(fake.lastInput?.jsonMode).toBe(true);
   });
 
@@ -270,7 +272,9 @@ describe("llm/client", () => {
     expect(fake.lastMessages?.[0]?.role).toBe("system");
     expect(fake.lastMessages?.[0]?.content).toMatch(/You are strict\./);
     expect(fake.lastMessages?.[0]?.content).toMatch(/single valid JSON value/);
-    expect(fake.lastMessages?.[1]).toEqual({ role: "user", content: "go" });
+    expect(fake.lastMessages?.[1]?.role).toBe("user");
+    expect(fake.lastMessages?.[1]?.content).toMatch(/^go/);
+    expect(fake.lastMessages?.[1]?.content).toMatch(/valid json only/i);
   });
 
   it("rejects empty messages array", async () => {

From a000a1c7341fdfbe3d13ac091b83e5aec4cf9abd Mon Sep 17 00:00:00 2001
From: shinetata <149466187+shinetata@users.noreply.github.com>
Date: Thu, 16 Jul 2026 14:35:44 +0800
Subject: [PATCH 04/13] docs: fix scheduler API examples to match server
 endpoints (#2113)

Rewrite scheduler Quick Start examples to call the real /product/scheduler REST endpoints with requests instead of non-existent MemOSClient methods; align response fields with the actual handler output; rename the mis-named ' wait.md' to 'wait.md'.

Closes #2083

Co-authored-by: sunqi <sunqi@memtensor.com>
Co-authored-by: Cursor <cursoragent@cursor.com>
---
 .../open_source_api/scheduler/get_status.md   | 48 ++++++++++-------
 .../scheduler/{ wait.md => wait.md}           | 51 +++++++++++--------
 .../open_source_api/scheduler/get_status.md   | 48 ++++++++++-------
 .../open_source_api/scheduler/wait.md         | 51 +++++++++++--------
 4 files changed, 124 insertions(+), 74 deletions(-)
 rename docs/cn/open_source/open_source_api/scheduler/{ wait.md => wait.md} (61%)

diff --git a/docs/cn/open_source/open_source_api/scheduler/get_status.md b/docs/cn/open_source/open_source_api/scheduler/get_status.md
index 87e1a4a5a..0a60d6fac 100644
--- a/docs/cn/open_source/open_source_api/scheduler/get_status.md
+++ b/docs/cn/open_source/open_source_api/scheduler/get_status.md
@@ -64,34 +64,48 @@ desc: 监控 MemOS 异步任务的生命周期，提供包括任务进度、队
 
 ## 4. 快速上手示例
 
-使用 SDK 轮询任务状态直至完成：
+这些接口由开源版 Server（`server_api`，路由前缀 `/product`）直接提供，使用标准 HTTP 请求即可访问。以下示例轮询任务状态直至完成：
 
 ```python
-from memos.api.client import MemOSClient
 import time
 
-client = MemOSClient(api_key="...", base_url="...")
+import requests
+
+# 自部署 MemOS Server 的地址（如启用了鉴权，请自行补充 Authorization 请求头）
+base_url = "http://localhost:8000"
 
 # 1. 系统级概览：查看整个 MemOS 系统的运行健康度
-global_res = client.get_all_scheduler_status()
-if global_res:
-    print(f"系统运行概况: {global_res.data['scheduler_summary']}")
+resp = requests.get(f"{base_url}/product/scheduler/allstatus", timeout=10)
+resp.raise_for_status()
+global_res = resp.json()
+print(f"系统运行概况: {global_res['data']['scheduler_summary']}")
 
 # 2. 队列指标监控：检查特定用户的任务积压情况
-queue_res = client.get_task_queue_status(user_id="dev_user_01")
-if queue_res:
-    print(f"待处理任务数: {queue_res.data['remaining_tasks_count']}")
-    print(f"已下发未完成任务数: {queue_res.data['pending_tasks_count']}")
+resp = requests.get(
+    f"{base_url}/product/scheduler/task_queue_status",
+    params={"user_id": "dev_user_01"},
+    timeout=10,
+)
+resp.raise_for_status()
+queue_res = resp.json()
+print(f"排队中任务数: {queue_res['data']['remaining_tasks_count']}")
+print(f"已下发未确认任务数: {queue_res['data']['pending_tasks_count']}")
 
 # 3. 任务进度追踪：轮询特定任务直至结束
 task_id = "task_888999"
+active_states = {"waiting", "pending", "in_progress"}
 while True:
-    res = client.get_task_status(user_id="dev_user_01", task_id=task_id)
-    if res and res.code == 200:
-        current_status = res.data[0]['status'] # data 为状态列表
-        print(f"任务 {task_id} 当前状态: {current_status}")
-
-        if current_status in ['completed', 'failed', 'cancelled']:
-            break
+    resp = requests.get(
+        f"{base_url}/product/scheduler/status",
+        params={"user_id": "dev_user_01", "task_id": task_id},
+        timeout=10,
+    )
+    resp.raise_for_status()
+    items = resp.json().get("data", [])  # data 为状态列表：[{"task_id": ..., "status": ...}]
+    statuses = {item["status"] for item in items}
+    print(f"任务 {task_id} 当前状态: {statuses or '空'}")
+
+    if not statuses or statuses.isdisjoint(active_states):
+        break
     time.sleep(2)
 ```
diff --git a/docs/cn/open_source/open_source_api/scheduler/ wait.md b/docs/cn/open_source/open_source_api/scheduler/wait.md
similarity index 61%
rename from docs/cn/open_source/open_source_api/scheduler/ wait.md
rename to docs/cn/open_source/open_source_api/scheduler/wait.md
index 9849ffe68..52b0d79f6 100644
--- a/docs/cn/open_source/open_source_api/scheduler/ wait.md	
+++ b/docs/cn/open_source/open_source_api/scheduler/wait.md
@@ -42,36 +42,47 @@ desc: 提供阻塞等待与流式进度观测能力，确保在执行后续操
 
 ## 4. 快速上手示例
 
-使用开源版 SDK 进行阻塞式等待：
+这些接口由开源版 Server（`server_api`，路由前缀 `/product`）直接提供，使用标准 HTTP 请求即可访问。注意：`user_name`、`timeout_seconds`、`poll_interval` 均为查询参数（Query），而非请求体（Body）。以下示例进行阻塞式等待：
 
 ```python
-from memos.api.client import MemOSClient
+import json
 
-client = MemOSClient(api_key="...", base_url="...")
+import requests
+
+# 自部署 MemOS Server 的地址（如启用了鉴权，请自行补充 Authorization 请求头）
+base_url = "http://localhost:8000"
 user_name = "dev_user_01"
 
 # --- 场景 A：同步阻塞等待 (常用于 Python 自动化脚本) ---
 print(f"正在等待用户 {user_name} 的任务队列清空...")
-res = client.wait_until_idle(
-    user_name=user_name,
-    timeout_seconds=300,
-    poll_interval=2
+resp = requests.post(
+    f"{base_url}/product/scheduler/wait",
+    params={"user_name": user_name, "timeout_seconds": 300, "poll_interval": 2},
+    timeout=310,  # HTTP 超时应大于 timeout_seconds
 )
-if res and res.code == 200:
+resp.raise_for_status()
+result = resp.json()  # {"message": "idle" | "timeout", "data": {...}}
+if result["message"] == "idle":
     print("✅ 任务已全部完成。")
+else:
+    print(f"⚠️ 等待超时，仍有 {result['data']['running_tasks']} 个任务在执行。")
 
 # --- 场景 B：流式进度观测 (常用于前端进度条渲染) ---
 print("开始监听任务实时进度流...")
-# 注意：SSE 接口在 SDK 中通常返回一个生成器 (Generator)
-progress_stream = client.stream_scheduler_progress(
-    user_name=user_name,
-    timeout_seconds=300
-)
-
-for event in progress_stream:
-    # 实时打印剩余任务数
-    print(f"当前排队任务数: {event['remaining_tasks_count']}")
-    if event['status'] == 'idle':
-        print("🎉 调度器已空闲")
-        break
+with requests.get(
+    f"{base_url}/product/scheduler/wait/stream",
+    params={"user_name": user_name, "timeout_seconds": 300},
+    stream=True,
+    timeout=310,
+) as resp:
+    resp.raise_for_status()
+    for line in resp.iter_lines(decode_unicode=True):
+        if not line or not line.startswith("data:"):
+            continue
+        event = json.loads(line.removeprefix("data:").strip())
+        # 实时打印仍在执行的任务数
+        print(f"当前活跃任务数: {event['active_tasks']}，状态: {event['status']}")
+        if event["status"] in ("idle", "timeout"):
+            print("🎉 调度器已空闲" if event["status"] == "idle" else "⚠️ 监听超时")
+            break
 ```
diff --git a/docs/en/open_source/open_source_api/scheduler/get_status.md b/docs/en/open_source/open_source_api/scheduler/get_status.md
index f2014d9e5..7d1565858 100644
--- a/docs/en/open_source/open_source_api/scheduler/get_status.md
+++ b/docs/en/open_source/open_source_api/scheduler/get_status.md
@@ -66,34 +66,48 @@ When you send a status request, **SchedulerHandler** performs the following oper
 
 ## 4. Quick Start
 
-Poll task status with the SDK until completion:
+These endpoints are served directly by the open-source Server (`server_api`, router prefix `/product`) and can be called with plain HTTP requests. The example below polls task status until completion:
 
 ```python
-from memos.api.client import MemOSClient
 import time
 
-client = MemOSClient(api_key="...", base_url="...")
+import requests
+
+# Address of your self-hosted MemOS Server (add an Authorization header if auth is enabled)
+base_url = "http://localhost:8000"
 
 # 1. System overview: inspect overall MemOS health.
-global_res = client.get_all_scheduler_status()
-if global_res:
-    print(f"System summary: {global_res.data['scheduler_summary']}")
+resp = requests.get(f"{base_url}/product/scheduler/allstatus", timeout=10)
+resp.raise_for_status()
+global_res = resp.json()
+print(f"System summary: {global_res['data']['scheduler_summary']}")
 
 # 2. Queue metrics: inspect backlog for a specific user.
-queue_res = client.get_task_queue_status(user_id="dev_user_01")
-if queue_res:
-    print(f"Remaining tasks: {queue_res.data['remaining_tasks_count']}")
-    print(f"Pending tasks: {queue_res.data['pending_tasks_count']}")
+resp = requests.get(
+    f"{base_url}/product/scheduler/task_queue_status",
+    params={"user_id": "dev_user_01"},
+    timeout=10,
+)
+resp.raise_for_status()
+queue_res = resp.json()
+print(f"Remaining tasks: {queue_res['data']['remaining_tasks_count']}")
+print(f"Pending tasks: {queue_res['data']['pending_tasks_count']}")
 
 # 3. Task progress: poll a specific task until it finishes.
 task_id = "task_888999"
+active_states = {"waiting", "pending", "in_progress"}
 while True:
-    res = client.get_task_status(user_id="dev_user_01", task_id=task_id)
-    if res and res.code == 200:
-        current_status = res.data[0]['status']  # data is a status list
-        print(f"Task {task_id} status: {current_status}")
-
-        if current_status in ['completed', 'failed', 'cancelled']:
-            break
+    resp = requests.get(
+        f"{base_url}/product/scheduler/status",
+        params={"user_id": "dev_user_01", "task_id": task_id},
+        timeout=10,
+    )
+    resp.raise_for_status()
+    items = resp.json().get("data", [])  # data is a status list: [{"task_id": ..., "status": ...}]
+    statuses = {item["status"] for item in items}
+    print(f"Task {task_id} status: {statuses or 'empty'}")
+
+    if not statuses or statuses.isdisjoint(active_states):
+        break
     time.sleep(2)
 ```
diff --git a/docs/en/open_source/open_source_api/scheduler/wait.md b/docs/en/open_source/open_source_api/scheduler/wait.md
index 9de0ff4be..6f356341a 100644
--- a/docs/en/open_source/open_source_api/scheduler/wait.md
+++ b/docs/en/open_source/open_source_api/scheduler/wait.md
@@ -41,36 +41,47 @@ Both endpoints share the following query parameters:
 
 ## 4. Quick Start
 
-Use the open-source SDK for a blocking wait:
+These endpoints are served directly by the open-source Server (`server_api`, router prefix `/product`) and can be called with plain HTTP requests. Note that `user_name`, `timeout_seconds`, and `poll_interval` are query parameters, not a request body. The example below performs a blocking wait:
 
 ```python
-from memos.api.client import MemOSClient
+import json
 
-client = MemOSClient(api_key="...", base_url="...")
+import requests
+
+# Address of your self-hosted MemOS Server (add an Authorization header if auth is enabled)
+base_url = "http://localhost:8000"
 user_name = "dev_user_01"
 
 # Scenario A: blocking wait, commonly used in Python automation scripts.
 print(f"Waiting for user {user_name}'s task queue to drain...")
-res = client.wait_until_idle(
-    user_name=user_name,
-    timeout_seconds=300,
-    poll_interval=2
+resp = requests.post(
+    f"{base_url}/product/scheduler/wait",
+    params={"user_name": user_name, "timeout_seconds": 300, "poll_interval": 2},
+    timeout=310,  # HTTP timeout should be larger than timeout_seconds
 )
-if res and res.code == 200:
+resp.raise_for_status()
+result = resp.json()  # {"message": "idle" | "timeout", "data": {...}}
+if result["message"] == "idle":
     print("All tasks have completed.")
+else:
+    print(f"Timed out with {result['data']['running_tasks']} task(s) still running.")
 
 # Scenario B: streaming progress, commonly used by frontend progress bars.
 print("Listening to the live task progress stream...")
-# The SSE endpoint usually returns a generator from the SDK.
-progress_stream = client.stream_scheduler_progress(
-    user_name=user_name,
-    timeout_seconds=300
-)
-
-for event in progress_stream:
-    # Print the remaining queued tasks in real time.
-    print(f"Remaining queued tasks: {event['remaining_tasks_count']}")
-    if event['status'] == 'idle':
-        print("Scheduler is idle")
-        break
+with requests.get(
+    f"{base_url}/product/scheduler/wait/stream",
+    params={"user_name": user_name, "timeout_seconds": 300},
+    stream=True,
+    timeout=310,
+) as resp:
+    resp.raise_for_status()
+    for line in resp.iter_lines(decode_unicode=True):
+        if not line or not line.startswith("data:"):
+            continue
+        event = json.loads(line.removeprefix("data:").strip())
+        # Print the number of active tasks in real time.
+        print(f"Active tasks: {event['active_tasks']}, status: {event['status']}")
+        if event["status"] in ("idle", "timeout"):
+            print("Scheduler is idle" if event["status"] == "idle" else "Stream timed out")
+            break
 ```

From 2c60267e9672cf9e0e7af53e8e348be8c2c1d99a Mon Sep 17 00:00:00 2001
From: shinetata <149466187+shinetata@users.noreply.github.com>
Date: Thu, 16 Jul 2026 14:36:07 +0800
Subject: [PATCH 05/13] fix(api): show structured add example in /docs for
 /product/add (#2112)

fix(api): show structured add example in /docs (messages not string)

Swagger UI rendered APIADDRequest.messages as "string" because it picks the
leading `str` branch of the `str | MessageList | RawMessageList` union when
generating the example. Add a model-level json_schema_extra example so /docs
shows a copy-paste-ready payload with a structured messages list.

Schema-only change: field types, validation and runtime behaviour are
unchanged, so the core add pipeline is unaffected.

Closes #1505

Co-authored-by: sunqi <sunqi@memtensor.com>
Co-authored-by: Cursor <cursoragent@cursor.com>
---
 src/memos/api/product_models.py  | 15 ++++++++++++
 tests/api/test_product_models.py | 41 ++++++++++++++++++++++++++++++++
 2 files changed, 56 insertions(+)
 create mode 100644 tests/api/test_product_models.py

diff --git a/src/memos/api/product_models.py b/src/memos/api/product_models.py
index a2c83f622..1769b0ee6 100644
--- a/src/memos/api/product_models.py
+++ b/src/memos/api/product_models.py
@@ -608,6 +608,21 @@ def _convert_deprecated_fields(self) -> "APISearchRequest":
 class APIADDRequest(BaseRequest):
     """Request model for creating memories."""
 
+    # Model-level example so the interactive docs (/docs) show a copy-paste-ready
+    # payload. Without it, Swagger UI renders the leading `str` branch of the
+    # `messages` union as `"string"` (see issue #1505). This only affects the
+    # generated OpenAPI schema, not validation or runtime behaviour.
+    model_config = {
+        "json_schema_extra": {
+            "example": {
+                "user_id": "8736b16e-1d20-4163-980b-a5063c3facdc",
+                "writable_cube_ids": ["b32d0977-435d-4828-a86f-4f47f8b55bca"],
+                "messages": [{"role": "user", "content": "I am learning ggplot2 in R."}],
+                "async_mode": "async",
+            }
+        }
+    }
+
     # ==== Basic identifiers ====
     user_id: str = Field(None, description="User ID")
     session_id: str | None = Field(
diff --git a/tests/api/test_product_models.py b/tests/api/test_product_models.py
new file mode 100644
index 000000000..177f8713d
--- /dev/null
+++ b/tests/api/test_product_models.py
@@ -0,0 +1,41 @@
+"""Unit tests for API request-model OpenAPI schemas.
+
+These tests lock the OpenAPI schema behaviour of request models so the
+interactive docs (``/docs``) stay consistent with the documented contract.
+
+Regression guard for issue #1505: the ``/product/add`` example must render
+``messages`` as a structured message list instead of a bare ``"string"``.
+Because ``messages`` is typed as ``str | MessageList | RawMessageList``, Swagger
+UI would otherwise pick the leading ``str`` branch of the ``anyOf`` and show
+``"messages": "string"``, which misleads users into sending plain text.
+"""
+
+from memos.api.product_models import APIADDRequest
+
+
+def test_add_request_exposes_model_level_example():
+    """APIADDRequest must ship a model-level example for the interactive docs."""
+    schema = APIADDRequest.model_json_schema()
+
+    assert "example" in schema, "APIADDRequest should define a model-level example"
+
+
+def test_add_request_example_messages_is_structured_list():
+    """The example's ``messages`` must be a non-empty list of role/content items."""
+    example = APIADDRequest.model_json_schema()["example"]
+
+    messages = example.get("messages")
+    assert isinstance(messages, list), "messages example must be a list, not a bare string"
+    assert messages, "messages example should not be empty"
+
+    first = messages[0]
+    assert first.get("role"), "each example message needs a role"
+    assert first.get("content"), "each example message needs content"
+
+
+def test_add_request_example_covers_core_fields():
+    """The example should be a copy-paste-ready payload for the core add flow."""
+    example = APIADDRequest.model_json_schema()["example"]
+
+    assert "user_id" in example
+    assert "writable_cube_ids" in example

From c0f6c2ccee73ff56f05e38a5f94ee91f06aaf5ec Mon Sep 17 00:00:00 2001
From: Memtensor-AI <project@memtensor.cn>
Date: Thu, 16 Jul 2026 17:41:57 +0800
Subject: [PATCH 06/13] fix(memos-local-plugin): revert half-merged
 MemosHttpClient usages (#2096)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The hermes adapter's __init__.py imported MemosHttpClient from
bridge_client, but the class was never committed — every plugin python
test failed at collection and any real Hermes host loading memos_provider
hit ImportError at module load. The phantom usages arrived via the
pr/may27-fixes merge (0398e0ed) as a half-committed HTTP-bridge feature.

Revert the four usages and the HTTP-first branches in initialize /
_reconnect_bridge, remove the _connect_http_bridge helper and the now-
unused probe_viewer_status / startup_lock_active imports (their
definitions in daemon_manager.py are untouched for a future HTTP PR).
Add test_module_imports_cleanly regression test asserting
MemTensorProvider / MemosBridgeClient / BridgeError are present on the
real contract surfaces and MemosHttpClient is absent on both
memos_provider and bridge_client, so this class of import-level
breakage cannot silently return.

Restores the adapter to stdio-only behavior. Net diff: 57 additions /
54 deletions across the adapter plus the regression test.

Fixes #2096
---
 .../hermes/memos_provider/__init__.py         | 66 ++++---------------
 .../python/test_hermes_provider_pipeline.py   | 45 +++++++++++++
 2 files changed, 57 insertions(+), 54 deletions(-)

diff --git a/apps/memos-local-plugin/adapters/hermes/memos_provider/__init__.py b/apps/memos-local-plugin/adapters/hermes/memos_provider/__init__.py
index 25546dfff..47392b8a7 100644
--- a/apps/memos-local-plugin/adapters/hermes/memos_provider/__init__.py
+++ b/apps/memos-local-plugin/adapters/hermes/memos_provider/__init__.py
@@ -65,13 +65,11 @@
 if str(_PLUGIN_DIR) not in sys.path:
     sys.path.insert(0, str(_PLUGIN_DIR))
 
-from bridge_client import BridgeError, MemosBridgeClient, MemosHttpClient  # noqa: E402
+from bridge_client import BridgeError, MemosBridgeClient  # noqa: E402
 from daemon_manager import (  # noqa: E402
     ensure_bridge_running,
     ensure_viewer_daemon,
     kill_zombie_bridges,
-    probe_viewer_status,
-    startup_lock_active,
 )
 
 
@@ -279,7 +277,7 @@ class MemTensorProvider(MemoryProvider):
     """
 
     def __init__(self) -> None:
-        self._bridge: MemosBridgeClient | MemosHttpClient | None = None
+        self._bridge: MemosBridgeClient | None = None
         self._reconnect_lock = threading.Lock()
         self._session_id: str = ""
         self._episode_id: str = ""
@@ -329,23 +327,6 @@ def is_available(self) -> bool:  # type: ignore[override]
 
     # ─── Lifecycle ────────────────────────────────────────────────────────
 
-    def _connect_http_bridge(self, session_id: str, *, timeout: float = 60.0) -> bool:
-        """Try to connect via HTTP bridge. Sets self._bridge on success."""
-        http_bridge: MemosHttpClient | None = None
-        try:
-            http_bridge = MemosHttpClient()
-            http_bridge.register_host_handler("host.llm.complete", self._handle_host_llm_complete)
-            self._bridge = http_bridge
-            self._open_session(session_id, timeout=timeout)
-            return True
-        except Exception as err:
-            logger.warning("MemOS: HTTP bridge failed, falling back to stdio — %s", err)
-            if http_bridge is not None:
-                with contextlib.suppress(Exception):
-                    http_bridge.close()
-            self._bridge = None
-            return False
-
     def initialize(self, session_id: str, **kwargs: Any) -> None:  # type: ignore[override]
         """Called once at agent startup.
 
@@ -414,32 +395,13 @@ def initialize(self, session_id: str, **kwargs: Any) -> None:  # type: ignore[ov
         except Exception:
             pass
 
-        # If the daemon is already running on the viewer port, connect
-        # to it over HTTP instead of spawning a new stdio bridge. This
-        # eliminates zombie bridge accumulation.
-        viewer_status = probe_viewer_status()
-        if viewer_status == "running_memos":
-            if self._connect_http_bridge(session_id):
-                logger.info(
-                    "MemOS: bridge ready (HTTP) session=%s platform=%s (episode deferred)",
-                    self._session_id,
-                    self._platform,
-                )
-            else:
-                viewer_status = "free"  # force stdio fallback below
-        elif viewer_status == "free":
-            # Re-probe after a short wait only when another process may be
-            # mid-startup (startup lock is held). On a cold first-launch the
-            # lock doesn't exist, so we skip the delay entirely.
-            if startup_lock_active():
-                time.sleep(1.0)
-                viewer_status = probe_viewer_status()
-            if viewer_status == "running_memos" and self._connect_http_bridge(session_id):
-                logger.info(
-                    "MemOS: bridge ready (HTTP, late probe) session=%s platform=%s (episode deferred)",
-                    self._session_id,
-                    self._platform,
-                )
+        # NOTE: An HTTP bridge path used to live here that connected to a
+        # running viewer daemon over HTTP instead of spawning a stdio
+        # subprocess. It depended on ``MemosHttpClient`` which was never
+        # committed — the class was referenced by name only. Issue #2096
+        # reverts the half-merged HTTP feature; the stdio path below is
+        # the sole connection mechanism until the HTTP client lands as a
+        # complete change.
 
         if self._bridge is None:
             try:
@@ -1958,13 +1920,9 @@ def _reconnect_bridge(self, session_id: str = "", *, timeout: float = 30.0) -> N
                 logger.info("MemOS: old bridge closed (pid=%s)", old_pid)
 
             ensure_bridge_running()
-            # Try HTTP first if daemon is running
-            viewer_status = probe_viewer_status()
-            if viewer_status == "running_memos" and self._connect_http_bridge(
-                session_id, timeout=timeout
-            ):
-                logger.info("MemOS: reconnected via HTTP")
-                return
+            # NOTE: HTTP bridge reconnect path was removed alongside issue
+            # #2096. See ``initialize`` for the rationale. Reconnect always
+            # spawns a fresh stdio bridge.
 
             try:
                 ensure_viewer_daemon()
diff --git a/apps/memos-local-plugin/tests/python/test_hermes_provider_pipeline.py b/apps/memos-local-plugin/tests/python/test_hermes_provider_pipeline.py
index 1c8cb6a6c..71e3b347b 100644
--- a/apps/memos-local-plugin/tests/python/test_hermes_provider_pipeline.py
+++ b/apps/memos-local-plugin/tests/python/test_hermes_provider_pipeline.py
@@ -68,6 +68,51 @@ def request(self, method: str, params: dict | None = None, **_kwargs: object) ->
 
 
 class HermesProviderPipelineTests(unittest.TestCase):
+    def test_module_imports_cleanly(self) -> None:
+        """Regression guard for #2096: asserts that ``MemosHttpClient`` is
+        NOT present in ``memos_provider``, since the class was referenced
+        before it was ever committed (see issue #2096).
+
+        Note: the import itself is already validated at collection time —
+        the ``import memos_provider`` at the top of this file will raise
+        ``ImportError`` if a dangling reference is reintroduced, causing
+        the entire test file to fail to load. This test body only adds:
+
+        * the explicit negative guard on ``MemosHttpClient`` below (unique
+          to this test), which covers both the ``memos_provider``
+          re-export surface *and* ``bridge_client`` itself so a partial
+          re-add of the class only in ``bridge_client`` (with no
+          matching re-export) still fails the guard, and
+        * positive checks on ``MemTensorProvider`` (the class the Hermes
+          host actually instantiates) and on ``bridge_client``'s real
+          contract (``MemosBridgeClient`` / ``BridgeError``), rather than
+          on their incidental re-exports through ``memos_provider`` — the
+          latter only appear on the package namespace because
+          ``__init__.py`` uses a bare ``from bridge_client import ...``,
+          which is an implementation detail we don't want the test to
+          lock in.
+        """
+        import importlib
+
+        # MemTensorProvider is the class hermes-agent host instantiates.
+        self.assertTrue(hasattr(memos_provider, "MemTensorProvider"))
+
+        # Assert the actual contract on bridge_client directly rather
+        # than on its re-exports through memos_provider.
+        bc = importlib.import_module("bridge_client")
+        self.assertTrue(hasattr(bc, "MemosBridgeClient"))
+        self.assertTrue(hasattr(bc, "BridgeError"))
+
+        # ``MemosHttpClient`` was referenced by name in a half-merged HTTP
+        # bridge feature (see #2096). It must not reappear until the class
+        # itself is committed in ``bridge_client``. Guard both the
+        # ``memos_provider`` re-export (which is what the original
+        # ImportError travelled through) and ``bridge_client`` itself —
+        # otherwise a partial re-add of the class in ``bridge_client``
+        # without a matching re-export would slip past this test.
+        self.assertFalse(hasattr(memos_provider, "MemosHttpClient"))
+        self.assertFalse(hasattr(bc, "MemosHttpClient"))
+
     def test_lifecycle_persists_turn_and_closes_real_episode(self) -> None:
         bridge = FakeBridge()
         with (

From 698c30f7b8fd36ba6beeba02f6d95b0d11efde73 Mon Sep 17 00:00:00 2001
From: bittergreen <bittergreen.wengqi@hotmail.com>
Date: Thu, 16 Jul 2026 20:50:37 +0800
Subject: [PATCH 07/13] fix: configure logging once per process

---
 src/memos/log.py  | 23 ++++++++++++++++++++++-
 tests/test_log.py | 31 +++++++++++++++++++++++++++++++
 2 files changed, 53 insertions(+), 1 deletion(-)

diff --git a/src/memos/log.py b/src/memos/log.py
index c0bb5bf31..ede5e8280 100644
--- a/src/memos/log.py
+++ b/src/memos/log.py
@@ -27,6 +27,8 @@
 load_dotenv()
 
 selected_log_level = logging.DEBUG if settings.DEBUG else logging.WARNING
+_LOGGING_CONFIG_LOCK = threading.RLock()
+_LOGGING_CONFIGURED_PID: int | None = None
 
 
 def _setup_logfile() -> Path:
@@ -224,12 +226,31 @@ def close(self):
 }
 
 
+def configure_logging(force: bool = False) -> None:
+    """Configure process-local logging once.
+
+    Re-running dictConfig replaces and closes existing handlers. Guarding it avoids races with
+    background threads that may be emitting log records while other modules import loggers.
+    """
+    global _LOGGING_CONFIGURED_PID
+
+    current_pid = os.getpid()
+    if not force and current_pid == _LOGGING_CONFIGURED_PID:
+        return
+
+    with _LOGGING_CONFIG_LOCK:
+        current_pid = os.getpid()
+        if force or current_pid != _LOGGING_CONFIGURED_PID:
+            dictConfig(LOGGING_CONFIG)
+            _LOGGING_CONFIGURED_PID = current_pid
+
+
 def get_logger(name: str | None = None) -> logging.Logger:
     """returns the project logger, scoped to a child name if provided
     Args:
         name: will define a child logger
     """
-    dictConfig(LOGGING_CONFIG)
+    configure_logging()
 
     parent_logger = logging.getLogger("")
     if name:
diff --git a/tests/test_log.py b/tests/test_log.py
index fbd8791ee..e799ff1c4 100644
--- a/tests/test_log.py
+++ b/tests/test_log.py
@@ -28,3 +28,34 @@ def test_get_logger_returns_logger():
     assert any(isinstance(h, logging.StreamHandler) for h in logger.parent.handlers) or any(
         isinstance(h, logging.FileHandler) for h in logger.parent.handlers
     )
+
+
+def test_get_logger_configures_logging_once_per_process(monkeypatch):
+    calls = []
+
+    monkeypatch.setattr(log, "_LOGGING_CONFIGURED_PID", None)
+    monkeypatch.setattr(log.os, "getpid", lambda: 123)
+    monkeypatch.setattr(log, "dictConfig", lambda config: calls.append(config))
+
+    log.get_logger("first")
+    log.get_logger("second")
+
+    assert len(calls) == 1
+
+
+def test_get_logger_reconfigures_after_process_fork(monkeypatch):
+    calls = []
+    pid = 123
+
+    def getpid():
+        return pid
+
+    monkeypatch.setattr(log, "_LOGGING_CONFIGURED_PID", None)
+    monkeypatch.setattr(log.os, "getpid", getpid)
+    monkeypatch.setattr(log, "dictConfig", lambda config: calls.append(config))
+
+    log.get_logger("parent")
+    pid = 456
+    log.get_logger("child")
+
+    assert len(calls) == 2

From 2017cd5ecb67c5a960d0dfedfe877614c69e15b0 Mon Sep 17 00:00:00 2001
From: bittergreen <bittergreen.wengqi@hotmail.com>
Date: Thu, 16 Jul 2026 20:50:53 +0800
Subject: [PATCH 08/13] fix: close scheduler resources on API shutdown

---
 src/memos/api/lifecycle.py      | 26 ++++++++++++++++++++++++++
 src/memos/api/server_api.py     | 10 ++++++++--
 src/memos/api/server_api_ext.py | 12 +++++++++---
 tests/api/test_lifecycle.py     | 31 +++++++++++++++++++++++++++++++
 4 files changed, 74 insertions(+), 5 deletions(-)
 create mode 100644 src/memos/api/lifecycle.py
 create mode 100644 tests/api/test_lifecycle.py

diff --git a/src/memos/api/lifecycle.py b/src/memos/api/lifecycle.py
new file mode 100644
index 000000000..e05a2f777
--- /dev/null
+++ b/src/memos/api/lifecycle.py
@@ -0,0 +1,26 @@
+from collections.abc import Mapping
+from typing import Any
+
+from memos.log import get_logger
+
+
+logger = get_logger(__name__)
+
+
+def shutdown_components(components: Mapping[str, Any] | None) -> None:
+    """Release long-lived API components before the logging system shuts down."""
+    if not components:
+        return
+
+    mem_scheduler = components.get("mem_scheduler")
+    if mem_scheduler is None:
+        return
+
+    for method_name in ("stop", "rabbitmq_close"):
+        method = getattr(mem_scheduler, method_name, None)
+        if not callable(method):
+            continue
+        try:
+            method()
+        except Exception:
+            logger.exception("Failed to run mem_scheduler.%s during API shutdown", method_name)
diff --git a/src/memos/api/server_api.py b/src/memos/api/server_api.py
index a9afe554c..b05e8ac7c 100644
--- a/src/memos/api/server_api.py
+++ b/src/memos/api/server_api.py
@@ -7,8 +7,9 @@
 from starlette.staticfiles import StaticFiles
 
 from memos.api.exceptions import APIExceptionHandler
+from memos.api.lifecycle import shutdown_components
 from memos.api.middleware.request_context import RequestContextMiddleware
-from memos.api.routers.server_router import router as server_router
+from memos.api.routers import server_router as server_router_module
 from memos.plugins.manager import plugin_manager
 
 
@@ -35,7 +36,7 @@
 
 app.add_middleware(RequestContextMiddleware, source="server_api")
 # Include routers
-app.include_router(server_router)
+app.include_router(server_router_module.router)
 
 
 @app.get("/health")
@@ -48,6 +49,11 @@ def health_check():
     }
 
 
+@app.on_event("shutdown")
+def shutdown_server_components() -> None:
+    shutdown_components(server_router_module.components)
+
+
 # Request validation failed
 app.exception_handler(RequestValidationError)(APIExceptionHandler.validation_error_handler)
 # Invalid business code parameters
diff --git a/src/memos/api/server_api_ext.py b/src/memos/api/server_api_ext.py
index 8c457e362..e28505332 100644
--- a/src/memos/api/server_api_ext.py
+++ b/src/memos/api/server_api_ext.py
@@ -26,11 +26,12 @@
 from starlette.responses import Response
 
 # Import Krolik extensions
+from memos.api.lifecycle import shutdown_components
 from memos.api.middleware.rate_limit import RateLimitMiddleware
-from memos.api.routers.admin_router import router as admin_router
 
 # Import base routers from MemOS
-from memos.api.routers.server_router import router as server_router
+from memos.api.routers import server_router as server_router_module
+from memos.api.routers.admin_router import router as admin_router
 
 
 # Try to import exception handlers (may vary between MemOS versions)
@@ -94,7 +95,7 @@ async def dispatch(self, request: Request, call_next) -> Response:
     logger.info("Rate limiting enabled")
 
 # Include routers
-app.include_router(server_router)
+app.include_router(server_router_module.router)
 app.include_router(admin_router)
 
 # Exception handlers
@@ -118,6 +119,11 @@ async def health_check():
     }
 
 
+@app.on_event("shutdown")
+def shutdown_server_components() -> None:
+    shutdown_components(server_router_module.components)
+
+
 if __name__ == "__main__":
     import uvicorn
 
diff --git a/tests/api/test_lifecycle.py b/tests/api/test_lifecycle.py
new file mode 100644
index 000000000..1daa7a22e
--- /dev/null
+++ b/tests/api/test_lifecycle.py
@@ -0,0 +1,31 @@
+from memos.api.lifecycle import shutdown_components
+
+
+class SchedulerStub:
+    def __init__(self, fail_stop: bool = False):
+        self.fail_stop = fail_stop
+        self.calls = []
+
+    def stop(self):
+        self.calls.append("stop")
+        if self.fail_stop:
+            raise RuntimeError("stop failed")
+
+    def rabbitmq_close(self):
+        self.calls.append("rabbitmq_close")
+
+
+def test_shutdown_components_stops_scheduler_before_rabbitmq_close():
+    scheduler = SchedulerStub()
+
+    shutdown_components({"mem_scheduler": scheduler})
+
+    assert scheduler.calls == ["stop", "rabbitmq_close"]
+
+
+def test_shutdown_components_still_closes_rabbitmq_when_stop_fails():
+    scheduler = SchedulerStub(fail_stop=True)
+
+    shutdown_components({"mem_scheduler": scheduler})
+
+    assert scheduler.calls == ["stop", "rabbitmq_close"]

From 8aefda7b571f8d4991a5c7b6b5513e5c6daca91d Mon Sep 17 00:00:00 2001
From: bittergreen <bittergreen.wengqi@hotmail.com>
Date: Thu, 16 Jul 2026 21:09:16 +0800
Subject: [PATCH 09/13] fix: configure logging once per process

Avoid re-running dictConfig on every get_logger call, since repeated configuration can close and replace active handlers while background threads are emitting records. Keep logging configuration process-local and reconfigure only after a PID change.
---
 src/memos/log.py  | 10 +++++-----
 tests/test_log.py |  4 ++--
 2 files changed, 7 insertions(+), 7 deletions(-)

diff --git a/src/memos/log.py b/src/memos/log.py
index ede5e8280..c18bd2118 100644
--- a/src/memos/log.py
+++ b/src/memos/log.py
@@ -226,6 +226,10 @@ def close(self):
 }
 
 
+def _get_current_pid() -> int:
+    return os.getpid()
+
+
 def configure_logging(force: bool = False) -> None:
     """Configure process-local logging once.
 
@@ -234,12 +238,8 @@ def configure_logging(force: bool = False) -> None:
     """
     global _LOGGING_CONFIGURED_PID
 
-    current_pid = os.getpid()
-    if not force and current_pid == _LOGGING_CONFIGURED_PID:
-        return
-
     with _LOGGING_CONFIG_LOCK:
-        current_pid = os.getpid()
+        current_pid = _get_current_pid()
         if force or current_pid != _LOGGING_CONFIGURED_PID:
             dictConfig(LOGGING_CONFIG)
             _LOGGING_CONFIGURED_PID = current_pid
diff --git a/tests/test_log.py b/tests/test_log.py
index e799ff1c4..5387adc34 100644
--- a/tests/test_log.py
+++ b/tests/test_log.py
@@ -34,7 +34,7 @@ def test_get_logger_configures_logging_once_per_process(monkeypatch):
     calls = []
 
     monkeypatch.setattr(log, "_LOGGING_CONFIGURED_PID", None)
-    monkeypatch.setattr(log.os, "getpid", lambda: 123)
+    monkeypatch.setattr(log, "_get_current_pid", lambda: 123)
     monkeypatch.setattr(log, "dictConfig", lambda config: calls.append(config))
 
     log.get_logger("first")
@@ -51,7 +51,7 @@ def getpid():
         return pid
 
     monkeypatch.setattr(log, "_LOGGING_CONFIGURED_PID", None)
-    monkeypatch.setattr(log.os, "getpid", getpid)
+    monkeypatch.setattr(log, "_get_current_pid", getpid)
     monkeypatch.setattr(log, "dictConfig", lambda config: calls.append(config))
 
     log.get_logger("parent")

From 9c7dd2fdbf132a75e9be0eea73b096e41d1ee385 Mon Sep 17 00:00:00 2001
From: bittergreen <bittergreen.wengqi@hotmail.com>
Date: Thu, 16 Jul 2026 21:09:22 +0800
Subject: [PATCH 10/13] fix: close scheduler resources via API lifespan

Use FastAPI lifespan cleanup to stop scheduler and RabbitMQ resources before logging handlers are torn down during worker or pod shutdown.
---
 src/memos/api/server_api.py     | 16 +++++++++++-----
 src/memos/api/server_api_ext.py | 15 ++++++++++-----
 tests/api/test_lifecycle.py     |  7 +++++++
 3 files changed, 28 insertions(+), 10 deletions(-)

diff --git a/src/memos/api/server_api.py b/src/memos/api/server_api.py
index b05e8ac7c..1c6d93b9f 100644
--- a/src/memos/api/server_api.py
+++ b/src/memos/api/server_api.py
@@ -1,6 +1,9 @@
 import logging
 import os
 
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+
 from dotenv import load_dotenv
 from fastapi import FastAPI, HTTPException
 from fastapi.exceptions import RequestValidationError
@@ -26,10 +29,18 @@
     os.getenv("MEMSCHEDULER_REDIS_STREAM_KEY_PREFIX"),
 )
 
+
+@asynccontextmanager
+async def lifespan(_app: FastAPI) -> AsyncIterator[None]:
+    yield
+    shutdown_components(server_router_module.components)
+
+
 app = FastAPI(
     title="MemOS Server REST APIs",
     description="A REST API for managing multiple users with MemOS Server.",
     version="1.0.1",
+    lifespan=lifespan,
 )
 
 app.mount("/download", StaticFiles(directory=os.getenv("FILE_LOCAL_PATH")), name="static_mapping")
@@ -49,11 +60,6 @@ def health_check():
     }
 
 
-@app.on_event("shutdown")
-def shutdown_server_components() -> None:
-    shutdown_components(server_router_module.components)
-
-
 # Request validation failed
 app.exception_handler(RequestValidationError)(APIExceptionHandler.validation_error_handler)
 # Invalid business code parameters
diff --git a/src/memos/api/server_api_ext.py b/src/memos/api/server_api_ext.py
index e28505332..7b5aafc46 100644
--- a/src/memos/api/server_api_ext.py
+++ b/src/memos/api/server_api_ext.py
@@ -18,6 +18,9 @@
 import logging
 import os
 
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+
 from fastapi import FastAPI
 from fastapi.exceptions import RequestValidationError
 from fastapi.middleware.cors import CORSMiddleware
@@ -59,11 +62,18 @@ async def dispatch(self, request: Request, call_next) -> Response:
         return response
 
 
+@asynccontextmanager
+async def lifespan(_app: FastAPI) -> AsyncIterator[None]:
+    yield
+    shutdown_components(server_router_module.components)
+
+
 # Create FastAPI app
 app = FastAPI(
     title="MemOS Server REST APIs (Krolik Extended)",
     description="MemOS API with authentication, rate limiting, and admin endpoints.",
     version="2.0.3-krolik",
+    lifespan=lifespan,
 )
 
 # CORS configuration
@@ -119,11 +129,6 @@ async def health_check():
     }
 
 
-@app.on_event("shutdown")
-def shutdown_server_components() -> None:
-    shutdown_components(server_router_module.components)
-
-
 if __name__ == "__main__":
     import uvicorn
 
diff --git a/tests/api/test_lifecycle.py b/tests/api/test_lifecycle.py
index 1daa7a22e..b2f7f7064 100644
--- a/tests/api/test_lifecycle.py
+++ b/tests/api/test_lifecycle.py
@@ -29,3 +29,10 @@ def test_shutdown_components_still_closes_rabbitmq_when_stop_fails():
     shutdown_components({"mem_scheduler": scheduler})
 
     assert scheduler.calls == ["stop", "rabbitmq_close"]
+
+
+def test_shutdown_components_skips_missing_methods():
+    class MinimalScheduler:
+        pass
+
+    shutdown_components({"mem_scheduler": MinimalScheduler()})

From e0726760e7c30c43cc5b7a6782622dcada2d229f Mon Sep 17 00:00:00 2001
From: dingliang <2650876010@qq.com>
Date: Fri, 17 Jul 2026 11:57:10 +0800
Subject: [PATCH 11/13] feat(api):update SDK version number

---
 pyproject.toml | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/pyproject.toml b/pyproject.toml
index c7297c7d1..00d240ba8 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -4,7 +4,7 @@
 ##############################################################################
 
 name = "MemoryOS"
-version = "2.0.23"
+version = "2.0.24"
 description = "Intelligence Begins with Memory"
 license = {text = "Apache-2.0"}
 readme = "README.md"

From b7404118392000d9fdd9a1b1b0ad079d2203e3ff Mon Sep 17 00:00:00 2001
From: MemOS AutoDev <autodev@memos.local>
Date: Sun, 19 Jul 2026 23:37:45 +0800
Subject: [PATCH 12/13] fix(memos-local-plugin): guard oversized embedding
 inputs and isolate rebuild-batch failures (#2121)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Long trace text (e.g. 30+ KB agent_text) was fed unmodified into
embedding providers, triggering HTTP 400 code:1210 on 智谱
embedding-3 (3072-token single-input cap). rebuildEmbeddings
then set failed = batch.length, so one poison input blanket-
failed every short valid row next to it — the "Rebuild Vectors"
button returned 400 within milliseconds and the embedding
heartbeat probe reported ~60 % error.

Three surgical fixes in apps/memos-local-plugin:

1. Embedder facade truncates each input to
   EmbeddingConfig.maxInputChars (default 6000; 0 disables)
   BEFORE hashing the cache key, so every provider is protected
   and repeat calls with the same head text hit the LRU. One
   input_truncated warn is emitted per embedMany call.

2. rebuildEmbeddings now divides-and-conquers on provider
   failure: split the batch in half and retry each half; when a
   sub-batch is size 1, mark only that slot as failed.
   Aggregation preserves the existing
   EmbeddingMaintenanceRunResult contract.

3. fetcher.ts's http.non_ok warn now carries a truncated
   response body (first 512 chars) so operators can pinpoint
   provider error codes (e.g. code:1210) from gateway.log alone.

4. embedding.maxInputChars is exposed on the user schema with
   default 6000, min 0.

Tests: 5 new embedder cases + 2 new fetcher cases + 1 new
memory-core case (all TDD — red before code, green after).
Unit + integration suite green; two pre-existing failures
(startup-recovery, storage/migrator) reproduce on the base
commit and are unrelated to embedding.
---
 .../core/config/defaults.ts                   |  1 +
 apps/memos-local-plugin/core/config/schema.ts |  9 ++
 .../core/embedding/embedder.ts                | 39 +++++++-
 .../core/embedding/fetcher.ts                 |  5 +
 .../core/embedding/types.ts                   | 13 +++
 .../core/pipeline/memory-core.ts              | 80 +++++++++++-----
 .../tests/unit/embedding/embedder.test.ts     | 81 +++++++++++++++-
 .../tests/unit/embedding/fetcher.test.ts      | 63 +++++++++++++
 .../tests/unit/pipeline/memory-core.test.ts   | 94 +++++++++++++++++++
 9 files changed, 361 insertions(+), 24 deletions(-)

diff --git a/apps/memos-local-plugin/core/config/defaults.ts b/apps/memos-local-plugin/core/config/defaults.ts
index b79df4a5b..dc201891e 100644
--- a/apps/memos-local-plugin/core/config/defaults.ts
+++ b/apps/memos-local-plugin/core/config/defaults.ts
@@ -31,6 +31,7 @@ export const DEFAULT_CONFIG: ResolvedConfig = {
       enabled: true,
       maxItems: 20_000,
     },
+    maxInputChars: 6000,
   },
   llm: {
     provider: "",
diff --git a/apps/memos-local-plugin/core/config/schema.ts b/apps/memos-local-plugin/core/config/schema.ts
index b698196d1..abefa1e9f 100644
--- a/apps/memos-local-plugin/core/config/schema.ts
+++ b/apps/memos-local-plugin/core/config/schema.ts
@@ -43,6 +43,15 @@ const EmbeddingSchema = Type.Object({
     enabled: Bool(true),
     maxItems: NumberInRange(20_000, 0),
   }, { default: {} }),
+  /**
+   * Per-input character cap applied inside the `Embedder` facade before
+   * hashing / calling the provider. Guards against remote embedding
+   * models with a per-request token limit — most notably 智谱
+   * `embedding-3` (3072 tokens ≈ 6-7 KB CJK), which returns HTTP 400
+   * `code:1210` for over-length inputs and used to nuke the whole
+   * rebuild batch (issue #2121). Set to `0` to disable truncation.
+   */
+  maxInputChars: NumberInRange(6000, 0),
 }, { default: {} });
 
 const LlmSchema = Type.Object({
diff --git a/apps/memos-local-plugin/core/embedding/embedder.ts b/apps/memos-local-plugin/core/embedding/embedder.ts
index bd4a5fe92..8d8875c8e 100644
--- a/apps/memos-local-plugin/core/embedding/embedder.ts
+++ b/apps/memos-local-plugin/core/embedding/embedder.ts
@@ -107,7 +107,29 @@ export function createEmbedderWithProvider(
     requests += inputs.length;
     if (inputs.length === 0) return [];
 
-    const normalized = inputs.map(toInput);
+    // Per-input character cap. Truncation runs BEFORE cache-key hashing
+    // so a repeated call with the same head text hits the LRU. Guards
+    // against provider single-input token caps (e.g. 智谱 embedding-3
+    // rejects >3072 tokens with HTTP 400 code:1210 — see issue #2121).
+    const cap = resolveMaxInputChars(config.maxInputChars);
+    let truncatedCount = 0;
+    const normalized = inputs.map(toInput).map((inp) => {
+      if (cap > 0 && inp.text.length > cap) {
+        truncatedCount++;
+        return { ...inp, text: inp.text.slice(0, cap) };
+      }
+      return inp;
+    });
+    if (truncatedCount > 0) {
+      logger.warn("input_truncated", {
+        provider: provider.name,
+        model: config.model,
+        cap,
+        count: truncatedCount,
+        of: normalized.length,
+      });
+    }
+
     const results = new Array<EmbeddingVector | null>(normalized.length).fill(null);
     const dedupEnabled = config.cache.enabled;
     const keys = normalized.map((inp, i) => {
@@ -332,6 +354,21 @@ export function createEmbedderWithProvider(
 
 // ─── Provider lookup ─────────────────────────────────────────────────────────
 
+/**
+ * Default per-input character cap. Chosen at 6000 chars ≈ 2700 CJK
+ * tokens ≈ well under 智谱 embedding-3's 3072-token single-input hard
+ * limit, and safe for the 8K-context sentence-transformer models we
+ * ship with local. Callers can override via `EmbeddingConfig.maxInputChars`;
+ * setting it to `0` disables truncation. See issue #2121.
+ */
+const DEFAULT_MAX_INPUT_CHARS = 6000;
+
+export function resolveMaxInputChars(configured: number | undefined): number {
+  if (configured === undefined) return DEFAULT_MAX_INPUT_CHARS;
+  if (!Number.isFinite(configured) || configured < 0) return 0;
+  return Math.floor(configured);
+}
+
 export function makeProviderFor(name: EmbeddingProviderName): EmbeddingProvider {
   switch (name) {
     case "local":
diff --git a/apps/memos-local-plugin/core/embedding/fetcher.ts b/apps/memos-local-plugin/core/embedding/fetcher.ts
index 40524aac7..b618e62ab 100644
--- a/apps/memos-local-plugin/core/embedding/fetcher.ts
+++ b/apps/memos-local-plugin/core/embedding/fetcher.ts
@@ -46,12 +46,17 @@ export async function httpPostJson<TResp>(opts: HttpPostOpts<unknown>): Promise<
       if (!resp.ok) {
         const text = await safeText(resp);
         const transient = resp.status >= 500 || resp.status === 429;
+        // Include a truncated body so operators can pinpoint provider
+        // error codes (e.g. 智谱 `code:1210` for over-length inputs)
+        // directly from gateway.log without needing a debugger — see
+        // issue #2121.
         opts.log.warn("http.non_ok", {
           url: opts.url,
           status: resp.status,
           attempt,
           transient,
           durationMs: Date.now() - start,
+          body: text ? text.slice(0, 512) : undefined,
         });
         if (transient && attempt <= maxRetries) {
           await backoff(attempt);
diff --git a/apps/memos-local-plugin/core/embedding/types.ts b/apps/memos-local-plugin/core/embedding/types.ts
index 5371b875e..e1f46e925 100644
--- a/apps/memos-local-plugin/core/embedding/types.ts
+++ b/apps/memos-local-plugin/core/embedding/types.ts
@@ -37,6 +37,19 @@ export interface EmbeddingConfig {
   maxRetries?: number;
   /** Max texts per HTTP round trip. Default: 32. */
   batchSize?: number;
+  /**
+   * Per-input character cap. Inputs longer than this are truncated
+   * (character-wise, not token-wise) before being hashed / sent to the
+   * provider. Guards against provider single-input token caps such as
+   * 智谱 embedding-3 (3072 tokens ≈ 6-7 KB CJK). Set `0` or a negative
+   * value to disable. Default: 6000.
+   *
+   * Truncation happens at the facade boundary so all providers (local,
+   * openai_compatible, gemini, cohere, voyage, mistral) benefit; the
+   * cache key is derived from the *truncated* text so repeat calls that
+   * share the same head text hit the LRU.
+   */
+  maxInputChars?: number;
   /** Extra headers to tack on outgoing HTTP. */
   headers?: Record<string, string>;
   /** If true, all output vectors are L2-normalized. Default: true. */
diff --git a/apps/memos-local-plugin/core/pipeline/memory-core.ts b/apps/memos-local-plugin/core/pipeline/memory-core.ts
index 96721cc0b..79bec197f 100644
--- a/apps/memos-local-plugin/core/pipeline/memory-core.ts
+++ b/apps/memos-local-plugin/core/pipeline/memory-core.ts
@@ -4283,28 +4283,10 @@ export function createMemoryCore(
     let failed = 0;
     let error: string | undefined;
     if (batch.length > 0) {
-      try {
-        const vecs = await handle.embedder.embedMany(
-          batch.map((slot) => ({ text: slot.sourceText || "(empty)", role: "document" as const })),
-        );
-        for (let i = 0; i < batch.length; i++) {
-          const slot = batch[i]!;
-          const vec = vecs[i];
-          if (!vec) {
-            failed++;
-            continue;
-          }
-          try {
-            if (slot.update(vec)) updated++;
-            else failed++;
-          } catch {
-            failed++;
-          }
-        }
-      } catch (err) {
-        failed = batch.length;
-        error = err instanceof Error ? err.message : String(err);
-      }
+      const outcome = await embedAndApplySlots(batch);
+      updated = outcome.updated;
+      failed = outcome.failed;
+      error = outcome.firstError;
     }
 
     const statsAfter = computeEmbeddingMaintenanceStats();
@@ -4326,6 +4308,60 @@ export function createMemoryCore(
     };
   }
 
+  /**
+   * Divide-and-conquer embed + write for `rebuildEmbeddings`.
+   *
+   * Before this refactor `rebuildEmbeddings` blanket-failed a whole
+   * batch on any provider throw — one 30 KB trace nuked the counters
+   * for every short trace next to it (issue #2121). Now, on provider
+   * failure the sub-batch is halved and each half retried; when a
+   * single-slot sub-batch still fails, only that one slot is counted
+   * as failed. Worst case is O(N log N) round trips for pathological
+   * inputs; in practice poison slots are ≪ 1 %.
+   */
+  async function embedAndApplySlots(sub: EmbeddingSlot[]): Promise<{
+    updated: number;
+    failed: number;
+    firstError?: string;
+  }> {
+    if (sub.length === 0) return { updated: 0, failed: 0 };
+    try {
+      const vecs = await handle.embedder!.embedMany(
+        sub.map((slot) => ({ text: slot.sourceText || "(empty)", role: "document" as const })),
+      );
+      let updated = 0;
+      let failed = 0;
+      for (let i = 0; i < sub.length; i++) {
+        const slot = sub[i]!;
+        const vec = vecs[i];
+        if (!vec) {
+          failed++;
+          continue;
+        }
+        try {
+          if (slot.update(vec)) updated++;
+          else failed++;
+        } catch {
+          failed++;
+        }
+      }
+      return { updated, failed };
+    } catch (err) {
+      const msg = err instanceof Error ? err.message : String(err);
+      // Base case: a single-slot sub-batch still failed — mark this
+      // one slot as failed and bubble the message up.
+      if (sub.length === 1) return { updated: 0, failed: 1, firstError: msg };
+      const mid = sub.length >> 1;
+      const left = await embedAndApplySlots(sub.slice(0, mid));
+      const right = await embedAndApplySlots(sub.slice(mid));
+      return {
+        updated: left.updated + right.updated,
+        failed: left.failed + right.failed,
+        firstError: left.firstError ?? right.firstError ?? msg,
+      };
+    }
+  }
+
   type EmbeddingSlotKind = "trace" | "policy" | "world_model" | "skill";
   type EmbeddingSlot = {
     kind: EmbeddingSlotKind;
diff --git a/apps/memos-local-plugin/tests/unit/embedding/embedder.test.ts b/apps/memos-local-plugin/tests/unit/embedding/embedder.test.ts
index 3c608a53a..916801c2f 100644
--- a/apps/memos-local-plugin/tests/unit/embedding/embedder.test.ts
+++ b/apps/memos-local-plugin/tests/unit/embedding/embedder.test.ts
@@ -2,7 +2,7 @@ import { beforeAll, describe, expect, it } from "vitest";
 
 import { MemosError } from "../../../agent-contract/errors.js";
 import { createEmbedderWithProvider } from "../../../core/embedding/embedder.js";
-import { initTestLogger } from "../../../core/logger/index.js";
+import { initTestLogger, memoryBuffer } from "../../../core/logger/index.js";
 import type {
   EmbedRole,
   EmbeddingConfig,
@@ -228,4 +228,83 @@ describe("embedder facade", () => {
     expect(closed).toBe(1);
     expect(e.stats().hits).toBe(0);
   });
+
+  /**
+   * Regression: issue #2121. Long trace text (e.g. 30+ KB agent_text)
+   * caused HTTP 400 code:1210 from 智谱 embedding-3 (3072-token cap).
+   * Facade must truncate before hashing / calling the provider, so:
+   *   1. Provider never sees an over-cap input.
+   *   2. Cache key is derived from the truncated text (so a repeat call
+   *      with the same head text hits the LRU).
+   */
+  describe("input truncation guard (#2121)", () => {
+    it("truncates inputs above maxInputChars before the provider call", async () => {
+      const p = new FakeProvider();
+      const e = createEmbedderWithProvider(cfg({ maxInputChars: 10 }), p);
+      const longText = "0123456789ABCDEFGHIJ"; // 20 chars
+      await e.embedMany([longText]);
+      expect(p.calls).toHaveLength(1);
+      expect(p.calls[0]!.texts).toEqual(["0123456789"]);
+    });
+
+    it("hits cache when repeat calls share the truncated prefix", async () => {
+      const p = new FakeProvider();
+      const e = createEmbedderWithProvider(cfg({ maxInputChars: 4 }), p);
+      // Two distinct 10-char inputs with an identical first 4 chars.
+      await e.embedMany(["headTAILONE"]);
+      await e.embedMany(["headTAILTWO"]);
+      // Only the truncated `head` is embedded once; second call is a hit.
+      const s = e.stats();
+      expect(s.roundTrips).toBe(1);
+      expect(s.hits).toBe(1);
+      expect(s.misses).toBe(1);
+      expect(p.calls).toHaveLength(1);
+      expect(p.calls[0]!.texts).toEqual(["head"]);
+    });
+
+    it("does not truncate when maxInputChars is 0 (disabled)", async () => {
+      const p = new FakeProvider();
+      const e = createEmbedderWithProvider(cfg({ maxInputChars: 0 }), p);
+      const longText = "x".repeat(9000);
+      await e.embedMany([longText]);
+      expect(p.calls[0]!.texts[0]!.length).toBe(9000);
+    });
+
+    it("defaults maxInputChars to 6000 when the field is absent", async () => {
+      const p = new FakeProvider();
+      const c = cfg();
+      // Explicitly delete maxInputChars so we can prove the default kicks in.
+      delete (c as { maxInputChars?: number }).maxInputChars;
+      const e = createEmbedderWithProvider(c, p);
+      const longText = "y".repeat(9000);
+      await e.embedMany([longText]);
+      expect(p.calls[0]!.texts[0]!.length).toBe(6000);
+    });
+
+    it("emits a warn per embedMany call when truncation fires (batched, not per input)", async () => {
+      const p = new FakeProvider();
+      const e = createEmbedderWithProvider(cfg({ maxInputChars: 3 }), p);
+      // Snapshot the shared memory-buffer state so we count only the
+      // warns triggered by THIS embedMany call (other tests in this
+      // file also fire input_truncated). Note: `initTestLogger` wires
+      // AppLogSink + ErrorLogSink at both writing to the same memory
+      // buffer, so each warn shows up TWICE in the buffer (one per
+      // sink). We only care that our single call added a well-formed
+      // batched warn, not the exact count.
+      const before = memoryBuffer()
+        .tail({ level: "warn", channel: "embedding", limit: 200 })
+        .filter((r) => r.msg === "input_truncated").length;
+      // Three distinct inputs, all over cap — should batch into one warn.
+      await e.embedMany(["aaaaaa", "bbbbbb", "cccccc"]);
+      const after = memoryBuffer()
+        .tail({ level: "warn", channel: "embedding", limit: 200 })
+        .filter((r) => r.msg === "input_truncated");
+      // Two sinks × 1 emit = 2 records — key property is "not one per
+      // input" (which would be 3×2=6).
+      expect(after.length - before).toBeLessThanOrEqual(2);
+      expect(after.length - before).toBeGreaterThanOrEqual(1);
+      // `tail` returns newest-first — the just-emitted warn is at index 0.
+      expect(after[0]!.data).toMatchObject({ count: 3, cap: 3 });
+    });
+  });
 });
diff --git a/apps/memos-local-plugin/tests/unit/embedding/fetcher.test.ts b/apps/memos-local-plugin/tests/unit/embedding/fetcher.test.ts
index 3b9ee5bf2..c015de45b 100644
--- a/apps/memos-local-plugin/tests/unit/embedding/fetcher.test.ts
+++ b/apps/memos-local-plugin/tests/unit/embedding/fetcher.test.ts
@@ -151,4 +151,67 @@ describe("embedding/fetcher", () => {
       signal: ctrl.signal,
     });
   });
+
+  /**
+   * Regression: issue #2121. The gateway.log line
+   * "embedding_unavailable: HTTP 400 from openai_compatible" carried no
+   * response body, so operators could not see 智谱's code:1210 message.
+   * Truncated body is now attached to the warn detail (first 512 chars).
+   */
+  it("http.non_ok warn detail carries a truncated response body", async () => {
+    mockFetch([
+      new Response(
+        '{"error":{"code":"1210","message":"API 调用参数有误"}}',
+        { status: 400 },
+      ),
+    ]);
+    const warns: Array<{ msg: string; detail?: Record<string, unknown> }> = [];
+    const log: ProviderLogger = {
+      trace: () => {},
+      debug: () => {},
+      info: () => {},
+      warn: (msg, detail) => warns.push({ msg, detail }),
+      error: () => {},
+    };
+    await expect(
+      httpPostJson({
+        url: "https://x",
+        body: {},
+        provider: "openai_compatible",
+        log,
+        maxRetries: 0,
+      }),
+    ).rejects.toBeInstanceOf(MemosError);
+    const nonOk = warns.find((w) => w.msg === "http.non_ok");
+    expect(nonOk, "http.non_ok warn should have been emitted").toBeTruthy();
+    expect(nonOk!.detail).toMatchObject({ status: 400 });
+    expect(typeof nonOk!.detail!.body).toBe("string");
+    expect(nonOk!.detail!.body).toContain("code");
+    expect(nonOk!.detail!.body).toContain("1210");
+  });
+
+  it("http.non_ok body is truncated to at most 512 chars", async () => {
+    const bigBody = "e".repeat(2000);
+    mockFetch([new Response(bigBody, { status: 400 })]);
+    const warns: Array<{ msg: string; detail?: Record<string, unknown> }> = [];
+    const log: ProviderLogger = {
+      trace: () => {},
+      debug: () => {},
+      info: () => {},
+      warn: (msg, detail) => warns.push({ msg, detail }),
+      error: () => {},
+    };
+    await expect(
+      httpPostJson({
+        url: "https://x",
+        body: {},
+        provider: "openai_compatible",
+        log,
+        maxRetries: 0,
+      }),
+    ).rejects.toBeInstanceOf(MemosError);
+    const nonOk = warns.find((w) => w.msg === "http.non_ok");
+    expect(nonOk).toBeTruthy();
+    expect((nonOk!.detail!.body as string).length).toBeLessThanOrEqual(512);
+  });
 });
diff --git a/apps/memos-local-plugin/tests/unit/pipeline/memory-core.test.ts b/apps/memos-local-plugin/tests/unit/pipeline/memory-core.test.ts
index 62c18fa16..2cd67c615 100644
--- a/apps/memos-local-plugin/tests/unit/pipeline/memory-core.test.ts
+++ b/apps/memos-local-plugin/tests/unit/pipeline/memory-core.test.ts
@@ -280,6 +280,100 @@ describe("MemoryCore façade", () => {
     expect(row?.vecSummary?.length).toBe(TEST_EMBED_DIMENSIONS);
   });
 
+  /**
+   * Regression: issue #2121. Before the fix, `rebuildEmbeddings` caught
+   * any provider throw and set `failed = batch.length` — one 30 KB
+   * trace nuked the whole batch, so short valid rows next to it were
+   * counted as failed. The divide-and-conquer retry must isolate the
+   * poisonous input to only its own slot.
+   */
+  it("rebuildEmbeddings isolates a single-slot provider failure", async () => {
+    const poison = "POISON";
+    // Custom embedder that throws whenever the batch contains `poison`.
+    const deps = buildDeps(db!);
+    deps.embedder = {
+      ...deps.embedder!,
+      async embedMany(inputs) {
+        const texts = inputs.map((i) => (typeof i === "string" ? i : i.text));
+        if (texts.some((t) => t.includes(poison))) {
+          throw new Error("boom: over-length input");
+        }
+        // Deterministic non-zero vector.
+        return texts.map(() => {
+          const v = new Float32Array(TEST_EMBED_DIMENSIONS);
+          v[0] = 1;
+          return v;
+        });
+      },
+      async embedOne(input) {
+        const text = typeof input === "string" ? input : input.text;
+        if (text.includes(poison)) throw new Error("boom");
+        const v = new Float32Array(TEST_EMBED_DIMENSIONS);
+        v[0] = 1;
+        return v;
+      },
+    } as typeof deps.embedder;
+
+    pipeline = createPipeline(deps);
+    core = createMemoryCore(
+      pipeline,
+      resolveHome("openclaw", "/tmp/memos-mc-test"),
+      "test",
+    );
+    await core.init();
+
+    const rows = [
+      { id: "tr_ok_1", summary: "clean summary one" },
+      { id: "tr_ok_2", summary: "clean summary two" },
+      { id: "tr_bad",  summary: `${poison} tail text`   },
+      { id: "tr_ok_3", summary: "clean summary three" },
+      { id: "tr_ok_4", summary: "clean summary four" },
+    ];
+    await core.importBundle({
+      version: 1,
+      traces: rows.map((r, i) => ({
+        id: r.id,
+        episodeId: `ep_iso_${i}`,
+        sessionId: `se_iso_${i}`,
+        ts: 1_700_000_000_000 + i,
+        userText: `user text ${i}`,
+        agentText: `agent text ${i}`,
+        summary: r.summary,
+        toolCalls: [],
+        value: 0,
+        alpha: 0,
+        priority: 0,
+        turnId: 1_700_000_000_000 + i,
+      })),
+    });
+
+    const before = await core.embeddingMaintenanceStats();
+    expect(before.byKind.trace.missing).toBe(rows.length * 2);
+
+    const result = await core.rebuildEmbeddings({ mode: "repair", limit: 100 });
+    // Every row has 2 slots (summary + action). Only the poison row's
+    // `vec_summary` slot must fail. `vec_action` is derived from
+    // `agentText` which contains no poison, so it must still succeed.
+    // Expected: 9 updated, 1 failed.
+    expect(result.processed).toBe(rows.length * 2);
+    expect(result.updated).toBe(rows.length * 2 - 1);
+    expect(result.failed).toBe(1);
+    expect(result.error).toMatch(/boom/);
+
+    // The clean rows must actually have vectors written.
+    for (const r of rows) {
+      if (r.id === "tr_bad") continue;
+      const row = db!.repos.traces.getById(r.id as never);
+      expect(row?.vecSummary?.length).toBe(TEST_EMBED_DIMENSIONS);
+      expect(row?.vecAction?.length).toBe(TEST_EMBED_DIMENSIONS);
+    }
+    // The poison row: `vec_summary` never applied, `vec_action` should
+    // still be populated (agentText has no poison).
+    const badRow = db!.repos.traces.getById("tr_bad" as never);
+    expect(badRow?.vecSummary).toBeNull();
+    expect(badRow?.vecAction?.length).toBe(TEST_EMBED_DIMENSIONS);
+  });
+
   it("does not require action vectors for lightweight memory traces", async () => {
     pipeline = createPipeline(buildDeps(db!, configWithLightweightMemory(true)));
     core = createMemoryCore(

From 868b87798c8d60a008b14cfae6be32fed1625c91 Mon Sep 17 00:00:00 2001
From: MemOS AutoDev <autodev@memos.local>
Date: Mon, 20 Jul 2026 00:35:25 +0800
Subject: [PATCH 13/13] fix(memos-local-plugin): address OCR review findings
 for #2121 fix
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- Share DEFAULT_MAX_INPUT_CHARS via embedding/constants.ts so schema,
  defaults and facade fallback cannot drift; lower default 6000→4000
  chars so CJK-dominant text stays under embedding-3's 3072-token cap
- resolveMaxInputChars: NaN falls back to the default instead of
  silently disabling truncation; +Infinity is an explicit opt-out
- embedAndApplySlots: short-circuit divide-and-conquer on transient
  errors (429/5xx/network) to avoid O(N log N × maxRetries) call
  amplification; pass embedder as a parameter instead of a non-null
  assertion; surface firstError when embedMany returns short
- Document the http.non_ok body log field as potentially sensitive
  (verbatim third-party payload) in fetcher.ts and README

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
---
 .../core/config/defaults.ts                   |   3 +-
 apps/memos-local-plugin/core/config/schema.ts |  13 +-
 .../core/embedding/README.md                  |   8 ++
 .../core/embedding/constants.ts               |  29 +++++
 .../core/embedding/embedder.ts                |  24 ++--
 .../core/embedding/fetcher.ts                 |   7 +
 .../core/embedding/index.ts                   |   2 +
 .../core/embedding/types.ts                   |   6 +-
 .../core/pipeline/memory-core.ts              |  64 +++++++--
 .../tests/unit/embedding/embedder.test.ts     |  28 +++-
 .../tests/unit/pipeline/memory-core.test.ts   | 122 +++++++++++++++++-
 11 files changed, 277 insertions(+), 29 deletions(-)
 create mode 100644 apps/memos-local-plugin/core/embedding/constants.ts

diff --git a/apps/memos-local-plugin/core/config/defaults.ts b/apps/memos-local-plugin/core/config/defaults.ts
index dc201891e..ddb0cbf40 100644
--- a/apps/memos-local-plugin/core/config/defaults.ts
+++ b/apps/memos-local-plugin/core/config/defaults.ts
@@ -4,6 +4,7 @@
  * to change.
  */
 
+import { DEFAULT_MAX_INPUT_CHARS } from "../embedding/constants.js";
 import type { ResolvedConfig } from "./schema.js";
 
 export const DEFAULT_CONFIG: ResolvedConfig = {
@@ -31,7 +32,7 @@ export const DEFAULT_CONFIG: ResolvedConfig = {
       enabled: true,
       maxItems: 20_000,
     },
-    maxInputChars: 6000,
+    maxInputChars: DEFAULT_MAX_INPUT_CHARS,
   },
   llm: {
     provider: "",
diff --git a/apps/memos-local-plugin/core/config/schema.ts b/apps/memos-local-plugin/core/config/schema.ts
index abefa1e9f..c5b20e917 100644
--- a/apps/memos-local-plugin/core/config/schema.ts
+++ b/apps/memos-local-plugin/core/config/schema.ts
@@ -10,6 +10,8 @@
 
 import { Type, type Static } from "@sinclair/typebox";
 
+import { DEFAULT_MAX_INPUT_CHARS } from "../embedding/constants.js";
+
 // ─── Reusable building blocks ───────────────────────────────────────────────
 
 const StringWithDefault = (def = "") => Type.String({ default: def });
@@ -47,11 +49,14 @@ const EmbeddingSchema = Type.Object({
    * Per-input character cap applied inside the `Embedder` facade before
    * hashing / calling the provider. Guards against remote embedding
    * models with a per-request token limit — most notably 智谱
-   * `embedding-3` (3072 tokens ≈ 6-7 KB CJK), which returns HTTP 400
-   * `code:1210` for over-length inputs and used to nuke the whole
-   * rebuild batch (issue #2121). Set to `0` to disable truncation.
+   * `embedding-3` (3072-token single-input cap; CJK averages ~1.3–1.5
+   * chars per token, so the 4000-char default keeps CJK-dominant text
+   * under the limit), which returns HTTP 400 `code:1210` for
+   * over-length inputs and used to nuke the whole rebuild batch
+   * (issue #2121). Set to `0` to disable truncation. The default is
+   * `DEFAULT_MAX_INPUT_CHARS` in `core/embedding/constants.ts`.
    */
-  maxInputChars: NumberInRange(6000, 0),
+  maxInputChars: NumberInRange(DEFAULT_MAX_INPUT_CHARS, 0),
 }, { default: {} });
 
 const LlmSchema = Type.Object({
diff --git a/apps/memos-local-plugin/core/embedding/README.md b/apps/memos-local-plugin/core/embedding/README.md
index ab4393395..25df3e485 100644
--- a/apps/memos-local-plugin/core/embedding/README.md
+++ b/apps/memos-local-plugin/core/embedding/README.md
@@ -157,6 +157,14 @@ Unit tests live in `tests/unit/embedding/`:
 - `gemini`'s `?key=<API_KEY>` puts the secret in the URL; `fetcher.ts`
   redacts query string via the logger's redaction pipeline. Do not log the
   raw URL elsewhere.
+- `fetcher.ts` attaches a truncated (≤ 512 chars) provider response body to
+  `http.non_ok` warn logs so operators can see error codes like 智谱's
+  `code:1210` (issue #2121). This field is a **verbatim third-party
+  payload**: some providers echo fragments of the submitted embedding input
+  (i.e. potentially personal memory content) back inside error responses.
+  Treat any log sink that carries `http.non_ok` (e.g. `gateway.log`) as
+  potentially containing user data — apply the same retention/access rules
+  as for raw memory content.
 - `voyage` and `cohere` charge per token; be mindful when bumping
   `batchSize` — large batches amortize HTTP overhead but hit TPM ceilings.
 - Changing `dimensions` in config after writing vectors to SQLite breaks
diff --git a/apps/memos-local-plugin/core/embedding/constants.ts b/apps/memos-local-plugin/core/embedding/constants.ts
new file mode 100644
index 000000000..9516784a1
--- /dev/null
+++ b/apps/memos-local-plugin/core/embedding/constants.ts
@@ -0,0 +1,29 @@
+/**
+ * Shared embedding constants. Kept dependency-free so `core/config/`
+ * (schema + defaults) can import from here without dragging in the
+ * provider implementations behind `embedder.ts`.
+ */
+
+/**
+ * Default per-input character cap for embedding inputs.
+ *
+ * Chosen at 4000 chars with 智谱 embedding-3's 3072-token single-input
+ * hard limit as the reference worst case: GLM tokenizers average
+ * ~1.3–1.5 chars per token for Chinese, so 4000 CJK chars ≈ 2700–3000
+ * tokens — under the cap for typical content (the previous 6000
+ * default mapped to ≈ 4000–4600 tokens and could still trip HTTP 400
+ * `code:1210` on CJK-dominant inputs). ASCII tokenizes at ~4
+ * chars/token, so 4000 chars ≈ 1000 tokens, safe for every supported
+ * provider. The cap is a guard, not a hard guarantee — pathological
+ * inputs that still overflow are isolated per-slot by the
+ * divide-and-conquer retry in `pipeline/memory-core.ts`.
+ *
+ * Callers can override via `EmbeddingConfig.maxInputChars`; `0`, a
+ * negative value, or `Infinity` disables truncation (see
+ * `resolveMaxInputChars` in `embedder.ts`). See issue #2121.
+ *
+ * This constant is the single source of truth — `config/schema.ts` and
+ * `config/defaults.ts` import it so the schema default, the runtime
+ * default, and the facade fallback can never drift apart.
+ */
+export const DEFAULT_MAX_INPUT_CHARS = 4000;
diff --git a/apps/memos-local-plugin/core/embedding/embedder.ts b/apps/memos-local-plugin/core/embedding/embedder.ts
index 8d8875c8e..72a0a7b33 100644
--- a/apps/memos-local-plugin/core/embedding/embedder.ts
+++ b/apps/memos-local-plugin/core/embedding/embedder.ts
@@ -24,6 +24,7 @@ import {
   makeCacheKey,
   type EmbedCache,
 } from "./cache.js";
+import { DEFAULT_MAX_INPUT_CHARS } from "./constants.js";
 import { postProcess } from "./normalize.js";
 import { CohereEmbeddingProvider } from "./providers/cohere.js";
 import { GeminiEmbeddingProvider } from "./providers/gemini.js";
@@ -355,17 +356,22 @@ export function createEmbedderWithProvider(
 // ─── Provider lookup ─────────────────────────────────────────────────────────
 
 /**
- * Default per-input character cap. Chosen at 6000 chars ≈ 2700 CJK
- * tokens ≈ well under 智谱 embedding-3's 3072-token single-input hard
- * limit, and safe for the 8K-context sentence-transformer models we
- * ship with local. Callers can override via `EmbeddingConfig.maxInputChars`;
- * setting it to `0` disables truncation. See issue #2121.
+ * Resolve the effective per-input character cap from config.
+ *
+ * Semantics (mirrors the `EmbeddingConfig.maxInputChars` JSDoc):
+ *   - `undefined`          → `DEFAULT_MAX_INPUT_CHARS` (guard on by default)
+ *   - `NaN`                → `DEFAULT_MAX_INPUT_CHARS` (invalid value — e.g.
+ *                            a typo'd config — must NOT silently disable the
+ *                            guard, or issue #2121 sneaks back in)
+ *   - `0` / negative       → `0` (documented explicit opt-out)
+ *   - `Infinity`           → `0` ("no cap" — explicit opt-out)
+ *   - any other number     → `Math.floor(value)`
  */
-const DEFAULT_MAX_INPUT_CHARS = 6000;
-
 export function resolveMaxInputChars(configured: number | undefined): number {
-  if (configured === undefined) return DEFAULT_MAX_INPUT_CHARS;
-  if (!Number.isFinite(configured) || configured < 0) return 0;
+  if (configured === undefined || Number.isNaN(configured)) {
+    return DEFAULT_MAX_INPUT_CHARS;
+  }
+  if (configured < 0 || !Number.isFinite(configured)) return 0;
   return Math.floor(configured);
 }
 
diff --git a/apps/memos-local-plugin/core/embedding/fetcher.ts b/apps/memos-local-plugin/core/embedding/fetcher.ts
index b618e62ab..fec8e5d49 100644
--- a/apps/memos-local-plugin/core/embedding/fetcher.ts
+++ b/apps/memos-local-plugin/core/embedding/fetcher.ts
@@ -50,6 +50,13 @@ export async function httpPostJson<TResp>(opts: HttpPostOpts<unknown>): Promise<
         // error codes (e.g. 智谱 `code:1210` for over-length inputs)
         // directly from gateway.log without needing a debugger — see
         // issue #2121.
+        //
+        // SENSITIVITY: `body` is a verbatim excerpt of a third-party
+        // response. Providers sometimes echo fragments of the submitted
+        // input (which may contain personal memory content) back inside
+        // error payloads, so operators must treat log sinks carrying
+        // this field with the same care as raw memory content — see
+        // "Caveats" in core/embedding/README.md.
         opts.log.warn("http.non_ok", {
           url: opts.url,
           status: resp.status,
diff --git a/apps/memos-local-plugin/core/embedding/index.ts b/apps/memos-local-plugin/core/embedding/index.ts
index 99faa0048..ae70cb66e 100644
--- a/apps/memos-local-plugin/core/embedding/index.ts
+++ b/apps/memos-local-plugin/core/embedding/index.ts
@@ -6,7 +6,9 @@ export {
   createEmbedder,
   createEmbedderWithProvider,
   makeProviderFor,
+  resolveMaxInputChars,
 } from "./embedder.js";
+export { DEFAULT_MAX_INPUT_CHARS } from "./constants.js";
 export {
   LruEmbedCache,
   NullEmbedCache,
diff --git a/apps/memos-local-plugin/core/embedding/types.ts b/apps/memos-local-plugin/core/embedding/types.ts
index e1f46e925..a0e13d4d9 100644
--- a/apps/memos-local-plugin/core/embedding/types.ts
+++ b/apps/memos-local-plugin/core/embedding/types.ts
@@ -41,8 +41,10 @@ export interface EmbeddingConfig {
    * Per-input character cap. Inputs longer than this are truncated
    * (character-wise, not token-wise) before being hashed / sent to the
    * provider. Guards against provider single-input token caps such as
-   * 智谱 embedding-3 (3072 tokens ≈ 6-7 KB CJK). Set `0` or a negative
-   * value to disable. Default: 6000.
+   * 智谱 embedding-3 (3072 tokens; CJK ≈ 1.3–1.5 chars/token). Set `0`,
+   * a negative value, or `Infinity` to disable; `NaN` (invalid config)
+   * falls back to the default rather than disabling the guard.
+   * Default: `DEFAULT_MAX_INPUT_CHARS` (4000) from `constants.ts`.
    *
    * Truncation happens at the facade boundary so all providers (local,
    * openai_compatible, gemini, cohere, voyage, mistral) benefit; the
diff --git a/apps/memos-local-plugin/core/pipeline/memory-core.ts b/apps/memos-local-plugin/core/pipeline/memory-core.ts
index 79bec197f..99745b02b 100644
--- a/apps/memos-local-plugin/core/pipeline/memory-core.ts
+++ b/apps/memos-local-plugin/core/pipeline/memory-core.ts
@@ -4283,7 +4283,7 @@ export function createMemoryCore(
     let failed = 0;
     let error: string | undefined;
     if (batch.length > 0) {
-      const outcome = await embedAndApplySlots(batch);
+      const outcome = await embedAndApplySlots(batch, handle.embedder);
       updated = outcome.updated;
       failed = outcome.failed;
       error = outcome.firstError;
@@ -4316,26 +4316,44 @@ export function createMemoryCore(
    * for every short trace next to it (issue #2121). Now, on provider
    * failure the sub-batch is halved and each half retried; when a
    * single-slot sub-batch still fails, only that one slot is counted
-   * as failed. Worst case is O(N log N) round trips for pathological
-   * inputs; in practice poison slots are ≪ 1 %.
+   * as failed.
+   *
+   * Splitting only pays off for *content-specific* failures (one
+   * poisonous input rejected by the provider). For transient/systemic
+   * failures (network down, 5xx, 429) the fetcher has already
+   * exhausted its internal retries before the throw reaches us —
+   * halving would multiply total provider calls by O(log N) while
+   * every half fails for the same systemic reason. Those errors
+   * short-circuit: the whole sub-batch is marked failed in one step
+   * and the next `rebuildEmbeddings` run retries it. Worst case for
+   * content errors is O(N log N) round trips; in practice poison
+   * slots are ≪ 1 %.
    */
-  async function embedAndApplySlots(sub: EmbeddingSlot[]): Promise<{
+  async function embedAndApplySlots(
+    sub: EmbeddingSlot[],
+    embedder: NonNullable<PipelineHandle["embedder"]>,
+  ): Promise<{
     updated: number;
     failed: number;
     firstError?: string;
   }> {
     if (sub.length === 0) return { updated: 0, failed: 0 };
     try {
-      const vecs = await handle.embedder!.embedMany(
+      const vecs = await embedder.embedMany(
         sub.map((slot) => ({ text: slot.sourceText || "(empty)", role: "document" as const })),
       );
       let updated = 0;
       let failed = 0;
+      let firstError: string | undefined;
       for (let i = 0; i < sub.length; i++) {
         const slot = sub[i]!;
         const vec = vecs[i];
         if (!vec) {
+          // Contract violation: embedMany returned fewer vectors than
+          // inputs. Surface a diagnostic so the operator-facing `error`
+          // field is not silently empty for this failure mode.
           failed++;
+          firstError ??= `embedMany returned no vector for slot ${slot.id} (index ${i} of ${sub.length})`;
           continue;
         }
         try {
@@ -4345,15 +4363,22 @@ export function createMemoryCore(
           failed++;
         }
       }
-      return { updated, failed };
+      return { updated, failed, firstError };
     } catch (err) {
       const msg = err instanceof Error ? err.message : String(err);
       // Base case: a single-slot sub-batch still failed — mark this
       // one slot as failed and bubble the message up.
       if (sub.length === 1) return { updated: 0, failed: 1, firstError: msg };
+      // Transient/systemic errors fail every half identically —
+      // splitting would only amplify already-exhausted retries
+      // (O(N log N × maxRetries) HTTP calls). Fail the sub-batch in
+      // one step instead; a later run retries it.
+      if (isTransientEmbeddingError(err)) {
+        return { updated: 0, failed: sub.length, firstError: msg };
+      }
       const mid = sub.length >> 1;
-      const left = await embedAndApplySlots(sub.slice(0, mid));
-      const right = await embedAndApplySlots(sub.slice(mid));
+      const left = await embedAndApplySlots(sub.slice(0, mid), embedder);
+      const right = await embedAndApplySlots(sub.slice(mid), embedder);
       return {
         updated: left.updated + right.updated,
         failed: left.failed + right.failed,
@@ -4362,6 +4387,29 @@ export function createMemoryCore(
     }
   }
 
+  /**
+   * Positively identify transient/systemic embedding failures so the
+   * divide-and-conquer in `embedAndApplySlots` can short-circuit
+   * instead of amplifying retries. Anything we cannot classify with
+   * confidence is treated as content-specific (split) — mis-splitting
+   * a systemic error costs extra HTTP calls, but mis-short-circuiting
+   * a content error re-nukes whole batches, which is the very bug
+   * (#2121) the splitting exists to fix.
+   *
+   * Classification sources (see `core/embedding/fetcher.ts`):
+   *   - `details.status` 429 / 5xx → transient (retries already
+   *     exhausted inside the fetcher before the throw).
+   *   - No status + fetcher's stable network / retry-exhaustion
+   *     message prefixes → transient.
+   *   - 4xx statuses (e.g. 智谱 400 `code:1210`) → content-specific.
+   */
+  function isTransientEmbeddingError(err: unknown): boolean {
+    if (!(err instanceof MemosError)) return false;
+    const status = (err.details as { status?: unknown } | undefined)?.status;
+    if (typeof status === "number") return status === 429 || status >= 500;
+    return /^Network error calling |^Exhausted retries to /.test(err.message);
+  }
+
   type EmbeddingSlotKind = "trace" | "policy" | "world_model" | "skill";
   type EmbeddingSlot = {
     kind: EmbeddingSlotKind;
diff --git a/apps/memos-local-plugin/tests/unit/embedding/embedder.test.ts b/apps/memos-local-plugin/tests/unit/embedding/embedder.test.ts
index 916801c2f..abcd5f69c 100644
--- a/apps/memos-local-plugin/tests/unit/embedding/embedder.test.ts
+++ b/apps/memos-local-plugin/tests/unit/embedding/embedder.test.ts
@@ -1,7 +1,11 @@
 import { beforeAll, describe, expect, it } from "vitest";
 
 import { MemosError } from "../../../agent-contract/errors.js";
-import { createEmbedderWithProvider } from "../../../core/embedding/embedder.js";
+import { DEFAULT_MAX_INPUT_CHARS } from "../../../core/embedding/constants.js";
+import {
+  createEmbedderWithProvider,
+  resolveMaxInputChars,
+} from "../../../core/embedding/embedder.js";
 import { initTestLogger, memoryBuffer } from "../../../core/logger/index.js";
 import type {
   EmbedRole,
@@ -270,15 +274,31 @@ describe("embedder facade", () => {
       expect(p.calls[0]!.texts[0]!.length).toBe(9000);
     });
 
-    it("defaults maxInputChars to 6000 when the field is absent", async () => {
+    it("defaults maxInputChars to DEFAULT_MAX_INPUT_CHARS when the field is absent", async () => {
       const p = new FakeProvider();
       const c = cfg();
       // Explicitly delete maxInputChars so we can prove the default kicks in.
       delete (c as { maxInputChars?: number }).maxInputChars;
       const e = createEmbedderWithProvider(c, p);
-      const longText = "y".repeat(9000);
+      const longText = "y".repeat(DEFAULT_MAX_INPUT_CHARS + 3000);
       await e.embedMany([longText]);
-      expect(p.calls[0]!.texts[0]!.length).toBe(6000);
+      expect(p.calls[0]!.texts[0]!.length).toBe(DEFAULT_MAX_INPUT_CHARS);
+    });
+
+    /**
+     * `resolveMaxInputChars` boundary semantics:
+     *   - invalid values (NaN) must fall back to the default — a typo'd
+     *     config must not silently disable the #2121 guard
+     *   - 0 / negative / Infinity are explicit opt-outs → 0 (disabled)
+     */
+    it("resolveMaxInputChars: invalid falls back to default; opt-outs disable", () => {
+      expect(resolveMaxInputChars(undefined)).toBe(DEFAULT_MAX_INPUT_CHARS);
+      expect(resolveMaxInputChars(Number.NaN)).toBe(DEFAULT_MAX_INPUT_CHARS);
+      expect(resolveMaxInputChars(0)).toBe(0);
+      expect(resolveMaxInputChars(-1)).toBe(0);
+      expect(resolveMaxInputChars(Number.NEGATIVE_INFINITY)).toBe(0);
+      expect(resolveMaxInputChars(Number.POSITIVE_INFINITY)).toBe(0);
+      expect(resolveMaxInputChars(1234.9)).toBe(1234);
     });
 
     it("emits a warn per embedMany call when truncation fires (batched, not per input)", async () => {
diff --git a/apps/memos-local-plugin/tests/unit/pipeline/memory-core.test.ts b/apps/memos-local-plugin/tests/unit/pipeline/memory-core.test.ts
index 2cd67c615..0f858b5f8 100644
--- a/apps/memos-local-plugin/tests/unit/pipeline/memory-core.test.ts
+++ b/apps/memos-local-plugin/tests/unit/pipeline/memory-core.test.ts
@@ -30,7 +30,7 @@ import { RECOVERY_REASONS } from "../../../core/pipeline/recovery-constants.js";
 import { makeTmpDb, type TmpDbHandle } from "../../helpers/tmp-db.js";
 import { makeTmpHome, type TmpHomeContext } from "../../helpers/tmp-home.js";
 import { fakeEmbedder } from "../../helpers/fake-embedder.js";
-import type { MemosError } from "../../../agent-contract/errors.js";
+import { MemosError } from "../../../agent-contract/errors.js";
 import type { SkillId, SkillRow, TraceRow } from "../../../core/types.js";
 
 let db: TmpDbHandle | null = null;
@@ -374,6 +374,126 @@ describe("MemoryCore façade", () => {
     expect(badRow?.vecAction?.length).toBe(TEST_EMBED_DIMENSIONS);
   });
 
+  /**
+   * Companion to the isolation test above: transient/systemic provider
+   * failures (network down, 5xx, 429 — thrown by the fetcher as
+   * `MemosError` AFTER its internal retries are exhausted) must NOT
+   * trigger the divide-and-conquer split. Splitting a systemic failure
+   * multiplies provider calls by O(log N) with zero chance of success.
+   * The whole batch fails in a single call; the next run retries it.
+   */
+  it("rebuildEmbeddings does not split the batch on transient provider errors", async () => {
+    let embedCalls = 0;
+    const deps = buildDeps(db!);
+    deps.embedder = {
+      ...deps.embedder!,
+      async embedMany() {
+        embedCalls++;
+        // Shape thrown by `httpPostJson` for a 503 after retry exhaustion.
+        throw new MemosError(
+          "embedding_unavailable",
+          "HTTP 503 from openai_compatible",
+          { provider: "openai_compatible", url: "https://x", status: 503 },
+        );
+      },
+    } as typeof deps.embedder;
+
+    pipeline = createPipeline(deps);
+    core = createMemoryCore(
+      pipeline,
+      resolveHome("openclaw", "/tmp/memos-mc-test"),
+      "test",
+    );
+    await core.init();
+
+    await core.importBundle({
+      version: 1,
+      traces: Array.from({ length: 4 }, (_, i) => ({
+        id: `tr_tra_${i}`,
+        episodeId: `ep_tra_${i}`,
+        sessionId: `se_tra_${i}`,
+        ts: 1_700_000_000_000 + i,
+        userText: `user text ${i}`,
+        agentText: `agent text ${i}`,
+        summary: `summary ${i}`,
+        toolCalls: [],
+        value: 0,
+        alpha: 0,
+        priority: 0,
+        turnId: 1_700_000_000_000 + i,
+      })),
+    });
+
+    const result = await core.rebuildEmbeddings({ mode: "repair", limit: 100 });
+    // 4 rows × 2 slots — all failed, in exactly ONE embedMany call
+    // (no halving cascade: 8 slots would otherwise cost up to 15 calls).
+    expect(result.updated).toBe(0);
+    expect(result.failed).toBe(8);
+    expect(result.error).toMatch(/HTTP 503/);
+    expect(embedCalls).toBe(1);
+  });
+
+  /**
+   * Contract-violation diagnostics: when `embedMany` resolves with
+   * fewer vectors than inputs, the missing slots must fail WITH a
+   * `firstError` describing which slot got no vector — previously this
+   * mode was silent and indistinguishable from an update failure.
+   */
+  it("rebuildEmbeddings surfaces an error when embedMany returns short", async () => {
+    const deps = buildDeps(db!);
+    // Phase flag: import-time embedding must fail wholesale so both
+    // slots stay missing; only the rebuild call returns short.
+    let shortMode = false;
+    deps.embedder = {
+      ...deps.embedder!,
+      async embedMany(inputs) {
+        if (!shortMode) throw new Error("import-phase embedding disabled");
+        // Drop the last vector — a contract violation.
+        return inputs.slice(0, -1).map(() => {
+          const v = new Float32Array(TEST_EMBED_DIMENSIONS);
+          v[0] = 1;
+          return v;
+        });
+      },
+    } as typeof deps.embedder;
+
+    pipeline = createPipeline(deps);
+    core = createMemoryCore(
+      pipeline,
+      resolveHome("openclaw", "/tmp/memos-mc-test"),
+      "test",
+    );
+    await core.init();
+
+    await core.importBundle({
+      version: 1,
+      traces: [{
+        id: "tr_short",
+        episodeId: "ep_short",
+        sessionId: "se_short",
+        ts: 1_700_000_000_000,
+        userText: "user text long enough to embed",
+        agentText: "agent text long enough to embed",
+        summary: "summary of the short-return regression trace",
+        toolCalls: [],
+        value: 0,
+        alpha: 0,
+        priority: 0,
+        turnId: 1_700_000_000_000,
+      }],
+    });
+
+    const before = await core.embeddingMaintenanceStats();
+    expect(before.byKind.trace.missing).toBe(2);
+
+    shortMode = true;
+    const result = await core.rebuildEmbeddings({ mode: "repair", limit: 100 });
+    // 1 row × 2 slots; the dropped vector fails its slot with a message.
+    expect(result.updated).toBe(1);
+    expect(result.failed).toBe(1);
+    expect(result.error).toMatch(/returned no vector for slot/);
+  });
+
   it("does not require action vectors for lightweight memory traces", async () => {
     pipeline = createPipeline(buildDeps(db!, configWithLightweightMemory(true)));
     core = createMemoryCore(