Feat: reorg playground MemCube logs\n\n* feat: normalize show_web_logs title/content (HH:mm:ss, truncate, zh types, desc sort)\n* feat: add addMessage web logs (query/answer)\n* feat: emit merge logs (merged keys + merged result)\n* feat: add mem_update/mem_archive labels and logger methods\n* feat: add non-breaking add/update log split by metadata.key\n* chore: keep schema unchanged; no frontend coupling

Author: [email protected]

examples/mem_scheduler/memos_w_scheduler.py

@@ -11,6 +11,18 @@
 from memos.log import get_logger
 from memos.mem_cube.general import GeneralMemCube
 from memos.mem_os.main import MOS
+from datetime import datetime
+import re
+from memos.mem_scheduler.schemas.general_schemas import (
+    QUERY_LABEL,
+    ANSWER_LABEL,
+    ADD_LABEL,
+    MEM_ORGANIZE_LABEL,
+    MEM_UPDATE_LABEL,
+    MEM_ARCHIVE_LABEL,
+    NOT_APPLICABLE_TYPE,
+)
+from memos.mem_scheduler.utils.filter_utils import transform_name_to_key
 from memos.mem_scheduler.general_scheduler import GeneralScheduler
 
 
@@ -70,6 +82,89 @@ def init_task():
     return conversations, questions
 
 
+def _truncate_with_rules(text: str) -> str:
+    has_cjk = bool(re.search(r"[\u4e00-\u9fff]", text))
+    limit = 32 if has_cjk else 64
+    s = text.strip().replace("\n", " ")
+    if len(s) <= limit:
+        return s
+    return s[: limit - 1] + "… 查看详情"
+
+
+def _format_title(ts: datetime, title_text: str) -> str:
+    return f"{ts.strftime('%H:%M:%S')} {title_text}"
+
+
+def _cube_display_from(mem_cube_id: str) -> str:
+    if "public" in (mem_cube_id or "").lower():
+        return "公共MemCube"
+    return "用户MemCube"
+
+_TYPE_ZH = {
+    "LongTermMemory": "长期",
+    "UserMemory": "用户",
+    "WorkingMemory": "工作",
+    "ActivationMemory": "激活",
+    "ParameterMemory": "参数",
+    "TextMemory": "明文",
+    "UserInput": "消息",
+    "NotApplicable": "NA",
+}
+
+
+def _format_entry(item) -> tuple[str, str]:
+    local_ts = item.timestamp.astimezone()
+    cube_display = _cube_display_from(item.mem_cube_id)
+    label = item.label
+    from_t = item.from_memory_type
+    to_t = item.to_memory_type
+    content = item.log_content or ""
+
+    if label == QUERY_LABEL:
+        xx = cube_display.replace("MemCube", "")
+        title = _format_title(local_ts, f"addMessages至{xx} MemCube")
+        return title, _truncate_with_rules(content)
+    if label == ANSWER_LABEL:
+        xx = cube_display.replace("MemCube", "")
+        title = _format_title(local_ts, f"addMessages至{xx} MemCube")
+        return title, _truncate_with_rules(content)
+
+    if label == ADD_LABEL:
+        key = transform_name_to_key(content)
+        title = _format_title(local_ts, f"{cube_display}新增了1条记忆")
+        return title, _truncate_with_rules(f"{key}：{content}")
+
+    if label == MEM_UPDATE_LABEL:
+        key = transform_name_to_key(content)
+        title = _format_title(local_ts, f"{cube_display}更新了1条记忆")
+        return title, _truncate_with_rules(f"{key}：{content}")
+
+    if label == MEM_ARCHIVE_LABEL:
+        key = transform_name_to_key(content)
+        title = _format_title(local_ts, f"{cube_display}归档了1条记忆")
+        return title, _truncate_with_rules(f"{key}：{content}")
+
+    if label == MEM_ORGANIZE_LABEL:
+        m = re.search(r"被合并的记忆：(.+)", content)
+        if m:
+            keys_str = m.group(1)
+            n = len([x for x in keys_str.split("|") if x.strip()])
+            title = _format_title(local_ts, f"{cube_display}合并了{n}条记忆")
+            return title, _truncate_with_rules(content)
+        title = _format_title(local_ts, f"{cube_display}合并了1条记忆")
+        return title, _truncate_with_rules(content)
+
+    if from_t != to_t and to_t != NOT_APPLICABLE_TYPE:
+        key = transform_name_to_key(content)
+        title = _format_title(local_ts, f"{cube_display}调度了1条记忆")
+        from_zh = _TYPE_ZH.get(from_t, from_t)
+        to_zh = _TYPE_ZH.get(to_t, to_t)
+        return title, _truncate_with_rules(f"[{from_zh} → {to_zh}] {key}：{content}")
+
+    title = _format_title(local_ts, f"{cube_display}事件")
+    return title, _truncate_with_rules(content)
+
+
 def show_web_logs(mem_scheduler: GeneralScheduler):
     """Display all web log entries from the scheduler's log queue.
 
@@ -86,16 +181,20 @@ def show_web_logs(mem_scheduler: GeneralScheduler):
     temp_queue = Queue()
     log_count = 0
 
+    # drain all items for sorting by time desc
+    items = []
     while not mem_scheduler._web_log_message_queue.empty():
         log_item: ScheduleLogForWebItem = mem_scheduler._web_log_message_queue.get()
+        items.append(log_item)
         temp_queue.put(log_item)
-        log_count += 1
-
-        # Print log entry details
-        print(f"\nLog Entry #{log_count}:")
-        print(f'- "{log_item.label}" log: {log_item}')
-
-        print("-" * 50)
+    # sort by timestamp desc
+    items_sorted = sorted(items, key=lambda x: x.timestamp, reverse=True)
+    for idx, log_item in enumerate(items_sorted, 1):
+        title, content = _format_entry(log_item)
+        print(f"\nLog Entry #{idx}:")
+        print(title)
+        print(content)
+    log_count = len(items_sorted)
 
     # Restore items back to the original queue
     while not temp_queue.empty():

src/memos/mem_scheduler/general_modules/scheduler_logger.py

@@ -13,6 +13,8 @@
     TEXT_MEMORY_TYPE,
     USER_INPUT_TYPE,
     WORKING_MEMORY_TYPE,
+    MEM_UPDATE_LABEL,
+    MEM_ARCHIVE_LABEL,
 )
 from memos.mem_scheduler.schemas.message_schemas import (
     ScheduleLogForWebItem,
@@ -233,6 +235,48 @@ def log_adding_memory(
             f"converted to {memory_type} memory in mem_cube {mem_cube_id}: {memory}"
         )
 
+    @log_exceptions(logger=logger)
+    def log_updating_memory(
+        self,
+        memory: str,
+        memory_type: str,
+        user_id: str,
+        mem_cube_id: str,
+        mem_cube: GeneralMemCube,
+        log_func_callback: Callable[[list[ScheduleLogForWebItem]], None],
+    ):
+        log_message = self.create_autofilled_log_item(
+            log_content=memory,
+            label=MEM_UPDATE_LABEL,
+            from_memory_type=memory_type,
+            to_memory_type=memory_type,
+            user_id=user_id,
+            mem_cube_id=mem_cube_id,
+            mem_cube=mem_cube,
+        )
+        log_func_callback([log_message])
+
+    @log_exceptions(logger=logger)
+    def log_archiving_memory(
+        self,
+        memory: str,
+        memory_type: str,
+        user_id: str,
+        mem_cube_id: str,
+        mem_cube: GeneralMemCube,
+        log_func_callback: Callable[[list[ScheduleLogForWebItem]], None],
+    ):
+        log_message = self.create_autofilled_log_item(
+            log_content=memory,
+            label=MEM_ARCHIVE_LABEL,
+            from_memory_type=memory_type,
+            to_memory_type=memory_type,
+            user_id=user_id,
+            mem_cube_id=mem_cube_id,
+            mem_cube=mem_cube,
+        )
+        log_func_callback([log_message])
+
     @log_exceptions(logger=logger)
     def validate_schedule_message(self, message: ScheduleMessageItem, label: str):
         """Validate if the message matches the expected label.

