tianxing02
diff --git a/‎src/memos/mem_scheduler/base_scheduler.py‎
Lines changed: 41 additions & 1 deletion b/‎src/memos/mem_scheduler/base_scheduler.py‎
Lines changed: 41 additions & 1 deletion
diff --git a/‎src/memos/mem_scheduler/schemas/general_schemas.py‎
Lines changed: 6 additions & 1 deletion b/‎src/memos/mem_scheduler/schemas/general_schemas.py‎
Lines changed: 6 additions & 1 deletion
diff --git a/‎src/memos/mem_scheduler/schemas/message_schemas.py‎
Lines changed: 4 additions & 0 deletions b/‎src/memos/mem_scheduler/schemas/message_schemas.py‎
Lines changed: 4 additions & 0 deletions
diff --git a/‎src/memos/mem_scheduler/task_schedule_modules/dispatcher.py‎
Lines changed: 71 additions & 5 deletions b/‎src/memos/mem_scheduler/task_schedule_modules/dispatcher.py‎
Lines changed: 71 additions & 5 deletions
@@ -4,7 +4,7 @@
 import time
 
 from collections.abc import Callable
-from datetime import datetime
+from datetime import datetime, timezone
 from pathlib import Path
 from typing import TYPE_CHECKING, Union
 
@@ -49,6 +49,7 @@
 from memos.mem_scheduler.utils.filter_utils import (
     transform_name_to_key,
 )
+from memos.mem_scheduler.utils.monitor_event_utils import emit_monitor_event, to_iso
 from memos.mem_scheduler.utils.status_tracker import TaskStatusTracker
 from memos.mem_scheduler.webservice_modules.rabbitmq_service import RabbitMQSchedulerModule
 from memos.mem_scheduler.webservice_modules.redis_service import RedisSchedulerModule
@@ -175,6 +176,8 @@ def init_mem_cube(
         searcher: Searcher | None = None,
         feedback_server: Searcher | None = None,
     ):
+        if mem_cube is None:
+            logger.error("mem_cube is None, cannot initialize", stack_info=True)
         self.mem_cube = mem_cube
         self.text_mem: TreeTextMemory = self.mem_cube.text_mem
         self.reranker: HTTPBGEReranker = self.text_mem.reranker
@@ -258,6 +261,15 @@ def _cleanup_on_init_failure(self):
     @property
     def mem_cube(self) -> BaseMemCube:
         """The memory cube associated with this MemChat."""
+        if self.current_mem_cube is None:
+            logger.error("mem_cube is None when accessed", stack_info=True)
+            try:
+                self.components = init_components()
+                self.current_mem_cube: BaseMemCube = self.components["naive_mem_cube"]
+            except Exception:
+                logger.info(
+                    "No environment available to initialize mem cube. Using fallback naive_mem_cube."
+                )
         return self.current_mem_cube
 
     @mem_cube.setter
@@ -757,7 +769,35 @@ def _message_consumer(self) -> None:
                 messages = self.memos_message_queue.get_messages(batch_size=self.consume_batch)
 
                 if messages:
+                    now = time.time()
                     for msg in messages:
+                        enqueue_ts_obj = getattr(msg, "timestamp", None)
+                        enqueue_epoch = None
+                        if isinstance(enqueue_ts_obj, int | float):
+                            enqueue_epoch = float(enqueue_ts_obj)
+                        elif hasattr(enqueue_ts_obj, "timestamp"):
+                            dt = enqueue_ts_obj
+                            if dt.tzinfo is None:
+                                dt = dt.replace(tzinfo=timezone.utc)
+                            enqueue_epoch = dt.timestamp()
+
+                        queue_wait_ms = None
+                        if enqueue_epoch is not None:
+                            queue_wait_ms = max(0.0, now - enqueue_epoch) * 1000
+
+                        msg.dequeue_ts = now
+                        emit_monitor_event(
+                            "dequeue",
+                            msg,
+                            {
+                                "enqueue_ts": to_iso(enqueue_ts_obj),
+                                "dequeue_ts": datetime.fromtimestamp(
+                                    now, tz=timezone.utc
+                                ).isoformat(),
+                                "queue_wait_ms": queue_wait_ms,
+                            },
+                        )
+
                         self.metrics.task_dequeued(user_id=msg.user_id, task_type=msg.label)
                     try:
                         import contextlib
 
