MemTensor
diff --git a/‎examples/mem_scheduler/show_redis_status.py‎
Lines changed: 67 additions & 0 deletions b/‎examples/mem_scheduler/show_redis_status.py‎
Lines changed: 67 additions & 0 deletions
diff --git a/‎src/memos/mem_reader/simple_struct.py‎
Lines changed: 30 additions & 42 deletions b/‎src/memos/mem_reader/simple_struct.py‎
Lines changed: 30 additions & 42 deletions
diff --git a/‎src/memos/mem_scheduler/general_scheduler.py‎
Lines changed: 5 additions & 6 deletions b/‎src/memos/mem_scheduler/general_scheduler.py‎
Lines changed: 5 additions & 6 deletions
diff --git a/‎src/memos/mem_scheduler/memory_manage_modules/retriever.py‎
Lines changed: 64 additions & 13 deletions b/‎src/memos/mem_scheduler/memory_manage_modules/retriever.py‎
Lines changed: 64 additions & 13 deletions
diff --git a/‎src/memos/mem_scheduler/schemas/task_schemas.py‎
Lines changed: 13 additions & 10 deletions b/‎src/memos/mem_scheduler/schemas/task_schemas.py‎
Lines changed: 13 additions & 10 deletions
@@ -0,0 +1,67 @@
+import time
+
+from memos.api.routers.server_router import mem_scheduler
+from memos.mem_scheduler.task_schedule_modules.redis_queue import SchedulerRedisQueue
+
+
+queue = mem_scheduler.memos_message_queue.memos_message_queue
+
+
+def fetch_status(queue: SchedulerRedisQueue) -> dict[str, dict[str, int]]:
+    """Fetch and print per-user Redis queue status using built-in API.
+
+    Returns a dict mapping user_id -> {"pending": int, "remaining": int}.
+    """
+    # This method will also print a summary and per-user counts.
+    return queue.show_task_status()
+
+
+def print_diff(prev: dict[str, dict[str, int]], curr: dict[str, dict[str, int]]) -> None:
+    """Print aggregated totals and per-user changes compared to previous snapshot."""
+    ts = time.strftime("%Y-%m-%d %H:%M:%S")
+    tot_p_prev = sum(v.get("pending", 0) for v in prev.values()) if prev else 0
+    tot_r_prev = sum(v.get("remaining", 0) for v in prev.values()) if prev else 0
+    tot_p_curr = sum(v.get("pending", 0) for v in curr.values())
+    tot_r_curr = sum(v.get("remaining", 0) for v in curr.values())
+
+    dp_tot = tot_p_curr - tot_p_prev
+    dr_tot = tot_r_curr - tot_r_prev
+
+    print(f"[{ts}] Total pending={tot_p_curr} ({dp_tot:+d}), remaining={tot_r_curr} ({dr_tot:+d})")
+
+    # Print per-user deltas (current counts are already printed by show_task_status)
+    all_uids = sorted(set(prev.keys()) | set(curr.keys()))
+    for uid in all_uids:
+        p_prev = prev.get(uid, {}).get("pending", 0)
+        r_prev = prev.get(uid, {}).get("remaining", 0)
+        p_curr = curr.get(uid, {}).get("pending", 0)
+        r_curr = curr.get(uid, {}).get("remaining", 0)
+        dp = p_curr - p_prev
+        dr = r_curr - r_prev
+        # Only print when there is any change to reduce noise
+        if dp != 0 or dr != 0:
+            print(f"  Δ {uid}: pending={dp:+d}, remaining={dr:+d}")
+
+
+# Note: queue.show_task_status() handles printing per-user counts internally.
+
+
+def main(interval_sec: float = 5.0) -> None:
+    prev: dict[str, dict[str, int]] = {}
+    while True:
+        try:
+            curr = fetch_status(queue)
+            print_diff(prev, curr)
+            print(f"stream_cache ({len(queue._stream_keys_cache)}): {queue._stream_keys_cache}")
+            prev = curr
+            time.sleep(interval_sec)
+        except KeyboardInterrupt:
+            print("Stopped.")
+            break
+        except Exception as e:
+            print(f"Error while fetching status: {e}")
+            time.sleep(interval_sec)
+
+
+if __name__ == "__main__":
+    main()
@@ -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": {"delete": bool, "rewritten": str, "reason": str}, ... }
+        Expected shape: { "0": {"need_rewrite": bool, "rewritten": str, "reason": str}, ... }
         Returns (success, parsed_dict) with int keys.
         """
         try:
@@ -476,27 +476,33 @@ def _parse_hallucination_filter_response(text: str) -> tuple[bool, dict[int, dic
                     continue
             if not isinstance(v, dict):
                 continue
-            delete_flag = v.get("delete")
+            need_rewrite = v.get("need_rewrite")
             rewritten = v.get("rewritten", "")
             reason = v.get("reason", "")
             if (
-                isinstance(delete_flag, bool)
+                isinstance(need_rewrite, bool)
                 and isinstance(rewritten, str)
                 and isinstance(reason, str)
             ):
-                result[idx] = {"delete": delete_flag, "rewritten": rewritten, "reason": reason}
+                result[idx] = {
+                    "need_rewrite": need_rewrite,
+                    "rewritten": rewritten,
+                    "reason": reason,
+                }
 
         return (len(result) > 0), result
 
     def filter_hallucination_in_memories(
-        self, user_messages: list[str], memory_list: list[TextualMemoryItem]
+        self, messages: list[dict], memory_list: list[TextualMemoryItem]
     ) -> list[TextualMemoryItem]:
-        flat_memories = [one.memory for one in memory_list]
+        # Build input objects with memory text and metadata (timestamps, sources, etc.)
         template = PROMPT_MAPPING["hallucination_filter"]
         prompt_args = {
-            "user_messages_inline": "\n".join([f"- {memory}" for memory in user_messages]),
+            "messages_inline": "\n".join(
+                [f"- [{message['role']}]: {message['content']}" for message in messages]
+            ),
             "memories_inline": json.dumps(
-                {str(i): memory for i, memory in enumerate(flat_memories)},
+                {idx: mem.memory for idx, mem in enumerate(memory_list)},
                 ensure_ascii=False,
                 indent=2,
             ),
@@ -511,40 +517,25 @@ def filter_hallucination_in_memories(
                 f"[filter_hallucination_in_memories] Hallucination filter parsed successfully: {success}"
             )
             if success:
+                new_mem_list = []
                 logger.info(f"Hallucination filter result: {parsed}")
-                total = len(memory_list)
-                keep_flags = [True] * total
+                assert len(parsed) == len(memory_list)
                 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(
-                            f"[filter_hallucination_in_memories] Ignoring out-of-range index: {mem_idx}"
-                        )
-                        continue
-
-                    delete_flag = content.get("delete", False)
-                    rewritten = content.get("rewritten", None)
+                    need_rewrite = content.get("need_rewrite", False)
+                    rewritten = content.get("rewritten", "")
                     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]]
+                    # Apply rewriting if requested
+                    if (
+                        need_rewrite
+                        and isinstance(rewritten, str)
+                        and len(rewritten) > len(memory_list[mem_idx].memory)
+                    ):
+                        memory_list[mem_idx].memory = rewritten
+                        logger.info(
+                            f"[filter_hallucination_in_memories] index={mem_idx}, need_rewrite={need_rewrite}, rewritten='{rewritten}', reason='{reason}', original memory='{memory_list[mem_idx].memory}'"
+                        )
+                    new_mem_list.append(memory_list[mem_idx])
                 return new_mem_list
             else:
                 logger.warning("Hallucination filter parsing failed or returned empty result.")
@@ -602,11 +593,8 @@ def _read_memory(
             # Build inputs
             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
+                    messages=unit_messages, memory_list=unit_memory_list
                 )
                 new_memory_list.append(unit_memory_list)
             memory_list = new_memory_list
 
@@ -144,7 +144,7 @@ def long_memory_update_process(
         old_memory_texts = [mem.memory for mem in cur_working_memory]
         new_memory_texts = [mem.memory for mem in new_order_working_memory]
 
-        logger.debug(
+        logger.info(
             f"[long_memory_update_process] For user_id='{user_id}', mem_cube_id='{mem_cube_id}': "
             f"Scheduler replaced working memory based on query history {queries}. "
             f"Old working memory ({len(old_memory_texts)} items): {old_memory_texts}. "
@@ -1413,17 +1413,16 @@ def process_session_turn(
             logger.info(
                 f"[process_session_turn] Searching for missing evidence: '{item}' with top_k={k_per_evidence} for user_id={user_id}"
             )
-            info = {
-                "user_id": user_id,
-                "session_id": "",
-            }
 
+            search_args = {}
             results: list[TextualMemoryItem] = self.retriever.search(
                 query=item,
+                user_id=user_id,
+                mem_cube_id=mem_cube_id,
                 mem_cube=mem_cube,
                 top_k=k_per_evidence,
                 method=self.search_method,
-                info=info,
+                search_args=search_args,
             )
             logger.info(
                 f"[process_session_turn] Search results for missing evidence '{item}': {[one.memory for one in results]}"
 
@@ -22,7 +22,11 @@
 from memos.mem_scheduler.utils.misc_utils import extract_json_obj, extract_list_items_in_answer
 from memos.memories.textual.item import TextualMemoryMetadata
 from memos.memories.textual.tree import TextualMemoryItem, TreeTextMemory
-from memos.types.general_types import FINE_STRATEGY, FineStrategy
+from memos.types.general_types import (
+    FINE_STRATEGY,
+    FineStrategy,
+    SearchMode,
+)
 
 # Extract JSON response
 from .memory_filter import MemoryFilter
@@ -237,10 +241,12 @@ def recall_for_missing_memories(
     def search(
         self,
         query: str,
+        user_id: str,
+        mem_cube_id: str,
         mem_cube: GeneralMemCube,
         top_k: int,
         method: str = TreeTextMemory_SEARCH_METHOD,
-        info: dict | None = None,
+        search_args: dict | None = None,
     ) -> list[TextualMemoryItem]:
         """Search in text memory with the given query.
 
