refactor: remove unnecessary parameters for local queue

Author: tangg555

src/memos/mem_scheduler/task_schedule_modules/local_queue.py

@@ -28,10 +28,7 @@ def __init__(
         maxsize: int = 0,
         stream_key_prefix: str = DEFAULT_STREAM_KEY_PREFIX,
         orchestrator: SchedulerOrchestrator | None = None,
-        consumer_group: str = "scheduler_group",
-        consumer_name: str | None = "scheduler_consumer",
         max_len: int | None = None,
-        auto_delete_acked: bool = True,
         status_tracker: TaskStatusTracker | None = None,
     ):
         """
@@ -42,10 +39,7 @@ def __init__(
             maxsize (int): Maximum number of messages allowed in each individual queue.
             stream_key_prefix (str): Prefix for stream keys (simulated).
             orchestrator: SchedulerOrchestrator instance (ignored).
-            consumer_group: Consumer group name (ignored).
-            consumer_name: Consumer name (ignored).
             max_len: Alias for maxsize if maxsize is 0.
-            auto_delete_acked: Whether to delete acked messages (ignored).
             status_tracker: TaskStatusTracker instance (ignored).
         """
         super().__init__()
@@ -62,9 +56,6 @@ def __init__(
         self.queue_streams: dict[str, Queue[ScheduleMessageItem]] = {}
 
         self.orchestrator = orchestrator
-        self.consumer_group = consumer_group
-        self.consumer_name = consumer_name
-        self.auto_delete_acked = auto_delete_acked
         self.status_tracker = status_tracker
 
         self._is_listening = False
@@ -122,7 +113,7 @@ def get(
         stream_key: str,
         block: bool = True,
         timeout: float | None = None,
-        batch_size: int | None = None,
+        batch_size: int | None = 1,
     ) -> list[ScheduleMessageItem]:
         if batch_size is not None and batch_size <= 0:
             logger.warning(
@@ -135,18 +126,19 @@ def get(
             logger.error(f"Stream {stream_key} does not exist when trying to get messages.")
             return []
 
+        # Ensure we always request a batch so we get a list back
+        effective_batch_size = batch_size if batch_size is not None else 1
+
         # Note: Assumes custom Queue implementation supports batch_size parameter
         res = self.queue_streams[stream_key].get(
-            block=block, timeout=timeout, batch_size=batch_size
+            block=block, timeout=timeout, batch_size=effective_batch_size
         )
         logger.debug(
             f"Retrieved {len(res)} messages from queue '{stream_key}'. Current size: {self.queue_streams[stream_key].qsize()}"
         )
         return res
 
-    def get_nowait(
-        self, stream_key: str, batch_size: int | None = None
-    ) -> list[ScheduleMessageItem]:
+    def get_nowait(self, stream_key: str, batch_size: int | None = 1) -> list[ScheduleMessageItem]:
         """
         Non-blocking version of get(). Equivalent to get(stream_key, block=False, batch_size=batch_size).
 
@@ -195,17 +187,24 @@ def qsize(self) -> dict:
         Return the current size of all internal queues as a dictionary.
 
         Each key is the stream name, and each value is the number of messages in that queue.
+        Also includes 'total_size'.
 
         Returns:
             Dict[str, int]: Mapping from stream name to current queue size.
         """
         sizes = {stream: queue.qsize() for stream, queue in self.queue_streams.items()}
+        total_size = sum(sizes.values())
+        sizes["total_size"] = total_size
         logger.debug(f"Current queue sizes: {sizes}")
         return sizes
 
-    def clear(self) -> None:
-        for queue in self.queue_streams.values():
-            queue.clear()
+    def clear(self, stream_key: str | None = None) -> None:
+        if stream_key:
+            if stream_key in self.queue_streams:
+                self.queue_streams[stream_key].clear()
+        else:
+            for queue in self.queue_streams.values():
+                queue.clear()
 
     @property
     def unfinished_tasks(self) -> int:
@@ -217,7 +216,10 @@ def unfinished_tasks(self) -> int:
         Returns:
             int: Sum of all message counts in all internal queues.
         """
-        total = sum(self.qsize().values())
+        # qsize() now includes "total_size", so we need to be careful not to double count if we use qsize() values
+        # But qsize() implementation above sums values from queue_streams, then adds total_size.
+        # So sum(self.queue_streams.values().qsize()) is safer.
+        total = sum(queue.qsize() for queue in self.queue_streams.values())
         logger.debug(f"Total unfinished tasks across all queues: {total}")
         return total
 
@@ -249,3 +251,15 @@ def full(self) -> bool:
         return any(
             q.qsize() >= self.max_internal_message_queue_size for q in self.queue_streams.values()
         )
+
+    def ack_message(
+        self,
+        user_id: str,
+        mem_cube_id: str,
+        task_label: str,
+        redis_message_id,
+        message: ScheduleMessageItem | None,
+    ) -> None:
+        """
+        Acknowledge a message (no-op for local queue as messages are popped immediately).
+        """