@@ -66,7 +66,12 @@
 DEFAULT_MAX_WEB_LOG_QUEUE_SIZE = 50
 
 # task queue
-DEFAULT_STREAM_KEY_PREFIX = "scheduler:messages:stream:v1.4"
+DEFAULT_STREAM_KEY_PREFIX = "scheduler:messages:stream:v1.5"
 exchange_name = os.getenv("MEMSCHEDULER_RABBITMQ_EXCHANGE_NAME", None)
 if exchange_name is not None:
     DEFAULT_STREAM_KEY_PREFIX += f":{exchange_name}"
+
+# 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
@@ -5,6 +5,7 @@
 from pydantic import BaseModel, ConfigDict, Field
 from typing_extensions import TypedDict
 
+from memos.context.context import generate_trace_id
 from memos.log import get_logger
 from memos.mem_scheduler.general_modules.misc import DictConversionMixin
 from memos.mem_scheduler.utils.db_utils import get_utc_now
@@ -36,6 +37,7 @@ class ScheduleMessageItem(BaseModel, DictConversionMixin):
     redis_message_id: str = Field(default="", description="the message get from redis stream")
     stream_key: str = Field("", description="stream_key for identifying the queue in line")
     user_id: str = Field(..., description="user id")
+    trace_id: str = Field(default_factory=generate_trace_id, description="trace id for logging")
     mem_cube_id: str = Field(..., description="memcube id")
     session_id: str = Field(default="", description="Session ID for soft-filtering memories")
     label: str = Field(..., description="Label of the schedule message")
@@ -80,6 +82,7 @@ def to_dict(self) -> dict:
             "item_id": self.item_id,
             "user_id": self.user_id,
             "cube_id": self.mem_cube_id,
+            "trace_id": self.trace_id,
             "label": self.label,
             "cube": "Not Applicable",  # Custom cube serialization
             "content": self.content,
@@ -95,6 +98,7 @@ def from_dict(cls, data: dict) -> "ScheduleMessageItem":
             item_id=data.get("item_id", str(uuid4())),
             user_id=data["user_id"],
             mem_cube_id=data["cube_id"],
+            trace_id=data.get("trace_id", generate_trace_id()),
             label=data["label"],
             content=data["content"],
             timestamp=datetime.fromisoformat(data["timestamp"]),
 
@@ -4,10 +4,15 @@
 
 from collections import defaultdict
 from collections.abc import Callable
-from datetime import timezone
+from datetime import datetime, timezone
 from typing import Any
 
-from memos.context.context import ContextThreadPoolExecutor
+from memos.context.context import (
+    ContextThreadPoolExecutor,
+    RequestContext,
+    generate_trace_id,
+    set_request_context,
+)
 from memos.log import get_logger
 from memos.mem_scheduler.general_modules.base import BaseSchedulerModule
 from memos.mem_scheduler.general_modules.task_threads import ThreadManager
@@ -19,6 +24,7 @@
 from memos.mem_scheduler.task_schedule_modules.redis_queue import SchedulerRedisQueue
 from memos.mem_scheduler.task_schedule_modules.task_queue import ScheduleTaskQueue
 from memos.mem_scheduler.utils.misc_utils import group_messages_by_user_and_mem_cube
+from memos.mem_scheduler.utils.monitor_event_utils import emit_monitor_event, to_iso
 from memos.mem_scheduler.utils.status_tracker import TaskStatusTracker
 
 
@@ -121,15 +127,26 @@ def _create_task_wrapper(self, handler: Callable, task_item: RunningTaskItem):
 
         def wrapped_handler(messages: list[ScheduleMessageItem]):
             start_time = time.time()
