new feat of scheduler

examples/mem_scheduler/task_stop_rerun.py

@@ -25,9 +25,9 @@ def my_test_handler(messages: list[ScheduleMessageItem]):
         task_id = str(msg.item_id)
         file_path = tmp_dir / f"{task_id}.txt"
         try:
-            print(f"writing {file_path}...")
-            file_path.write_text(f"Task {task_id} processed.\n")
             sleep(5)
+            file_path.write_text(f"Task {task_id} processed.\n")
+            print(f"writing {file_path} done")
         except Exception as e:
             print(f"Failed to write {file_path}: {e}")
 
@@ -89,4 +89,5 @@ def submit_tasks():
 
 # 7. Stop the scheduler
 print("Stopping the scheduler...")
+sleep(5)
 mem_scheduler.stop()

src/memos/mem_reader/simple_struct.py

@@ -453,7 +453,7 @@ def get_memory(
     @staticmethod
     def _parse_hallucination_filter_response(text: str) -> tuple[bool, dict[int, dict]]:
         """Parse index-keyed JSON from hallucination filter response.
-        Expected shape: { "0": {"if_delete": bool, "rewritten memory content": str}, ... }
+        Expected shape: { "0": {"delete": bool, "rewritten": str, "reason": str}, ... }
         Returns (success, parsed_dict) with int keys.
         """
         try:
@@ -476,54 +476,82 @@ def _parse_hallucination_filter_response(text: str) -> tuple[bool, dict[int, dic
                     continue
             if not isinstance(v, dict):
                 continue
-            delete_flag = v.get("delete_flag")
-            rewritten = v.get("rewritten memory content", "")
-            if isinstance(delete_flag, bool) and isinstance(rewritten, str):
-                result[idx] = {"delete_flag": delete_flag, "rewritten memory content": rewritten}
+            delete_flag = v.get("delete")
+            rewritten = v.get("rewritten", "")
+            reason = v.get("reason", "")
+            if (
+                isinstance(delete_flag, bool)
+                and isinstance(rewritten, str)
+                and isinstance(reason, str)
+            ):
+                result[idx] = {"delete": delete_flag, "rewritten": rewritten, "reason": reason}
 
         return (len(result) > 0), result
 
     def filter_hallucination_in_memories(
-        self, user_messages: list[str], memory_list: list[list[TextualMemoryItem]]
-    ):
-        filtered_memory_list = []
-        for group in memory_list:
-            try:
-                flat_memories = [one.memory for one in group]
-                template = PROMPT_MAPPING["hallucination_filter"]
-                prompt_args = {
-                    "user_messages_inline": "\n".join(user_messages),
-                    "memories_inline": json.dumps(flat_memories, ensure_ascii=False, indent=2),
-                }
-                prompt = template.format(**prompt_args)
+        self, user_messages: list[str], memory_list: list[TextualMemoryItem]
+    ) -> list[TextualMemoryItem]:
+        flat_memories = [one.memory for one in memory_list]
+        template = PROMPT_MAPPING["hallucination_filter"]
+        prompt_args = {
+            "user_messages_inline": "\n".join([f"- {memory}" for memory in user_messages]),
+            "memories_inline": json.dumps(
+                {str(i): memory for i, memory in enumerate(flat_memories)},
+                ensure_ascii=False,
+                indent=2,
+            ),
+        }
+        prompt = template.format(**prompt_args)
 
-                # Optionally run filter and parse the output
-                try:
-                    raw = self.llm.generate(prompt)
-                    success, parsed = self._parse_hallucination_filter_response(raw)
-                    logger.info(f"Hallucination filter parsed successfully: {success}")
-                    new_mem_list = []
-                    if success:
-                        logger.info(f"Hallucination filter result: {parsed}")
-                        for mem_idx, (delete_flag, rewritten_mem_content) in parsed.items():
-                            if not delete_flag:
-                                group[mem_idx].memory = rewritten_mem_content
-                                new_mem_list.append(group[mem_idx])
-                        filtered_memory_list.append(new_mem_list)
-                        logger.info(
-                            f"Successfully transform origianl memories from {group} to {new_mem_list}."
-                        )
-                    else:
+        # Optionally run filter and parse the output
+        try:
+            raw = self.llm.generate([{"role": "user", "content": prompt}])
+            success, parsed = self._parse_hallucination_filter_response(raw)
+            logger.info(
+                f"[filter_hallucination_in_memories] Hallucination filter parsed successfully: {success}"
+            )
+            if success:
+                logger.info(f"Hallucination filter result: {parsed}")
+                total = len(memory_list)
+                keep_flags = [True] * total
+                for mem_idx, content in parsed.items():
+                    # Validate index bounds
+                    if not isinstance(mem_idx, int) or mem_idx < 0 or mem_idx >= total:
                         logger.warning(
-                            "Hallucination filter parsing failed or returned empty result."
+                            f"[filter_hallucination_in_memories] Ignoring out-of-range index: {mem_idx}"
                         )
-                except Exception as e:
-                    logger.error(f"Hallucination filter execution error: {e}", stack_info=True)
-                    filtered_memory_list.append(group)
-            except Exception:
-                logger.error("Fail to filter memories", stack_info=True)
-                filtered_memory_list.append(group)
-        return filtered_memory_list
+                        continue
+
+                    delete_flag = content.get("delete", False)
+                    rewritten = content.get("rewritten", None)
+                    reason = content.get("reason", "")
+
+                    logger.info(
+                        f"[filter_hallucination_in_memories] index={mem_idx}, delete={delete_flag}, rewritten='{(rewritten or '')[:100]}', reason='{reason[:120]}'"
+                    )
+
+                    if delete_flag is True and rewritten is not None:
+                        # Mark for deletion
+                        keep_flags[mem_idx] = False
+                    else:
+                        # Apply rewrite if provided (safe-by-default: keep item when not mentioned or delete=False)
+                        try:
+                            if isinstance(rewritten, str):
+                                memory_list[mem_idx].memory = rewritten
+                        except Exception as e:
+                            logger.warning(
+                                f"[filter_hallucination_in_memories] Failed to apply rewrite for index {mem_idx}: {e}"
+                            )
+
+                # Build result, preserving original order; keep items not mentioned by LLM by default
+                new_mem_list = [memory_list[i] for i in range(total) if keep_flags[i]]
+                return new_mem_list
+            else:
+                logger.warning("Hallucination filter parsing failed or returned empty result.")
+        except Exception as e:
+            logger.error(f"Hallucination filter execution error: {e}", stack_info=True)
+
+        return memory_list
 
     def _read_memory(
         self, messages: list[MessagesType], type: str, info: dict[str, Any], mode: str = "fine"
@@ -572,11 +600,16 @@ def _read_memory(
 
         if os.getenv("SIMPLE_STRUCT_ADD_FILTER", "false") == "true":
             # Build inputs
-            user_messages = [msg.content for msg in messages if msg.role == "user"]
-            memory_list = self.filter_hallucination_in_memories(
-                user_messages=user_messages, memory_list=memory_list
-            )
-
+            new_memory_list = []
+            for unit_messages, unit_memory_list in zip(messages, memory_list, strict=False):
+                unit_user_messages = [
+                    msg["content"] for msg in unit_messages if msg["role"] == "user"
+                ]
+                unit_memory_list = self.filter_hallucination_in_memories(
+                    user_messages=unit_user_messages, memory_list=unit_memory_list
+                )
+                new_memory_list.append(unit_memory_list)
+            memory_list = new_memory_list
         return memory_list
 
     def fine_transfer_simple_mem(

src/memos/mem_scheduler/general_scheduler.py

@@ -66,7 +66,7 @@ def __init__(self, config: GeneralSchedulerConfig):
     def long_memory_update_process(
         self, user_id: str, mem_cube_id: str, messages: list[ScheduleMessageItem]
     ):
-        mem_cube = self.current_mem_cube
+        mem_cube = self.mem_cube
 
         # update query monitors
         for msg in messages:
@@ -109,8 +109,8 @@ def long_memory_update_process(
 
             query_db_manager = self.monitor.query_monitors[user_id][mem_cube_id]
             query_db_manager.obj.put(item=item)
-            # Sync with database after adding new item
-            query_db_manager.sync_with_orm()
+        # Sync with database after adding new item
+        query_db_manager.sync_with_orm()
         logger.debug(
             f"Queries in monitor for user_id={user_id}, mem_cube_id={mem_cube_id}: {query_db_manager.obj.get_queries_with_timesort()}"
         )
@@ -162,7 +162,7 @@ def long_memory_update_process(
                 label=QUERY_TASK_LABEL,
                 user_id=user_id,
                 mem_cube_id=mem_cube_id,
-                mem_cube=self.current_mem_cube,
+                mem_cube=self.mem_cube,
             )
 
     def _add_message_consumer(self, messages: list[ScheduleMessageItem]) -> None:
@@ -249,7 +249,7 @@ def _query_message_consumer(self, messages: list[ScheduleMessageItem]) -> None:
                             to_memory_type=NOT_APPLICABLE_TYPE,
                             user_id=msg.user_id,
                             mem_cube_id=msg.mem_cube_id,
-                            mem_cube=self.current_mem_cube,
+                            mem_cube=self.mem_cube,
                             memcube_log_content=[
                                 {
                                     "content": f"[User] {msg.content}",
@@ -305,7 +305,7 @@ def _answer_message_consumer(self, messages: list[ScheduleMessageItem]) -> None:
                             to_memory_type=NOT_APPLICABLE_TYPE,
                             user_id=msg.user_id,
                             mem_cube_id=msg.mem_cube_id,
-                            mem_cube=self.current_mem_cube,
+                            mem_cube=self.mem_cube,
                             memcube_log_content=[
                                 {
                                     "content": f"[Assistant] {msg.content}",
@@ -338,7 +338,7 @@ def log_add_messages(self, msg: ScheduleMessageItem):
             try:
                 # This mem_item represents the NEW content that was just added/processed
                 mem_item: TextualMemoryItem | None = None
-                mem_item = self.current_mem_cube.text_mem.get(
+                mem_item = self.mem_cube.text_mem.get(
                     memory_id=memory_id, user_name=msg.mem_cube_id
                 )
                 if mem_item is None:
@@ -352,8 +352,8 @@ def log_add_messages(self, msg: ScheduleMessageItem):
                 original_item_id = None
 
                 # Only check graph_store if a key exists and the text_mem has a graph_store
-                if key and hasattr(self.current_mem_cube.text_mem, "graph_store"):
-                    candidates = self.current_mem_cube.text_mem.graph_store.get_by_metadata(
+                if key and hasattr(self.mem_cube.text_mem, "graph_store"):
+                    candidates = self.mem_cube.text_mem.graph_store.get_by_metadata(
                         [
                             {"field": "key", "op": "=", "value": key},
                             {
@@ -368,7 +368,7 @@ def log_add_messages(self, msg: ScheduleMessageItem):
                         original_item_id = candidates[0]
                         # Crucial step: Fetch the original content for updates
                         # This `get` is for the *existing* memory that will be updated
-                        original_mem_item = self.current_mem_cube.text_mem.get(
+                        original_mem_item = self.mem_cube.text_mem.get(
                             memory_id=original_item_id, user_name=msg.mem_cube_id
                         )
                         original_content = original_mem_item.memory
@@ -481,7 +481,7 @@ def send_add_log_messages_to_local_env(
                 to_memory_type=LONG_TERM_MEMORY_TYPE,
                 user_id=msg.user_id,
                 mem_cube_id=msg.mem_cube_id,
-                mem_cube=self.current_mem_cube,
+                mem_cube=self.mem_cube,
                 memcube_log_content=add_content_legacy,
                 metadata=add_meta_legacy,
                 memory_len=len(add_content_legacy),
@@ -496,7 +496,7 @@ def send_add_log_messages_to_local_env(
                 to_memory_type=LONG_TERM_MEMORY_TYPE,
                 user_id=msg.user_id,
                 mem_cube_id=msg.mem_cube_id,
-                mem_cube=self.current_mem_cube,
+                mem_cube=self.mem_cube,
                 memcube_log_content=update_content_legacy,
                 metadata=update_meta_legacy,
                 memory_len=len(update_content_legacy),
@@ -562,7 +562,7 @@ def send_add_log_messages_to_cloud_env(
                 to_memory_type=LONG_TERM_MEMORY_TYPE,
                 user_id=msg.user_id,
                 mem_cube_id=msg.mem_cube_id,
-                mem_cube=self.current_mem_cube,
+                mem_cube=self.mem_cube,
                 memcube_log_content=kb_log_content,
                 metadata=None,
                 memory_len=len(kb_log_content),
@@ -577,7 +577,7 @@ def _mem_feedback_message_consumer(self, messages: list[ScheduleMessageItem]) ->
             if not messages:
                 return
             message = messages[0]
-            mem_cube = self.current_mem_cube
+            mem_cube = self.mem_cube
 
             user_id = message.user_id
             mem_cube_id = message.mem_cube_id
@@ -744,9 +744,9 @@ def process_message(message: ScheduleMessageItem):
             try:
                 user_id = message.user_id
                 mem_cube_id = message.mem_cube_id
-                mem_cube = self.current_mem_cube
+                mem_cube = self.mem_cube
                 if mem_cube is None:
-                    logger.warning(
+                    logger.error(
                         f"mem_cube is None for user_id={user_id}, mem_cube_id={mem_cube_id}, skipping processing",
                         stack_info=True,
                     )
@@ -923,7 +923,7 @@ def _process_memories_with_reader(
                                 to_memory_type=LONG_TERM_MEMORY_TYPE,
                                 user_id=user_id,
                                 mem_cube_id=mem_cube_id,
-                                mem_cube=self.current_mem_cube,
+                                mem_cube=self.mem_cube,
                                 memcube_log_content=kb_log_content,
                                 metadata=None,
                                 memory_len=len(kb_log_content),
@@ -968,7 +968,7 @@ def _process_memories_with_reader(
                                 to_memory_type=LONG_TERM_MEMORY_TYPE,
                                 user_id=user_id,
                                 mem_cube_id=mem_cube_id,
-                                mem_cube=self.current_mem_cube,
+                                mem_cube=self.mem_cube,
                                 memcube_log_content=add_content_legacy,
                                 metadata=add_meta_legacy,
                                 memory_len=len(add_content_legacy),
@@ -1036,7 +1036,7 @@ def _process_memories_with_reader(
                         to_memory_type=LONG_TERM_MEMORY_TYPE,
                         user_id=user_id,
                         mem_cube_id=mem_cube_id,
-                        mem_cube=self.current_mem_cube,
+                        mem_cube=self.mem_cube,
                         memcube_log_content=kb_log_content,
                         metadata=None,
                         memory_len=len(kb_log_content),
@@ -1054,7 +1054,7 @@ def process_message(message: ScheduleMessageItem):
             try:
                 user_id = message.user_id
                 mem_cube_id = message.mem_cube_id
-                mem_cube = self.current_mem_cube
+                mem_cube = self.mem_cube
                 if mem_cube is None:
                     logger.warning(
                         f"mem_cube is None for user_id={user_id}, mem_cube_id={mem_cube_id}, skipping processing"
@@ -1284,7 +1284,7 @@ def _pref_add_message_consumer(self, messages: list[ScheduleMessageItem]) -> Non
 
         def process_message(message: ScheduleMessageItem):
             try:
-                mem_cube = self.current_mem_cube
+                mem_cube = self.mem_cube
                 if mem_cube is None:
                     logger.warning(
                         f"mem_cube is None for user_id={message.user_id}, mem_cube_id={message.mem_cube_id}, skipping processing"

src/memos/mem_scheduler/optimized_scheduler.py

@@ -230,7 +230,7 @@ def update_search_memories_to_redis(
                 memories: list[TextualMemoryItem] = self.search_memories(
                     search_req=APISearchRequest(**content_dict["search_req"]),
                     user_context=UserContext(**content_dict["user_context"]),
-                    mem_cube=self.current_mem_cube,
+                    mem_cube=self.mem_cube,
                     mode=SearchMode.FAST,
                 )
                 formatted_memories = [format_textual_memory_item(data) for data in memories]

src/memos/mem_scheduler/schemas/task_schemas.py

@@ -60,6 +60,16 @@ class TaskPriorityLevel(Enum):
 # Interval in seconds for batching and cleaning up deletions (xdel)
 DEFAULT_DELETE_CLEANUP_INTERVAL_SEC = 30.0
 
+# Inactivity threshold for stream deletion
+# Delete streams whose last message ID timestamp is older than this threshold.
+# Unit: seconds. Default: 1 day.
+DEFAULT_STREAM_INACTIVITY_DELETE_SECONDS = 86_400.0
+
+# Recency threshold for active streams
+# Consider a stream "active" if its last message is within this window.
+# Unit: seconds. Default: 30 minutes.
+DEFAULT_STREAM_RECENT_ACTIVE_SECONDS = 1_800.0
+
 
 # task queue
 DEFAULT_STREAM_KEY_PREFIX = os.getenv(