src/memos/mem_scheduler/general_scheduler.py

@@ -16,6 +16,9 @@
     PREF_ADD_LABEL,
     QUERY_LABEL,
     WORKING_MEMORY_TYPE,
+    USER_INPUT_TYPE,
+    NOT_APPLICABLE_TYPE,
+    LONG_TERM_MEMORY_TYPE,
     MemCubeID,
     UserID,
 )
@@ -159,6 +162,24 @@ def _query_message_consumer(self, messages: list[ScheduleMessageItem]) -> None:
                 messages = grouped_messages[user_id][mem_cube_id]
                 if len(messages) == 0:
                     return
+                try:
+                    web_logs = []
+                    for msg in messages:
+                        web_logs.append(
+                            self.create_autofilled_log_item(
+                                log_content=f"[User] {msg.content}",
+                                label=QUERY_LABEL,
+                                from_memory_type=USER_INPUT_TYPE,
+                                to_memory_type=NOT_APPLICABLE_TYPE,
+                                user_id=msg.user_id,
+                                mem_cube_id=msg.mem_cube_id,
+                                mem_cube=self.current_mem_cube,
+                            )
+                        )
+                    if web_logs:
+                        self._submit_web_logs(web_logs)
+                except Exception:
+                    pass
                 self.long_memory_update_process(
                     user_id=user_id, mem_cube_id=mem_cube_id, messages=messages
                 )
@@ -181,6 +202,24 @@ def _answer_message_consumer(self, messages: list[ScheduleMessageItem]) -> None:
                 messages = grouped_messages[user_id][mem_cube_id]
                 if len(messages) == 0:
                     return