+            start_iso = datetime.fromtimestamp(start_time, tz=timezone.utc).isoformat()
             if self.status_tracker:
                 self.status_tracker.task_started(
                     task_id=task_item.item_id, user_id=task_item.user_id
                 )
             try:
+                first_msg = messages[0]
+                trace_id = getattr(first_msg, "trace_id", None) or generate_trace_id()
+                # Propagate trace_id and user info to logging context for this handler execution
+                ctx = RequestContext(
+                    trace_id=trace_id,
+                    user_name=getattr(first_msg, "user_name", None),
+                    user_type=None,
+                )
+                set_request_context(ctx)
+
                 # --- mark start: record queuing time(now - enqueue_ts)---
                 now = time.time()
-                m = messages[0]  # All messages in this batch have same user and type
-                enq_ts = getattr(m, "timestamp", None)
+                m = first_msg  # All messages in this batch have same user and type
+                enq_ts = getattr(first_msg, "timestamp", None)
 
                 # Path 1: epoch seconds (preferred)
                 if isinstance(enq_ts, int | float):
@@ -149,17 +166,51 @@ def wrapped_handler(messages: list[ScheduleMessageItem]):
                 wait_sec = max(0.0, now - enq_epoch)
                 self.metrics.observe_task_wait_duration(wait_sec, m.user_id, m.label)
 
+                dequeue_ts = getattr(first_msg, "dequeue_ts", None)
+                start_delay_ms = None
+                if isinstance(dequeue_ts, int | float):
+                    start_delay_ms = max(0.0, start_time - dequeue_ts) * 1000
+
+                emit_monitor_event(
+                    "start",
+                    first_msg,
+                    {
+                        "start_ts": start_iso,
+                        "start_delay_ms": start_delay_ms,
+                        "enqueue_ts": to_iso(enq_ts),
+                        "dequeue_ts": to_iso(
+                            datetime.fromtimestamp(dequeue_ts, tz=timezone.utc)
+                            if isinstance(dequeue_ts, int | float)
+                            else None
+                        ),
+                    },
+                )
+
                 # Execute the original handler
                 result = handler(messages)
 
                 # --- mark done ---
-                duration = time.time() - start_time
+                finish_time = time.time()
+                duration = finish_time - start_time
                 self.metrics.observe_task_duration(duration, m.user_id, m.label)
                 if self.status_tracker:
                     self.status_tracker.task_completed(
                         task_id=task_item.item_id, user_id=task_item.user_id
                     )
                 self.metrics.task_completed(user_id=m.user_id, task_type=m.label)
+
+                emit_monitor_event(
+                    "finish",
+                    first_msg,
+                    {
+                        "status": "ok",
+                        "start_ts": start_iso,
+                        "finish_ts": datetime.fromtimestamp(
+                            finish_time, tz=timezone.utc
+                        ).isoformat(),
+                        "exec_duration_ms": duration * 1000,
+                    },
+                )
                 # Redis ack is handled in finally to cover failure cases
 
                 # Mark task as completed and remove from tracking
@@ -172,11 +223,26 @@ def wrapped_handler(messages: list[ScheduleMessageItem]):
 
             except Exception as e:
                 m = messages[0]
+                finish_time = time.time()
                 self.metrics.task_failed(m.user_id, m.label, type(e).__name__)
                 if self.status_tracker:
                     self.status_tracker.task_failed(
                         task_id=task_item.item_id, user_id=task_item.user_id, error_message=str(e)
                     )
+                emit_monitor_event(
+                    "finish",
+                    m,
+                    {
+                        "status": "fail",
+                        "start_ts": start_iso,
+                        "finish_ts": datetime.fromtimestamp(
+                            finish_time, tz=timezone.utc
+                        ).isoformat(),
+                        "exec_duration_ms": (finish_time - start_time) * 1000,
+                        "error_type": type(e).__name__,
+                        "error_msg": str(e),
+                    },
+                )
                 # Mark task as failed and remove from tracking
                 with self._task_lock:
                     if task_item.item_id in self._running_tasks: