Fix queue bugs for v1.8.1 branch

cecli/coders/base_coder.py

@@ -1615,6 +1615,10 @@ async def run_stream(self, user_message):
         async for chunk in self.send_message(user_message):
             yield chunk
 
+        # Stream-driven interfaces bypass run_one(), so drain the queue here
+        # to keep queued prompts processing after the request completes.
+        await self.process_prompt_queue()
+
     def init_before_message(self):
         self.coder_edited_files = set()
         self.reflected_message = None
@@ -2014,6 +2018,56 @@ async def preproc_user_input(self, inp):
     def wrap_user_input(self, inp):
         return inp
 
+    async def process_prompt_queue(self, preproc=True):
+        """Drain the coder's prompt queue in FIFO order.
+
+        Re-entrancy safe: a nested call (e.g. a queued prompt that itself
+        triggers a drain) returns immediately. Each item is isolated so one
+        failing prompt does not abort the rest of the queue. An interrupt
+        (Ctrl+C / cancellation) re-queues the in-flight item at the front and
+        re-raises so the caller can resume later via /run-queue.
+
+        Each item runs through ``generate()`` rather than a bare ``run_one()``.
+        ``generate()`` owns the generation lifecycle: it clears the interrupt
+        event, sets ``run_one_completed`` False for the duration, drives the
+        spinner, and wires ``io.output_task``. A bare ``run_one()`` skips all
+        of that, so a drain that called it directly (e.g. the TUI idle drain on
+        the worker loop) started the turn but stalled silently. ``_processing_queue``
+        still guards against the nested drain that ``run_one()`` performs at its
+        end.
+        """
+        if self._processing_queue:
+            return
+
+        self._processing_queue = True
+        try:
+            while self.prompt_queue:
+                item = command_queue.dequeue_prompt(self)
+                if item is None:
+                    break
+
+                self.io.tool_output(f"Processing queued prompt (id: {item['id']})...")
+                try:
+                    await self.generate(item["text"], preproc)
+                except (asyncio.CancelledError, KeyboardInterrupt):
+                    # Put the in-flight prompt back at the front so it is not
+                    # lost, then let the interrupt propagate to the caller.
+                    command_queue.insert_prompt(self, item["text"], 0)
+                    remaining = command_queue.get_queue_length(self)
+                    self.io.tool_warning(
+                        f"Queue processing interrupted. {remaining} prompt(s) remain"
+                        " queued; use /run-queue to resume."
+                    )
+                    raise
+                except (SwitchCoderSignal, SystemExit):
+                    raise
+                except Exception as e:
+                    # Isolate failures: log and continue with the next item.
+                    self.io.tool_error(f"Error processing queued prompt (id: {item['id']}): {e}")
+                    continue
+        finally:
+            self._processing_queue = False
+
     async def run_one(self, user_message, preproc):
         self.init_before_message()
 
@@ -2029,6 +2083,10 @@ async def run_one(self, user_message, preproc):
         if self.commands.is_command(user_message) and not self.commands.is_test_command(
             user_message
         ):
+            # Command path: may have queued prompts (e.g. /queue); drain so
+            # idle-queued prompts start immediately. The _processing_queue
+            # guard prevents double-drain when this runs inside a drain loop.
+            await self.process_prompt_queue(preproc)
             return
 
         if not self.commands.is_command(user_message):
@@ -2100,19 +2158,10 @@ async def run_one(self, user_message, preproc):
 
             await self.auto_save_session(force=True)
 
-        # Move to the next queued prompt (CLI-33) only after the current message
-        # has fully completed, so the queue drains within run_one() instead of
+        # Drain any queued prompts (CLI-33) only after the current message has
+        # fully completed, so the queue is processed within run_one() instead of
         # being watched by the generation loops.
-        if self.prompt_queue and not self._processing_queue:
-            self._processing_queue = True
-            try:
-                item = command_queue.dequeue_prompt(self)
-            finally:
-                self._processing_queue = False
-
-            if item is not None:
-                self.io.tool_output(f"Processing queued prompt (id: {item['id']})...")
-                await self.run_one(item["text"], preproc)
+        await self.process_prompt_queue(preproc)
 
         if not await HookIntegration.call_end_hooks(self):
             self.io.tool_warning("Execution stopped by end hook")

