refactor & fix bugs: fix a range of bugs in scheduler and revise fine add apis

.gitignore

@@ -204,6 +204,7 @@ cython_debug/
 #  and can be added to the global gitignore or merged into this file.  For a more nuclear
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 .idea/
+.trae
 
 # VSCode
 .vscode*

docker/.env.example

@@ -123,6 +123,7 @@ API_SCHEDULER_ON=true
 API_SEARCH_WINDOW_SIZE=5
 # Specify how many rounds of previous conversations (history) to retrieve and consider during the 'hybrid search' (fast search+asynchronous fine search). This helps provide context aware search results
 API_SEARCH_HISTORY_TURNS=5
+MEMSCHEDULER_USE_REDIS_QUEUE=false
 
 ## Graph / vector stores
 # Neo4j database selection mode

examples/mem_scheduler/quick_start_examples.py

@@ -146,7 +146,9 @@ def kv_cache_only():
 
 def run_scheduler_example():
     # 使用 MemScheduler 加载主 MOS(Memory-Oriented System)配置文件
-    config = parse_yaml("./examples/data/config/mem_scheduler/memos_config_w_scheduler.yaml")
+    config = parse_yaml(
+        f"{BASE_DIR}/examples/data/config/mem_scheduler/memos_config_w_scheduler.yaml"
+    )
     # 将解析出的配置字典传入 MOSConfig 构造器, 构建配置对象
     mos_config = MOSConfig(**config)
     # 使用配置对象初始化 MOS 系统实例
@@ -159,12 +161,12 @@ def run_scheduler_example():
 
     # 从 YAML 文件加载 MemCube(记忆立方体)的通用配置
     config = GeneralMemCubeConfig.from_yaml_file(
-        "./examples/data/config/mem_scheduler/mem_cube_config.yaml"
+        f"{BASE_DIR}/examples/data/config/mem_scheduler/mem_cube_config.yaml"
     )
     # 定义 MemCube 的唯一标识符
     mem_cube_id = "mem_cube_5"
     # 定义 MemCube 的本地存储路径(路径中包含用户 ID 和 MemCube ID)
-    mem_cube_name_or_path = f"./outputs/mem_scheduler/{user_id}/{mem_cube_id}"
+    mem_cube_name_or_path = f"{BASE_DIR}/outputs/mem_scheduler/{user_id}/{mem_cube_id}"
 
     # 如果该路径已存在, 则先删除旧目录
     if Path(mem_cube_name_or_path).exists():

examples/mem_scheduler/scheduler_for_async_tasks.py

@@ -57,7 +57,7 @@ def submit_tasks():
 TEST_HANDLER_LABEL = "test_handler"
 mem_scheduler.register_handlers({TEST_HANDLER_LABEL: my_test_handler})
 
-# 10s to restart
+# 5s to restart
 mem_scheduler.orchestrator.tasks_min_idle_ms[TEST_HANDLER_LABEL] = 5_000
 
 tmp_dir = Path("./tmp")

src/memos/mem_reader/simple_struct.py

@@ -614,11 +614,9 @@ def _read_memory(
                     serialized_origin_memories = json.dumps(
                         [one.memory for one in original_memory_group], indent=2
                     )
-                    revised_memory_list = self.rewrite_memories(
+                    revised_memory_list = self.filter_hallucination_in_memories(
                         messages=combined_messages,
                         memory_list=original_memory_group,
-                        user_only=os.getenv("SIMPLE_STRUCT_REWRITE_USER_ONLY", "true").lower()
-                        == "false",
                     )
                     serialized_revised_memories = json.dumps(
                         [one.memory for one in revised_memory_list], indent=2

src/memos/mem_scheduler/base_scheduler.py

@@ -225,7 +225,7 @@ def initialize_modules(
             process_llm = chat_llm
 
         try:
-            if redis_client:
+            if redis_client and self.use_redis_queue:
                 self.status_tracker = TaskStatusTracker(redis_client)
                 if self.dispatcher:
                     self.dispatcher.status_tracker = self.status_tracker
@@ -305,7 +305,7 @@ def status_tracker(self) -> TaskStatusTracker | None:
         available via RedisSchedulerModule. This mirrors the lazy pattern used
         by `mem_cube` so downstream modules can safely access the tracker.
         """
-        if self._status_tracker is None:
+        if self._status_tracker is None and self.use_redis_queue:
             try:
                 self._status_tracker = TaskStatusTracker(self.redis)
                 # Propagate to submodules when created lazily
@@ -314,7 +314,8 @@ def status_tracker(self) -> TaskStatusTracker | None:
                 if self.memos_message_queue:
                     self.memos_message_queue.set_status_tracker(self._status_tracker)
             except Exception as e:
-                logger.warning(f"Failed to lazily initialize status_tracker: {e}", exc_info=True)
+                logger.warning(f"Failed to lazy-initialize status_tracker: {e}", exc_info=True)
+
         return self._status_tracker
 
     @status_tracker.setter
@@ -869,6 +870,8 @@ def _submit_web_logs(
             messages = [messages]  # transform single message to list
 
         for message in messages:
+            if self.rabbitmq_config is None:
+                return
             try:
                 # Always call publish; the publisher now caches when offline and flushes after reconnect
                 logger.info(

src/memos/mem_scheduler/task_schedule_modules/dispatcher.py

@@ -108,8 +108,6 @@ def __init__(
         )
 
         self.metrics = metrics
-        self._status_tracker: TaskStatusTracker | None = None
-        # Use setter to allow propagation and keep a single source of truth
         self.status_tracker = status_tracker
         self.submit_web_logs = submit_web_logs  # ADDED
 
@@ -118,35 +116,6 @@ def on_messages_enqueued(self, msgs: list[ScheduleMessageItem]) -> None:
             return
         # This is handled in BaseScheduler now
 
-    @property
-    def status_tracker(self) -> TaskStatusTracker | None:
-        """Lazy-initialized status tracker for the dispatcher.
-
-        If the tracker is None, attempt to initialize from the Redis-backed
-        components available to the dispatcher (queue or orchestrator).
-        """
-        if self._status_tracker is None:
-            try:
-                self._status_tracker = TaskStatusTracker(self.redis)
-                # Propagate to submodules when created lazily
-                if self.memos_message_queue:
-                    self.memos_message_queue.set_status_tracker(self._status_tracker)
-            except Exception as e:
-                logger.warning(f"Failed to lazily initialize status_tracker: {e}", exc_info=True)
-        return self._status_tracker
-
-    @status_tracker.setter
-    def status_tracker(self, value: TaskStatusTracker | None) -> None:
-        self._status_tracker = value
-        # Propagate to the queue if possible
-        try:
-            if self.memos_message_queue and hasattr(self.memos_message_queue, "status_tracker"):
-                self.memos_message_queue.status_tracker = value
-        except Exception as e:
-            logger.warning(
-                f"Failed to propagate dispatcher status_tracker to queue: {e}", exc_info=True
-            )
-
     def _create_task_wrapper(self, handler: Callable, task_item: RunningTaskItem):
         """
         Create a wrapper around the handler to track task execution and capture results.