@@ -253,22 +259,67 @@ def search(
             Search results or None if not implemented
         """
         text_mem_base = mem_cube.text_mem
+        # Normalize default for mutable argument
+        search_args = search_args or {}
         try:
             if method in [TreeTextMemory_SEARCH_METHOD, TreeTextMemory_FINE_SEARCH_METHOD]:
                 assert isinstance(text_mem_base, TreeTextMemory)
-                if info is None:
-                    logger.warning(
-                        "Please input 'info' when use tree.search so that "
-                        "the database would store the consume history."
-                    )
-                    info = {"user_id": "", "session_id": ""}
+                session_id = search_args.get("session_id", "default_session")
+                target_session_id = session_id
+                search_priority = (
+                    {"session_id": target_session_id} if "session_id" in search_args else None
+                )
+                search_filter = search_args.get("filter")
+                search_source = search_args.get("source")
+                plugin = bool(search_source is not None and search_source == "plugin")
+                user_name = search_args.get("user_name", mem_cube_id)
+                internet_search = search_args.get("internet_search", False)
+                chat_history = search_args.get("chat_history")
+                search_tool_memory = search_args.get("search_tool_memory", False)
+                tool_mem_top_k = search_args.get("tool_mem_top_k", 6)
+                playground_search_goal_parser = search_args.get(
+                    "playground_search_goal_parser", False
+                )
 
-                mode = "fast" if method == TreeTextMemory_SEARCH_METHOD else "fine"
-                results_long_term = text_mem_base.search(
-                    query=query, top_k=top_k, memory_type="LongTermMemory", mode=mode, info=info
+                info = search_args.get(
+                    "info",
+                    {
+                        "user_id": user_id,
+                        "session_id": target_session_id,
+                        "chat_history": chat_history,
+                    },
                 )
-                results_user = text_mem_base.search(
-                    query=query, top_k=top_k, memory_type="UserMemory", mode=mode, info=info
+
+                results_long_term = mem_cube.text_mem.search(
+                    query=query,
+                    user_name=user_name,
+                    top_k=top_k,
+                    mode=SearchMode.FAST,
+                    manual_close_internet=not internet_search,
+                    memory_type="LongTermMemory",
+                    search_filter=search_filter,
+                    search_priority=search_priority,
+                    info=info,
+                    plugin=plugin,
+                    search_tool_memory=search_tool_memory,
+                    tool_mem_top_k=tool_mem_top_k,
+                    playground_search_goal_parser=playground_search_goal_parser,
+                )
+
+                results_user = mem_cube.text_mem.search(
+                    query=query,
+                    user_name=user_name,
+                    top_k=top_k,
+                    mode=SearchMode.FAST,
+                    manual_close_internet=not internet_search,
+                    memory_type="UserMemory",
+                    search_filter=search_filter,
+                    search_priority=search_priority,
+                    info=info,
+                    plugin=plugin,
+                    search_tool_memory=search_tool_memory,
+                    tool_mem_top_k=tool_mem_top_k,
+                    playground_search_goal_parser=playground_search_goal_parser,
                 )
                 results = results_long_term + results_user
             else:
 
@@ -45,10 +45,6 @@ class TaskPriorityLevel(Enum):
 USER_INPUT_TYPE = "UserInput"
 NOT_APPLICABLE_TYPE = "NotApplicable"
 
-# pending claim configuration
-# Only claim pending messages whose idle time exceeds this threshold.
-# Unit: milliseconds. Default: 10 minute.
-DEFAULT_PENDING_CLAIM_MIN_IDLE_MS = 600_000
 
 # scheduler daemon defaults
 # Interval in seconds for periodically releasing stale pending messages
@@ -60,15 +56,22 @@ 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
+# pending claim configuration
+# Only claim pending messages whose idle time exceeds this threshold.
+# Unit: milliseconds. Default: 1 hour.
+DEFAULT_PENDING_CLAIM_MIN_IDLE_MS = 3_600_000
+
 
 # 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
+# Unit: seconds. Default: 1 hours.
+DEFAULT_STREAM_RECENT_ACTIVE_SECONDS = 3_600.0
+
+
+# Inactivity threshold for stream deletion
+# Delete streams whose last message ID timestamp is older than this threshold.
+# Unit: seconds. Default: 2 hour.
+DEFAULT_STREAM_INACTIVITY_DELETE_SECONDS = 7_200.0
 
 
 # task queue