cecli/commands/__init__.py

@@ -73,6 +73,7 @@
 from .reset import ResetCommand
 from .rules import RulesCommand
 from .run import RunCommand
+from .run_queue import RunQueueCommand
 from .save import SaveCommand
 from .save_session import SaveSessionCommand
 from .search_memory import SearchMemoryCommand
@@ -168,6 +169,7 @@
 CommandRegistry.register(ResetCommand)
 CommandRegistry.register(RulesCommand)
 CommandRegistry.register(RunCommand)
+CommandRegistry.register(RunQueueCommand)
 CommandRegistry.register(SaveCommand)
 CommandRegistry.register(SaveSessionCommand)
 CommandRegistry.register(SearchMemoryCommand)
@@ -263,6 +265,7 @@
     "ResetCommand",
     "RulesCommand",
     "RunCommand",
+    "RunQueueCommand",
     "SaveCommand",
     "SaveSessionCommand",
     "SearchMemoryCommand",

cecli/commands/core.py

@@ -14,6 +14,12 @@
 class Commands:
     scraper = None
 
+    # Queue commands need the FULL text after the command name (including the
+    # trailing space) passed as completion args, so nested slash-command
+    # completion (e.g. "/queue /add <tab>") can see the inner command and its
+    # arguments (plan 9.13). Both completers (io.py, tui/app.py) consult this.
+    FULL_ARGS_COMPLETION_COMMANDS = ("/queue", "/insert-queue")
+
     def _get_coder(self):
         """Return coder via weak reference, or None if collected."""
         if self._coder_ref is not None:
