fix:fix scheduler logs and

Author: fridayL

src/memos/mem_os/product.py

@@ -23,7 +23,11 @@
     remove_embedding_recursive,
     sort_children_by_memory_type,
 )
-from memos.mem_scheduler.schemas import ANSWER_LABEL, QUERY_LABEL, ScheduleMessageItem
+from memos.mem_scheduler.schemas.general_schemas import (
+    ANSWER_LABEL,
+    QUERY_LABEL,
+)
+from memos.mem_scheduler.schemas.message_schemas import ScheduleMessageItem
 from memos.mem_user.persistent_user_manager import PersistentUserManager
 from memos.mem_user.user_manager import UserRole
 from memos.memories.textual.item import (
@@ -975,8 +979,6 @@ def search(
         self, query: str, user_id: str, install_cube_ids: list[str] | None = None, top_k: int = 10
     ):
         """Search memories for a specific user."""
-        # Validate user access
-        self._validate_user_access(user_id)
 
         # Load user cubes if not already loaded
         self._load_user_cubes(user_id, self.default_cube_config)

src/memos/mem_scheduler/base_scheduler.py

@@ -194,10 +194,10 @@ def replace_working_memory(
             text_mem_base: TreeTextMemory = text_mem_base
 
             # process rerank memories with llm
-            quey_history = self.monitor.query_monitors.get_queries_with_timesort()
+            query_history = self.monitor.query_monitors.get_queries_with_timesort()
             memories_with_new_order, rerank_success_flag = (
                 self.retriever.process_and_rerank_memories(
-                    queries=quey_history,
+                    queries=query_history,
                     original_memory=original_memory,
                     new_memory=new_memory,
                     top_k=self.top_k,
@@ -350,54 +350,63 @@ def update_activation_memory_periodically(
     ):
         new_activation_memories = []
 
-        if self.monitor.timed_trigger(
-            last_time=self.monitor.last_activation_mem_update_time,
-            interval_seconds=interval_seconds,
-        ):
-            logger.info(f"Updating activation memory for user {user_id} and mem_cube {mem_cube_id}")
+        try:
+            if self.monitor.timed_trigger(
+                last_time=self.monitor.last_activation_mem_update_time,
+                interval_seconds=interval_seconds,
+            ):
+                logger.info(
+                    f"Updating activation memory for user {user_id} and mem_cube {mem_cube_id}"
+                )
+
+                if (
+                    user_id not in self.monitor.working_memory_monitors
+                    or mem_cube_id not in self.monitor.working_memory_monitors[user_id]
+                    or len(self.monitor.working_memory_monitors[user_id][mem_cube_id].memories) == 0
+                ):
+                    logger.warning(
+                        "No memories found in working_memory_monitors, initializing from current working_memories"
+                    )
+                    self.initialize_working_memory_monitors(
+                        user_id=user_id,
+                        mem_cube_id=mem_cube_id,
+                        mem_cube=mem_cube,
+                    )
 
-            if len(self.monitor.working_memory_monitors[user_id][mem_cube_id].memories) == 0:
-                logger.warning(
-                    "No memories found in working_memory_monitors, initializing from current working_memories"
+                self.monitor.update_activation_memory_monitors(
+                    user_id=user_id, mem_cube_id=mem_cube_id, mem_cube=mem_cube
                 )
-                self.initialize_working_memory_monitors(
+
+                new_activation_memories = [
+                    m.memory_text
+                    for m in self.monitor.activation_memory_monitors[user_id][mem_cube_id].memories
+                ]
+
+                logger.info(
+                    f"Collected {len(new_activation_memories)} new memory entries for processing"
+                )
+
+                self.update_activation_memory(
+                    new_memories=new_activation_memories,
+                    label=label,
                     user_id=user_id,
                     mem_cube_id=mem_cube_id,
                     mem_cube=mem_cube,
                 )
 
-            self.monitor.update_activation_memory_monitors(
-                user_id=user_id, mem_cube_id=mem_cube_id, mem_cube=mem_cube
-            )
-
-            new_activation_memories = [
-                m.memory_text
-                for m in self.monitor.activation_memory_monitors[user_id][mem_cube_id].memories
-            ]
-
-            logger.info(
-                f"Collected {len(new_activation_memories)} new memory entries for processing"
-            )
-
-            self.update_activation_memory(
-                new_memories=new_activation_memories,
-                label=label,
-                user_id=user_id,
-                mem_cube_id=mem_cube_id,
-                mem_cube=mem_cube,
-            )
-
-            self.monitor.last_activation_mem_update_time = datetime.now()
+                self.monitor.last_activation_mem_update_time = datetime.now()
 
-            logger.debug(
-                f"Activation memory update completed at {self.monitor.last_activation_mem_update_time}"
-            )
-        else:
-            logger.info(
-                f"Skipping update - {interval_seconds} second interval not yet reached. "
-                f"Last update time is {self.monitor.last_activation_mem_update_time} and now is"
-                f"{datetime.now()}"
-            )
+                logger.debug(
+                    f"Activation memory update completed at {self.monitor.last_activation_mem_update_time}"
+                )
+            else:
+                logger.info(
+                    f"Skipping update - {interval_seconds} second interval not yet reached. "
+                    f"Last update time is {self.monitor.last_activation_mem_update_time} and now is"
+                    f"{datetime.now()}"
+                )
+        except Exception as e:
+            logger.error(f"Error: {e}", exc_info=True)
 
     def submit_messages(self, messages: ScheduleMessageItem | list[ScheduleMessageItem]):
         """Submit multiple messages to the message queue."""

src/memos/mem_scheduler/general_scheduler.py

@@ -145,43 +145,45 @@ def _add_message_consumer(self, messages: list[ScheduleMessageItem]) -> None:
         grouped_messages = self.dispatcher.group_messages_by_user_and_cube(messages=messages)
 
         self.validate_schedule_messages(messages=messages, label=ADD_LABEL)
-
-        for user_id in grouped_messages:
-            for mem_cube_id in grouped_messages[user_id]:
-                messages = grouped_messages[user_id][mem_cube_id]
-                if len(messages) == 0:
-                    return
-
-                # for status update
-                self._set_current_context_from_message(msg=messages[0])
-
-                # submit logs
-                for msg in messages:
-                    userinput_memory_ids = json.loads(msg.content)
-                    mem_cube = msg.mem_cube
-                    for memory_id in userinput_memory_ids:
-                        mem_item: TextualMemoryItem = mem_cube.text_mem.get(memory_id=memory_id)
-                        mem_type = mem_item.meta_data.memory_type
-                        mem_content = mem_item.memory
-
-                        self.log_adding_memory(
-                            memory=mem_content,
-                            memory_type=mem_type,
-                            user_id=msg.user_id,
-                            mem_cube_id=msg.mem_cube_id,
-                            mem_cube=msg.mem_cube,
-                            log_func_callback=self._submit_web_logs,
+        try:
+            for user_id in grouped_messages:
+                for mem_cube_id in grouped_messages[user_id]:
+                    messages = grouped_messages[user_id][mem_cube_id]
+                    if len(messages) == 0:
+                        return
+
+                    # for status update
+                    self._set_current_context_from_message(msg=messages[0])
+
+                    # submit logs
+                    for msg in messages:
+                        userinput_memory_ids = json.loads(msg.content)
+                        mem_cube = msg.mem_cube
+                        for memory_id in userinput_memory_ids:
+                            mem_item: TextualMemoryItem = mem_cube.text_mem.get(memory_id=memory_id)
+                            mem_type = mem_item.metadata.memory_type
+                            mem_content = mem_item.memory
+
+                            self.log_adding_memory(
+                                memory=mem_content,
+                                memory_type=mem_type,
+                                user_id=msg.user_id,
+                                mem_cube_id=msg.mem_cube_id,
+                                mem_cube=msg.mem_cube,
+                                log_func_callback=self._submit_web_logs,
+                            )
+
+                    # update activation memories
+                    if self.enable_act_memory_update:
+                        self.update_activation_memory_periodically(
+                            interval_seconds=self.monitor.act_mem_update_interval,
+                            label=ADD_LABEL,
+                            user_id=user_id,
+                            mem_cube_id=mem_cube_id,
+                            mem_cube=messages[0].mem_cube,
                         )
-
-                # update activation memories
-                if self.enable_act_memory_update:
-                    self.update_activation_memory_periodically(
-                        interval_seconds=self.monitor.act_mem_update_interval,
-                        label=ADD_LABEL,
-                        user_id=user_id,
-                        mem_cube_id=mem_cube_id,
-                        mem_cube=messages[0].mem_cube,
-                    )
+        except Exception as e:
+            logger.error(f"Error: {e}", exc_info=True)
 
     def process_session_turn(
         self,