Feat(Log): Add comprehensive diagnostic logs for /product/add flow and apply ruff formatting

Introduces detailed INFO level diagnostic logs across the entire call chain
for the /product/add API endpoint. These logs include relevant context,
such as full request bodies, message items before scheduler submission,
and messages before RabbitMQ publication, to aid in debugging deployment
discrepancies and tracing data flow, especially concerning task_id
propagation.

Also applies automatic code formatting using ruff format to all modified files.

Logs added/enhanced in:
- src/memos/api/routers/product_router.py
- src/memos/api/handlers/add_handler.py
- src/memos/multi_mem_cube/single_cube.py
- src/memos/mem_os/core.py
- src/memos/mem_scheduler/general_scheduler.py
- src/memos/mem_scheduler/base_scheduler.py
- src/memos/mem_scheduler/webservice_modules/rabbitmq_service.py

Author: [email protected]

src/memos/api/handlers/add_handler.py

@@ -45,7 +45,9 @@ def handle_add_memories(self, add_req: APIADDRequest) -> MemoryResponse:
         Returns:
             MemoryResponse with added memory information
         """
-        self.logger.info(f"[DIAGNOSTIC] server_router -> add_handler.handle_add_memories called. Full request: {add_req.model_dump_json(indent=2)}")
+        self.logger.info(
+            f"[DIAGNOSTIC] server_router -> add_handler.handle_add_memories called. Full request: {add_req.model_dump_json(indent=2)}"
+        )
 
         if add_req.info:
             exclude_fields = list_all_fields()

src/memos/mem_os/core.py

@@ -788,7 +788,9 @@ def process_textual_memory():
                                 timestamp=datetime.utcnow(),
                                 task_id=task_id,
                             )
-                            logger.info(f"[DIAGNOSTIC] core.add: Submitting message to scheduler: {message_item.model_dump_json(indent=2)}")
+                            logger.info(
+                                f"[DIAGNOSTIC] core.add: Submitting message to scheduler: {message_item.model_dump_json(indent=2)}"
+                            )
                             self.mem_scheduler.memos_message_queue.submit_messages(
                                 messages=[message_item]
                             )

src/memos/mem_scheduler/base_scheduler.py

@@ -596,7 +596,9 @@ def _submit_web_logs(
         """
         messages_list = [messages] if isinstance(messages, ScheduleLogForWebItem) else messages
         for message in messages_list:
-            logger.info(f"[DIAGNOSTIC] base_scheduler._submit_web_logs called. Message to publish: {message.model_dump_json(indent=2)}")
+            logger.info(
+                f"[DIAGNOSTIC] base_scheduler._submit_web_logs called. Message to publish: {message.model_dump_json(indent=2)}"
+            )
         if self.rabbitmq_config is None:
             return
 

src/memos/mem_scheduler/general_scheduler.py

@@ -380,7 +380,9 @@ def _add_message_consumer(self, messages: list[ScheduleMessageItem]) -> None:
                                     memcube_name=self._map_memcube_name(msg.mem_cube_id),
                                 )
                                 # 3. 后置赋值 log_content
-                                event.log_content = f"Knowledge Base Memory Update: {len(kb_log_content)} changes."
+                                event.log_content = (
+                                    f"Knowledge Base Memory Update: {len(kb_log_content)} changes."
+                                )
                                 event.task_id = msg.task_id
                                 self._submit_web_logs([event])
                         else:
@@ -475,7 +477,9 @@ def _add_message_consumer(self, messages: list[ScheduleMessageItem]) -> None:
             logger.error(f"Error: {e}", exc_info=True)
 
     def _mem_read_message_consumer(self, messages: list[ScheduleMessageItem]) -> None:
-        logger.info(f"[DIAGNOSTIC] general_scheduler._mem_read_message_consumer called. Received messages: {[msg.model_dump_json(indent=2) for msg in messages]}")
+        logger.info(
+            f"[DIAGNOSTIC] general_scheduler._mem_read_message_consumer called. Received messages: {[msg.model_dump_json(indent=2) for msg in messages]}"
+        )
         logger.info(f"Messages {messages} assigned to {MEM_READ_LABEL} handler.")
 
         def process_message(message: ScheduleMessageItem):