@@ -90,7 +96,7 @@ def __init__(
         self.last_command_show_notification = True
 
         # Commands that should NOT trigger auto-processing of the queue
-        self._MANAGEMENT_COMMANDS = {"queue", "list-queue", "remove-queue"}
+        self._MANAGEMENT_COMMANDS = {"queue", "list-queue", "remove-queue", "run-queue"}
 
     def _load_custom_commands(self, custom_commands):
         """

cecli/commands/insert_queue.py

@@ -1,10 +1,12 @@
 """Insert-queue command for CLI-33: inserts a prompt at a specific queue position."""
 
+import asyncio
 from typing import List
 
 from cecli.commands.utils.base_command import BaseCommand
 from cecli.commands.utils.helpers import format_command_result
 from cecli.helpers import command_queue
+from cecli.signals import SwitchCoderSignal
 
 
 class InsertQueueCommand(BaseCommand):
@@ -60,6 +62,22 @@ async def execute(cls, io, coder, args, **kwargs):
         try:
             item = command_queue.insert_prompt(coder, prompt_text, index)
             io.tool_output(f"Prompt inserted at position {index + 1} (id: {item['id']})")
+
+            # Idle-immediate: when no request is in flight there is nothing to
+            # wait for, so drain the queue right away. When a request IS in
+            # flight, the queue auto-drains when run_one() completes.
+            target = command_queue.get_active_coder(coder)
+            if command_queue.idle_drain_needed(target):
+                try:
+                    await target.process_prompt_queue()
+                except (KeyboardInterrupt, asyncio.CancelledError):
+                    # The interrupted item was re-queued; let the user know the queue paused.
+                    io.tool_output("Queue processing paused. Use /run-queue to resume.")
+                except (SwitchCoderSignal, SystemExit):
+                    raise
+                except Exception as e:
+                    return format_command_result(io, cls.NORM_NAME, "", error=str(e))
+
             return f"Successfully executed {cls.NORM_NAME}."
         except ValueError as e:
             return format_command_result(io, cls.NORM_NAME, "", error=str(e))
@@ -68,8 +86,20 @@ async def execute(cls, io, coder, args, **kwargs):
 
     @classmethod
     def get_completions(cls, io, coder, args) -> List[str]:
-        """Get completion options for insert-queue command."""
-        return []
+        """Get completion options for insert-queue command.
+
+        Args are "<index> <prompt text>"; once the index token and a space
+        are present, mirror the nested prompt's completions (e.g. a nested
+        slash command delegates to the original command's completions).
+        """
+        text = args or ""
+        first_space = text.find(" ")
+        if first_space == -1:
+            # Still typing the index; suggest valid positions.
+            if not coder.commands:
+                return []
+            return [str(i) for i in range(1, command_queue.get_queue_length(coder) + 2)]
+        return command_queue.nested_command_completions(coder, text[first_space + 1 :])
 
     @classmethod
     def get_help(cls) -> str:

cecli/commands/queue.py

@@ -1,10 +1,12 @@
 """Queue command for CLI-33: adds a prompt to the processing queue."""
 
+import asyncio
 from typing import List
 
 from cecli.commands.utils.base_command import BaseCommand
 from cecli.commands.utils.helpers import format_command_result
 from cecli.helpers import command_queue
+from cecli.signals import SwitchCoderSignal
 
 
 class QueueCommand(BaseCommand):
@@ -59,6 +61,22 @@ async def execute(cls, io, coder, args, **kwargs):
             item = command_queue.enqueue_prompt(coder, prompt_text)
             position = command_queue.get_queue_length(coder)
             io.tool_output(f"Prompt queued at position {position} (id: {item['id']})")
+
+            # Idle-immediate: when no request is in flight there is nothing to
+            # wait for, so drain the queue right away. When a request IS in
+            # flight, the queue auto-drains when run_one() completes.
+            target = command_queue.get_active_coder(coder)
+            if command_queue.idle_drain_needed(target):
+                try:
+                    await target.process_prompt_queue()
+                except (KeyboardInterrupt, asyncio.CancelledError):
+                    # The interrupted item was re-queued; let the user know the queue paused.
+                    io.tool_output("Queue processing paused. Use /run-queue to resume.")
+                except (SwitchCoderSignal, SystemExit):
+                    raise
+                except Exception as e:
+                    return format_command_result(io, cls.NORM_NAME, "", error=str(e))
+
             return f"Successfully executed {cls.NORM_NAME}."
         except ValueError as e:
             return format_command_result(io, cls.NORM_NAME, "", error=str(e))
@@ -67,8 +85,12 @@ async def execute(cls, io, coder, args, **kwargs):
 
     @classmethod
     def get_completions(cls, io, coder, args) -> List[str]:
-        """Get completion options for queue command."""
-        return []
+        """Get completion options for queue command.
+
+        When the queued prompt starts with a slash command, mirror the
+        original command's completions (e.g. /queue /add <tab> offers files).
+        """
+        return command_queue.nested_command_completions(coder, args)
 
     @classmethod
     def get_help(cls) -> str:

cecli/commands/run_queue.py

@@ -0,0 +1,79 @@
+"""Run-queue command for CLI-33: processes queued prompts on demand.
+
+Primary use: after cancelling an in-flight response (Ctrl+C), the in-flight
+prompt is re-queued at the front. /run-queue resumes processing the queue.
+"""
+
+import asyncio
+from typing import List
+
+from cecli.commands.utils.base_command import BaseCommand
+from cecli.commands.utils.helpers import format_command_result
+from cecli.helpers import command_queue
+from cecli.signals import SwitchCoderSignal
+
+
+class RunQueueCommand(BaseCommand):
+    NORM_NAME = "run-queue"
+    DESCRIPTION = "Process queued prompts in order"
+
+    @classmethod
+    async def execute(cls, io, coder, args, **kwargs):
+        """Execute the run-queue command with given parameters.
+
+        Args:
+            io: InputOutput instance
+            coder: Coder instance (may be None for some commands)
+            args: Command arguments (unused for run-queue)
+            **kwargs: Additional context
+
+        Returns:
+            Formatted result string
+        """
+        # Sad path: coder.commands is None
+        if not coder.commands:
+            return format_command_result(
+                io, cls.NORM_NAME, "", error="Command system not available. Cannot run queue."
+            )
+
+        # Resolve the target coder (sub-agent aware: use the foreground agent's queue)
+        target = command_queue.get_active_coder(coder)
+
+        # Sad path: empty queue
+        if command_queue.get_queue_length(target) == 0:
+            io.tool_output("Queue is empty. Use /queue <prompt> to add prompts first.")
+            return f"Successfully executed {cls.NORM_NAME}."
+
+        # Happy path: drain the queue
+        try:
+            await target.process_prompt_queue()
+            return f"Successfully executed {cls.NORM_NAME}."
+        except (KeyboardInterrupt, asyncio.CancelledError):
+            # The interrupted item was re-queued; let the user know the queue paused.
+            io.tool_output("Queue processing paused. Use /run-queue to resume.")
+            return f"Successfully executed {cls.NORM_NAME}."
+        except (SwitchCoderSignal, SystemExit):
+            raise
+        except Exception as e:
+            return format_command_result(io, cls.NORM_NAME, "", error=str(e))
+
+    @classmethod
+    def get_completions(cls, io, coder, args) -> List[str]:
+        """Get completion options for run-queue command."""
+        return []
+
+    @classmethod
+    def get_help(cls) -> str:
+        """Get help text for the run-queue command."""
+        help_text = super().get_help()
+        help_text += "\nUsage:\n"
+        help_text += "  /run-queue  # Process all queued prompts in FIFO order\n"
+        help_text += "\nDescription:\n"
+        help_text += "  Processes every prompt currently in the queue, in order.\n"
+        help_text += "  Queued prompts normally run automatically after the current\n"
+        help_text += "  response finishes. Use /run-queue to resume a queue that was\n"
+        help_text += "  paused (e.g. after cancelling an in-flight response with Ctrl+C).\n"
+        help_text += "\nExamples:\n"
+        help_text += "  /run-queue  # Drains the queue, processing each prompt in order\n"
+        help_text += "\nSee also: /queue, /list-queue, /remove-queue\n"
+        return help_text

cecli/helpers/command_queue.py

@@ -151,6 +151,48 @@ def clear_queue(coder) -> list:
     return items
 
 
+def idle_drain_needed(coder) -> bool:
+    """True when the coder is idle and not already draining its queue.
+
+    Used by the command-side drains (queue.py / insert_queue.py) and the TUI
+    submit handler to decide whether an enqueue should trigger an immediate
+    drain. When a request is in flight (``run_one_completed`` False) or a
+    drain is already running (``_processing_queue`` True), the post-run_one
+    auto-drain covers the queue instead.
+    """
+    return bool(getattr(coder, "run_one_completed", True)) and not getattr(
+        coder, "_processing_queue", False
+    )
+
+
+def nested_command_completions(coder, text: str) -> list:
+    """Completions for a nested slash-command inside queued prompt text (CLI-33).
+
+    While the inner command is still being typed (no space yet), suggest
+    matching command names from the registry (e.g. "/sa" -> "/save-session").
+    Once the inner command is complete, delegate to the original command's
+    ``get_completions`` so ``/queue /add <tab>`` behaves exactly like ``/add <tab>``.
+    """
+    if not text or not text.startswith("/"):
+        return []
+
+    from cecli.commands.utils.registry import CommandRegistry
+
+    first_space = text.find(" ")
+    if first_space == -1:
+        prefix = text[1:].lower()
+        return sorted(
+            f"/{cmd}" for cmd in CommandRegistry.list_commands() if cmd.startswith(prefix)
+        )
+
+    inner_cmd = text[:first_space]
+    inner_args = text[first_space + 1 :]
+    try:
+        return list(coder.commands.get_completions(inner_cmd, inner_args, coder=coder))
+    except Exception:
+        return []
+
+
 def _get_lock(coder) -> threading.Lock:
     """Return the coder's queue lock, creating one if missing."""
     lock = getattr(coder, "_queue_lock", None)