+                try:
+                    web_logs = []
+                    for msg in messages:
+                        web_logs.append(
+                            self.create_autofilled_log_item(
+                                log_content=f"[Assistant] {msg.content}",
+                                label=ANSWER_LABEL,
+                                from_memory_type=USER_INPUT_TYPE,
+                                to_memory_type=NOT_APPLICABLE_TYPE,
+                                user_id=msg.user_id,
+                                mem_cube_id=msg.mem_cube_id,
+                                mem_cube=self.current_mem_cube,
+                            )
+                        )
+                    if web_logs:
+                        self._submit_web_logs(web_logs)
+                except Exception:
+                    pass
 
     def _add_message_consumer(self, messages: list[ScheduleMessageItem]) -> None:
         logger.info(f"Messages {messages} assigned to {ADD_LABEL} handler.")
@@ -216,18 +255,57 @@ def _add_message_consumer(self, messages: list[ScheduleMessageItem]) -> None:
                                 continue
                             mem_type = mem_item.metadata.memory_type
                             mem_content = mem_item.memory
+                            key = getattr(mem_item.metadata, "key", None)
 
                             if mem_type == WORKING_MEMORY_TYPE:
                                 continue
 
-                            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=self.current_mem_cube,
-                                log_func_callback=self._submit_web_logs,
-                            )
+                            # 判断新增 vs 更新：同类型内存在同 key 视为更新，否则视为新增
+                            try:
+                                text_mem = mem_cube.text_mem
+                                existing_found = False
+                                if key and hasattr(text_mem, "graph_store"):
+                                    try:
+                                        candidates = text_mem.graph_store.get_by_metadata(
+                                            [
+                                                {"field": "memory", "op": "=", "value": key},
+                                                {"field": "memory_type", "op": "=", "value": mem_type},
+                                            ]
+                                        )
+                                        existing_found = bool(candidates)
+                                    except Exception:
+                                        existing_found = False
+
+                                if existing_found:
+                                    # 更新记忆日志（同类型内更新）
+                                    self.log_updating_memory(
+                                        memory=mem_content,
+                                        memory_type=mem_type,
+                                        user_id=msg.user_id,
+                                        mem_cube_id=msg.mem_cube_id,
+                                        mem_cube=self.current_mem_cube,
+                                        log_func_callback=self._submit_web_logs,
+                                    )
+                                else:
+                                    # 新增记忆日志
+                                    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=self.current_mem_cube,
+                                        log_func_callback=self._submit_web_logs,
+                                    )
+                            except Exception:
+                                # 兜底：仍记录为新增
+                                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=self.current_mem_cube,
+                                    log_func_callback=self._submit_web_logs,
+                                )
 
         except Exception as e:
             logger.error(f"Error: {e}", exc_info=True)
@@ -432,6 +510,46 @@ def process_message(message: ScheduleMessageItem):
                     user_name=user_name,
                 )
 
+                try:
+                    mem_items = []
+                    for mid in mem_ids:
+                        try:
+                            mem_items.append(text_mem.get(mid))
+                        except Exception:
+                            pass
+                    if len(mem_items) > 1:
+                        keys = []
+                        for it in mem_items:
+                            key = getattr(getattr(it, "metadata", {}), "key", None) or ""
+                            if key:
+                                keys.append(key)
+                        merged_info = "被合并的记忆：" + " | ".join([str(k) for k in keys])
+                        log_item_a = self.create_autofilled_log_item(
+                            log_content=merged_info,
+                            label=MEM_ORGANIZE_LABEL,
+                            from_memory_type=LONG_TERM_MEMORY_TYPE,
+                            to_memory_type=LONG_TERM_MEMORY_TYPE,
+                            user_id=user_id,
+                            mem_cube_id=mem_cube_id,
+                            mem_cube=mem_cube,
+                        )
+                        combined_key = keys[0] if keys else ""
+                        merged_after = (
+                            "合并后的记忆单元：" + (combined_key if combined_key else "(无键)")
+                        )
+                        log_item_b = self.create_autofilled_log_item(
+                            log_content=merged_after,
+                            label=MEM_ORGANIZE_LABEL,
+                            from_memory_type=LONG_TERM_MEMORY_TYPE,
+                            to_memory_type=LONG_TERM_MEMORY_TYPE,
+                            user_id=user_id,
+                            mem_cube_id=mem_cube_id,
+                            mem_cube=mem_cube,
+                        )
+                        self._submit_web_logs([log_item_a, log_item_b])
+                except Exception:
+                    pass
+
                 logger.info(
                     f"Successfully processed mem_read for user_id={user_id}, mem_cube_id={mem_cube_id}"
                 )

src/memos/mem_scheduler/schemas/general_schemas.py

@@ -20,6 +20,8 @@ class SearchMode(str, Enum):
 ADD_LABEL = "add"
 MEM_READ_LABEL = "mem_read"
 MEM_ORGANIZE_LABEL = "mem_organize"
+MEM_UPDATE_LABEL = "mem_update"
+MEM_ARCHIVE_LABEL = "mem_archive"
 API_MIX_SEARCH_LABEL = "api_mix_search"
 PREF_ADD_LABEL = "pref_add"