@@ -540,7 +544,9 @@ def _process_memories_with_reader(
         task_id: str | None = None,
         info: dict | None = None,
     ) -> None:
-        logger.info(f"[DIAGNOSTIC] general_scheduler._process_memories_with_reader called. mem_ids: {mem_ids}, user_id: {user_id}, mem_cube_id: {mem_cube_id}, task_id: {task_id}")
+        logger.info(
+            f"[DIAGNOSTIC] general_scheduler._process_memories_with_reader called. mem_ids: {mem_ids}, user_id: {user_id}, mem_cube_id: {mem_cube_id}, task_id: {task_id}"
+        )
         """
         Process memories using mem_reader for enhanced memory processing.
 
@@ -638,7 +644,9 @@ def _process_memories_with_reader(
                                 }
                             )
                         if kb_log_content:
-                            logger.info(f"[DIAGNOSTIC] general_scheduler._process_memories_with_reader: Creating event log for KB update. Label: knowledgeBaseUpdate, user_id: {user_id}, mem_cube_id: {mem_cube_id}, task_id: {task_id}. KB content: {json.dumps(kb_log_content, indent=2)}")
+                            logger.info(
+                                f"[DIAGNOSTIC] general_scheduler._process_memories_with_reader: Creating event log for KB update. Label: knowledgeBaseUpdate, user_id: {user_id}, mem_cube_id: {mem_cube_id}, task_id: {task_id}. KB content: {json.dumps(kb_log_content, indent=2)}"
+                            )
                             event = self.create_event_log(
                                 label="knowledgeBaseUpdate",
                                 from_memory_type=USER_INPUT_TYPE,
@@ -651,7 +659,9 @@ def _process_memories_with_reader(
                                 memory_len=len(kb_log_content),
                                 memcube_name=self._map_memcube_name(mem_cube_id),
                             )
-                            event.log_content = f"Knowledge Base Memory Update: {len(kb_log_content)} changes."
+                            event.log_content = (
+                                f"Knowledge Base Memory Update: {len(kb_log_content)} changes."
+                            )
                             event.task_id = task_id
                             self._submit_web_logs([event])
                     else:

src/memos/mem_scheduler/webservice_modules/rabbitmq_service.py

@@ -5,6 +5,7 @@
 import time
 
 from pathlib import Path
+
 from memos.configs.mem_scheduler import AuthConfig, RabbitMQConfig
 from memos.context.context import ContextThread
 from memos.dependency import require_python_package
@@ -288,7 +289,9 @@ def rabbitmq_publish_message(self, message: dict):
                 logger.error("Cannot publish - no active connection")
                 return False
 
-            logger.info(f"[DIAGNOSTIC] rabbitmq_service.rabbitmq_publish_message: Attempting to publish message. Exchange: {self.rabbitmq_exchange_name}, Routing Key: {self.rabbit_queue_name}, Message Content: {json.dumps(message, indent=2)}")
+            logger.info(
+                f"[DIAGNOSTIC] rabbitmq_service.rabbitmq_publish_message: Attempting to publish message. Exchange: {self.rabbitmq_exchange_name}, Routing Key: {self.rabbit_queue_name}, Message Content: {json.dumps(message, indent=2)}"
+            )
             try:
                 self.rabbitmq_channel.basic_publish(
                     exchange=self.rabbitmq_exchange_name,

src/memos/multi_mem_cube/single_cube.py

@@ -56,7 +56,9 @@ def add_memories(self, add_req: APIADDRequest) -> list[dict[str, Any]]:
         but scoped to a single cube_id.
         """
         sync_mode = add_req.async_mode or self._get_sync_mode()
-        self.logger.info(f"[DIAGNOSTIC] single_cube.add_memories called for cube_id: {self.cube_id}. sync_mode: {sync_mode}. Request: {add_req.model_dump_json(indent=2)}")
+        self.logger.info(
+            f"[DIAGNOSTIC] single_cube.add_memories called for cube_id: {self.cube_id}. sync_mode: {sync_mode}. Request: {add_req.model_dump_json(indent=2)}"
+        )
         user_context = UserContext(
             user_id=add_req.user_id,
             mem_cube_id=self.cube_id,