refactor: Consolidate backward compatibility into API models; simplify handler logic

src/memos/api/handlers/add_handler.py

@@ -5,8 +5,6 @@
 using dependency injection for better modularity and testability.
 """
 
-from datetime import datetime
-
 from memos.api.handlers.base_handler import BaseHandler, HandlerDependencies
 from memos.api.product_models import APIADDRequest, MemoryResponse
 from memos.multi_mem_cube.composite_cube import CompositeCubeView
@@ -39,17 +37,13 @@ def handle_add_memories(self, add_req: APIADDRequest) -> MemoryResponse:
         supporting concurrent processing.
 
         Args:
-            add_req: Add memory request
+            add_req: Add memory request (deprecated fields are converted in model validator)
 
         Returns:
             MemoryResponse with added memory information
         """
         self.logger.info(f"[AddHandler] Add Req is: {add_req}")
 
-        if (not add_req.messages) and getattr(add_req, "memory_content", None):
-            add_req.messages = self._convert_content_messsage(add_req.memory_content)
-            self.logger.info(f"[AddHandler] Converted content to messages: {add_req.messages}")
-
         cube_view = self._build_cube_view(add_req)
 
         results = cube_view.add_memories(add_req)
@@ -65,16 +59,12 @@ def _resolve_cube_ids(self, add_req: APIADDRequest) -> list[str]:
         """
         Normalize target cube ids from add_req.
         Priority:
-        1) writable_cube_ids
-        2) mem_cube_id
-        3) fallback to user_id
+        1) writable_cube_ids (deprecated mem_cube_id is converted to this in model validator)
+        2) fallback to user_id
         """
-        if getattr(add_req, "writable_cube_ids", None):
+        if add_req.writable_cube_ids:
             return list(dict.fromkeys(add_req.writable_cube_ids))
 
-        if add_req.mem_cube_id:
-            return [add_req.mem_cube_id]
-
         return [add_req.user_id]
 
     def _build_cube_view(self, add_req: APIADDRequest) -> MemCubeView:
@@ -106,23 +96,3 @@ def _build_cube_view(self, add_req: APIADDRequest) -> MemCubeView:
                 cube_views=single_views,
                 logger=self.logger,
             )
-
-    def _convert_content_messsage(self, memory_content: str) -> list[dict[str, str]]:
-        """
-        Convert content string to list of message dictionaries.
-
-        Args:
-            content: add content string
-
-        Returns:
-            List of message dictionaries
-        """
-        messages_list = [
-            {
-                "role": "user",
-                "content": memory_content,
-                "chat_time": str(datetime.now().strftime("%Y-%m-%d %H:%M:%S")),
-            }
-        ]
-        # for only user-str input and convert message
-        return messages_list

src/memos/api/handlers/search_handler.py

@@ -63,16 +63,12 @@ def _resolve_cube_ids(self, search_req: APISearchRequest) -> list[str]:
         """
         Normalize target cube ids from search_req.
         Priority:
-        1) readable_cube_ids
-        2) mem_cube_id
-        3) fallback to user_id
+        1) readable_cube_ids (deprecated mem_cube_id is converted to this in model validator)
+        2) fallback to user_id
         """
-        if getattr(search_req, "readable_cube_ids", None):
+        if search_req.readable_cube_ids:
             return list(dict.fromkeys(search_req.readable_cube_ids))
 
-        if search_req.mem_cube_id:
-            return [search_req.mem_cube_id]
-
         return [search_req.user_id]
 
     def _build_cube_view(self, search_req: APISearchRequest) -> MemCubeView:

src/memos/api/product_models.py

@@ -2,13 +2,15 @@
 
 from typing import Any, Generic, Literal, TypeVar
 
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, model_validator
 
 # Import message types from core types module
+from memos.log import get_logger
 from memos.mem_scheduler.schemas.general_schemas import SearchMode
-from memos.types import MessageDict, PermissionDict
+from memos.types import MessageDict, MessagesType, PermissionDict
 
 
+logger = get_logger(__name__)
 T = TypeVar("T")
 
 
@@ -215,18 +217,11 @@ class APISearchRequest(BaseRequest):
     # ==== Basic inputs ====
     query: str = Field(
         ...,
-        description=("User search query"),
+        description="User search query",
     )
     user_id: str = Field(..., description="User ID")
 
     # ==== Cube scoping ====
-    mem_cube_id: str | None = Field(
-        None,
-        description=(
-            "(Deprecated) Single cube ID to search in. "
-            "Prefer `readable_cube_ids` for multi-cube search."
-        ),
-    )
     readable_cube_ids: list[str] | None = Field(
         None,
         description=(
@@ -297,7 +292,7 @@ class APISearchRequest(BaseRequest):
     )
 
     # ==== Context ====
-    chat_history: list[MessageDict] | None = Field(
+    chat_history: MessagesType | None = Field(
         None,
         description=(
             "Historical chat messages used internally by algorithms. "
@@ -307,6 +302,14 @@ class APISearchRequest(BaseRequest):
     )
 
     # ==== Backward compatibility ====
+    mem_cube_id: str | None = Field(
+        None,
+        description=(
+            "(Deprecated) Single cube ID to search in. "
+            "Prefer `readable_cube_ids` for multi-cube search."
+        ),
+    )
+
     moscube: bool = Field(
         False,
         description="(Deprecated / internal) Whether to use legacy MemOSCube path.",
@@ -317,6 +320,41 @@ class APISearchRequest(BaseRequest):
         description="(Internal) Operation definitions for multi-cube read permissions.",
     )
 
+    @model_validator(mode="after")
+    def _convert_deprecated_fields(self) -> "APISearchRequest":
+        """
+        Convert deprecated fields to new fields for backward compatibility.
+        Ensures full backward compatibility:
+            - mem_cube_id → readable_cube_ids
+            - moscube is ignored with warning
+            - operation ignored
+        """
+        # Convert mem_cube_id to readable_cube_ids (new field takes priority)
+        if self.mem_cube_id is not None:
+            if not self.readable_cube_ids:
+                self.readable_cube_ids = [self.mem_cube_id]
+            logger.warning(
+                "Deprecated field `mem_cube_id` is used in APISearchRequest. "
+                "It will be removed in a future version. "
+                "Please migrate to `readable_cube_ids`."
+            )
+
+        # Reject moscube if set to True (no longer supported)
+        if self.moscube:
+            logger.warning(
+                "Deprecated field `moscube` is used in APISearchRequest. "
+                "Legacy MemOSCube pipeline will be removed soon."
+            )
+
+        # Warn about operation (internal)
+        if self.operation:
+            logger.warning(
+                "Internal field `operation` is provided in APISearchRequest. "
+                "This field is deprecated and ignored."
+            )
+
+        return self
+
 
 class APIADDRequest(BaseRequest):
     """Request model for creating memories."""
@@ -328,12 +366,6 @@ class APIADDRequest(BaseRequest):
         description="Session ID. If not provided, a default session will be used.",
     )
 
-    # ==== Single-cube writing (Deprecated) ====
-    mem_cube_id: str | None = Field(
-        None,
-        description="(Deprecated) Target cube ID for this add request (optional for developer API).",
-    )
-
     # ==== Multi-cube writing ====
     writable_cube_ids: list[str] | None = Field(
         None, description="List of cube IDs user can write for multi-cube add"
@@ -374,7 +406,7 @@ class APIADDRequest(BaseRequest):
     )
 
     # ==== Input content ====
-    messages: list[MessageDict] | None = Field(
+    messages: MessagesType | None = Field(
         None,
         description=(
             "List of messages to store. Supports: "
@@ -390,7 +422,7 @@ class APIADDRequest(BaseRequest):
     )
 
     # ==== Chat history ====
-    chat_history: list[MessageDict] | None = Field(
+    chat_history: MessagesType | None = Field(
         None,
         description=(
             "Historical chat messages used internally by algorithms. "
@@ -406,6 +438,11 @@ class APIADDRequest(BaseRequest):
     )
 
     # ==== Backward compatibility fields (will delete later) ====
+    mem_cube_id: str | None = Field(
+        None,
+        description="(Deprecated) Target cube ID for this add request (optional for developer API).",
+    )
+
     memory_content: str | None = Field(
         None,
         description="(Deprecated) Plain memory content to store. Prefer using `messages`.",
@@ -426,6 +463,78 @@ class APIADDRequest(BaseRequest):
         description="(Internal) Operation definitions for multi-cube write permissions.",
     )
 
+    @model_validator(mode="after")
+    def _convert_deprecated_fields(self) -> "APIADDRequest":
+        """
+        Convert deprecated fields to new fields for backward compatibility.
+        This keeps the API fully backward-compatible while allowing
+        internal logic to use only the new fields.
+
+        Rules:
+            - mem_cube_id → writable_cube_ids
+            - memory_content → messages
+            - doc_path → messages (input_file)
+            - source → info["source"]
+            - operation → merged into writable_cube_ids (ignored otherwise)
+        """
+        # Convert mem_cube_id to writable_cube_ids (new field takes priority)
+        if self.mem_cube_id:
+            logger.warning(
+                "APIADDRequest.mem_cube_id is deprecated and will be removed in a future version. "
+                "Please use `writable_cube_ids` instead."
+            )
+            if not self.writable_cube_ids:
+                self.writable_cube_ids = [self.mem_cube_id]
+
+        # Handle deprecated operation field
+        if self.operation:
+            logger.warning(
+                "APIADDRequest.operation is deprecated and will be removed. "
+                "Use `writable_cube_ids` for multi-cube writes."
+            )
+
+        # Convert memory_content to messages (new field takes priority)
+        if self.memory_content:
+            logger.warning(
+                "APIADDRequest.memory_content is deprecated. "
+                "Use `messages` with a structured message instead."
+            )
+            if self.messages is None:
+                self.messages = []
+            self.messages.append(
+                {
+                    "type": "text",
+                    "text": self.memory_content,
+                }
+            )
+
+        # Handle deprecated doc_path
+        if self.doc_path:
+            logger.warning(
+                "APIADDRequest.doc_path is deprecated. "
+                "Use `messages` with an input_file item instead."
+            )
+            if self.messages is None:
+                self.messages = []
+            self.messages.append(
+                {
+                    "type": "file",
+                    "file": {"path": self.doc_path},
+                }
+            )
+
+        # Convert source to info.source_type (new field takes priority)
+        if self.source:
+            logger.warning(
+                "APIADDRequest.source is deprecated. "
+                "Use `info['source_type']` / `info['source_url']` instead."
+            )
+            if self.info is None:
+                self.info = {}
+            self.info.setdefault("source", self.source)
+
+        return self
+
 
 class APIChatCompleteRequest(BaseRequest):
     """Request model for chat operations."""

src/memos/types/types.py

@@ -27,6 +27,7 @@
     "MessageDict",
     "MessageList",
     "MessageRole",
+    "MessagesType",
     "Permission",
     "PermissionDict",
     "UserContext",