Feat(Log): Add comprehensive diagnostic logs for /product/add flow

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.

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: glin1993@outlook.com

src/memos/api/handlers/add_handler.py

@@ -45,7 +45,7 @@ def handle_add_memories(self, add_req: APIADDRequest) -> MemoryResponse:
         Returns:
             MemoryResponse with added memory information
         """
-        self.logger.info(f"[AddHandler] Add Req is: {add_req}")
+        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,6 +788,7 @@ 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)}")
                             self.mem_scheduler.memos_message_queue.submit_messages(
                                 messages=[message_item]
                             )

src/memos/mem_scheduler/base_scheduler.py

@@ -594,6 +594,9 @@ def _submit_web_logs(
         Args:
             messages: Single log message or list of log messages
         """
+        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)}")
         if self.rabbitmq_config is None:
             return
 

src/memos/mem_scheduler/general_scheduler.py

@@ -475,6 +475,7 @@ 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"Messages {messages} assigned to {MEM_READ_LABEL} handler.")
 
         def process_message(message: ScheduleMessageItem):
@@ -539,6 +540,7 @@ 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}")
         """
         Process memories using mem_reader for enhanced memory processing.
 
@@ -636,7 +638,7 @@ def _process_memories_with_reader(
                                 }
                             )
                         if kb_log_content:
-                            logger.info("DIAGNOSTIC: Preparing to create event log for KB update in _process_memories_with_reader.")
+                            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,

src/memos/mem_scheduler/webservice_modules/rabbitmq_service.py

@@ -288,6 +288,7 @@ 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)}")
             try:
                 self.rabbitmq_channel.basic_publish(
                     exchange=self.rabbitmq_exchange_name,

src/memos/multi_mem_cube/single_cube.py

@@ -55,6 +55,8 @@ def add_memories(self, add_req: APIADDRequest) -> list[dict[str, Any]]:
         This is basically your current handle_add_memories logic,
         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)}")
         user_context = UserContext(
             user_id=add_req.user_id,
             mem_cube_id=self.cube_id,