feat(scheduler): implement status-driven failure logging and fix redis_queue status_tracker init

Author: [email protected]

src/memos/mem_scheduler/task_schedule_modules/dispatcher.py

@@ -319,7 +319,10 @@ def _maybe_emit_task_completion(
 
                 status = status_data.get("status")
 
-                if status == "completed" and error is None:
+                if status == "completed":
+                    # Only emit success log if we didn't just catch an exception locally
+                    # (Although if status is 'completed', local error shouldn't happen theoretically,
+                    # unless status update lags or is inconsistent. We trust status_tracker here.)
                     event = ScheduleLogForWebItem(
                         task_id=task_id,
                         user_id=user_id,
@@ -331,15 +334,26 @@ def _maybe_emit_task_completion(
                         status="completed",
                     )
                     self.submit_web_logs(event)
-                elif status == "failed" and error is not None:
+
+                elif status == "failed":
+                    # Construct error message
+                    error_msg = str(error) if error else None
+                    if not error_msg:
+                        # Try to get errors from status_tracker aggregation
+                        errors = status_data.get("errors", [])
+                        if errors:
+                            error_msg = "; ".join(errors)
+                        else:
+                            error_msg = "Unknown error (check system logs)"
+
                     event = ScheduleLogForWebItem(
                         task_id=task_id,
                         user_id=user_id,
                         mem_cube_id=mem_cube_id,
                         label="taskStatus",
                         from_memory_type="status",
                         to_memory_type="status",
-                        log_content=f"Task {task_id} failed: {error!s}",
+                        log_content=f"Task {task_id} failed: {error_msg}",
                         status="failed",
                     )
                     self.submit_web_logs(event)

src/memos/mem_scheduler/task_schedule_modules/redis_queue.py

@@ -21,8 +21,9 @@
     DEFAULT_STREAM_KEY_PREFIX,
     DEFAULT_STREAM_KEYS_REFRESH_INTERVAL_SEC,
 )
-from memos.mem_scheduler.utils.monitor_event_utils import emit_monitor_event, to_iso
+from memos.mem_scheduler.task_schedule_modules.orchestrator import SchedulerOrchestrator
 from memos.mem_scheduler.utils.status_tracker import TaskStatusTracker
+from memos.mem_scheduler.webservice_modules.redis_service import RedisSchedulerModule
 
 
 logger = get_logger(__name__)

src/memos/mem_scheduler/utils/status_tracker.py

@@ -142,11 +142,14 @@ def get_task_status_by_business_id(self, business_task_id: str, user_id: str) ->
         # Get statuses for all items
         key = self._get_key(user_id)
         item_statuses = []
+        errors = []
         for item_id in item_ids:
             item_data_json = self.redis.hget(key, item_id)
             if item_data_json:
                 item_data = json.loads(item_data_json)
                 item_statuses.append(item_data["status"])
+                if item_data.get("status") == "failed" and "error" in item_data:
+                    errors.append(item_data["error"])
 
         if not item_statuses:
             return None
@@ -167,6 +170,7 @@ def get_task_status_by_business_id(self, business_task_id: str, user_id: str) ->
             "business_task_id": business_task_id,
             "item_count": len(item_ids),
             "item_statuses": item_statuses,
+            "errors": errors,
         }
 
     def get_all_tasks_global(self) -> dict[str, dict[str, dict]]: