diff --git a/CHANGELOG.md b/CHANGELOG.md
index 988b4e2c460..bddae2220fd 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -30,3 +30,17 @@
 * [Benchmark Results By Language](https://github.com/dwash96/aider-ce/pull/27)
 * [Allow Benchmarks to Use Repo Map For Better Accuracy](https://github.com/dwash96/aider-ce/pull/25)
 * [Read File Globbing](https://github.com/Aider-AI/aider/pull/3395)
+
+### Prompt Queueing (CLI-33)
+- Added `/queue <prompt>` to add prompts to a FIFO queue.
+- Added `/list-queue` to view queued prompts.
+- Added `/remove-queue [index|*]` to remove specific prompts or clear the queue.
+- Queued prompts process automatically after the current command completes.
+- Queue state persists across command executions within a session.
+- Management commands do not trigger queue processing.
+- Queue data structure implemented in `Commands` class with thread-safe operations using `asyncio.Lock`.
+- Max queue size limit of 100 items.
+* [Prompt Queueing](https://github.com/cecli-dev/cecli/issues/33)
+    * Added `/queue`, `/list-queue`, and `/remove-queue` commands for deferred prompt processing.
+    * Queued prompts process automatically in FIFO order when the system is idle.
+    * Queue state is in-memory and session-specific.
diff --git a/README.md b/README.md
index aaf57773b5c..a482d24910f 100644
--- a/README.md
+++ b/README.md
@@ -24,7 +24,7 @@ Use curl to download the script and execute it with sh:
 curl -LsSf https://cecli.dev/install.sh | sh
 ```
 
-If your system doesn't have curl, you can use wget:
+If your system does not have curl, you can use wget:
 
 ```bash
 wget -qO- https://cecli.dev/install.sh | sh
@@ -53,7 +53,7 @@ pip install cecli-dev
 uv tool install --native-tls --python python3.12 cecli-dev
 ```
 
-Use the tool installation so cecli doesn't interfere with your development environment
+Use the tool installation so cecli doesn't interfere with your development environment.
 
 ## Features and Documentation:
 
diff --git a/cecli/coders/base_coder.py b/cecli/coders/base_coder.py
index bf52fff2d51..2a0cfc61895 100755
--- a/cecli/coders/base_coder.py
+++ b/cecli/coders/base_coder.py
@@ -338,6 +338,15 @@ async def create(
             kwargs = use_kwargs
             from_coder.ok_to_warm_cache = False
 
+            # Preserve the prompt queue across a model switch (same coder
+            # identity) but NOT when spawning a distinct sub-agent (which is
+            # assigned a fresh uuid in its kwargs). Model switches keep
+            # from_coder.uuid; sub-agents override it with a new uuid.
+            is_model_switch = from_coder.uuid == kwargs.get("uuid", "")
+            if is_model_switch:
+                kwargs.setdefault("prompt_queue", from_coder.prompt_queue)
+                kwargs.setdefault("_queue_counter", from_coder._queue_counter)
+
         res = None
         if (
             getattr(main_model, "copy_paste_mode", False)
@@ -357,6 +366,17 @@ async def create(
                 if from_coder.tui:
                     res.tui = from_coder.tui
 
+                # Preserve prompt queue state across model switches (CLI-33).
+                # The queue lives on the coder (see command_queue.py) so the
+                # new coder instance must inherit the list and counter, while
+                # the lock and processing flag are intentionally fresh.
+                # Only model switches inherit the queue; sub-agents get their
+                # own isolated queue (each agent has its own context/queue).
+                if is_model_switch:
+                    res.prompt_queue = from_coder.prompt_queue
+                    res._queue_counter = from_coder._queue_counter
+                res.tui = from_coder.tui
+
                 # Sub-agents get a dedicated, independent MCP manager so they
                 # can rebuild a custom tool list (their own LocalServer tools /
                 # filters) and be disconnected independently from the parent.
@@ -457,6 +477,8 @@ def __init__(
         root=None,
         primary_root=None,
         init_metadata={},
+        prompt_queue=None,
+        _queue_counter=None,
     ):
         from cecli.helpers.agents.service import AgentService
 
@@ -611,12 +633,12 @@ def __init__(
         self.commands = commands or Commands(self.io, self, args=args)
         self.commands.coder = self
 
-        # Prompt queue for CLI-33: in-memory FIFO queue for deferred prompt
-        # processing. The queue lives on the coder so primary agents and
-        # sub-agents each have their own independent queue, managed by
-        # cecli.helpers.command_queue.
-        self.prompt_queue = []
-        self._queue_counter = 0
+        # Prompt queue for CLI-33: in-memory FIFO queue managed by
+        # cecli.helpers.command_queue. The queue lives on the Coder so it
+        # survives model switches (Coder.create preserves identity) while each
+        # sub-agent gets its own isolated queue.
+        self.prompt_queue = list(prompt_queue) if prompt_queue else []
+        self._queue_counter = _queue_counter if _queue_counter else 0
         self._queue_lock = threading.Lock()
         self._processing_queue = False
 
@@ -1962,97 +1984,138 @@ async def run_one(self, user_message, preproc):
         else:
             message = user_message
 
-        if self.commands.is_command(user_message) and not self.commands.is_test_command(
-            user_message
-        ):
-            return
+        # /list-queue and /remove-queue must not trigger auto-processing of the
+        # queue (CLI-33). Every other message/command — including /queue, so a
+        # prompt queued while idle processes immediately — drains the queue once
+        # it has fully completed, so queued prompts are sent to the LLM after
+        # the current prompt finishes.
+        is_management_command = self._is_queue_management_command(user_message)
 
-        if not self.commands.is_command(user_message):
-            ConversationService.get_chunks(self).flush_removals()
-            self.last_user_message = user_message
-            self.error_code = None
-            self.num_tool_calls = 0
-            # Trim memory in the background so it doesn't delay the response
-            coroutines.fire_and_forget(asyncio.to_thread(trim_memory))
-            # Fire memorizer after each user request
-            # if self.auto_memory and self.edit_format not in ["subagent"]:
-            #    from cecli.helpers.memory.utils import invoke_memorizer
-            #
-            #    context = "If the user has stated any preferences, please remember them"
-            #    asyncio.create_task(invoke_memorizer(self, additional_context=context))
+        try:
+            if self.commands.is_command(user_message) and not self.commands.is_test_command(
+                user_message
+            ):
+                return
 
-        while True:
-            self.reflected_message = None
-            self.empty_response = False
-            self.tool_reflection = False
+            if not self.commands.is_command(user_message):
+                ConversationService.get_chunks(self).flush_removals()
+                self.last_user_message = user_message
+                self.error_code = None
+                self.num_tool_calls = 0
+                # Trim memory in the background so it doesn't delay the response
+                coroutines.fire_and_forget(asyncio.to_thread(trim_memory))
+                # Fire memorizer after each user request
+                # if self.auto_memory and self.edit_format not in ["subagent"]:
+                #    from cecli.helpers.memory.utils import invoke_memorizer
+                #
+                #    context = "If the user has stated any preferences, please remember them"
+                #    asyncio.create_task(invoke_memorizer(self, additional_context=context))
 
-            if float(self.total_cost) > self.cost_multiplier * (
-                nested.getter(self.args, "cost_limit", float("inf")) or float("inf")
-            ):
-                if await self.io.confirm_ask(
-                    "You have reached your configured cost limit. Continue?",
-                    group_response="Cost Limit",
-                    explicit_yes_required=True,
+            while True:
+                self.reflected_message = None
+                self.empty_response = False
+                self.tool_reflection = False
+
+                if float(self.total_cost) > self.cost_multiplier * (
+                    nested.getter(self.args, "cost_limit", float("inf")) or float("inf")
                 ):
-                    Coder.cost_multiplier += 1
-                else:
-                    return
+                    if await self.io.confirm_ask(
+                        "You have reached your configured cost limit. Continue?",
+                        group_response="Cost Limit",
+                        explicit_yes_required=True,
+                    ):
+                        Coder.cost_multiplier += 1
+                    else:
+                        return
 
-            async for _ in self.send_message(message):
-                pass
+                async for _ in self.send_message(message):
+                    pass
 
-            await self.hot_reload()
+                await self.hot_reload()
 
-            if not self.empty_response:
-                if not self.reflected_message:
-                    await self.auto_save_session(force=True)
-                    break
+                if not self.empty_response:
+                    if not self.reflected_message:
+                        await self.auto_save_session(force=True)
+                        break
 
-                if self.num_reflections >= self.max_reflections:
-                    self.io.tool_warning(
-                        f"Only {self.max_reflections} reflections allowed, stopping."
-                    )
-                    break
+                    if self.num_reflections >= self.max_reflections:
+                        self.io.tool_warning(
+                            f"Only {self.max_reflections} reflections allowed, stopping."
+                        )
+                        break
 
-                self.num_reflections += 1
+                    self.num_reflections += 1
 
-                if self.tool_reflection:
-                    self.num_reflections -= 1
+                    if self.tool_reflection:
+                        self.num_reflections -= 1
 
-                if self.reflected_message is True:
-                    message = None
-                else:
-                    message = self.reflected_message
-            elif self.stop_on_empty:
-                await self.auto_save_session(force=True)
-                break
+                    if self.reflected_message is True:
+                        message = None
+                    else:
+                        message = self.reflected_message
+                elif self.stop_on_empty:
+                    await self.auto_save_session(force=True)
+                    break
 
-            if self.enable_context_compaction:
-                await self.compact_context_if_needed()
+                if self.enable_context_compaction:
+                    await self.compact_context_if_needed()
+
+                if nested.getter(self, "agent_finished", False):
+                    await self.auto_save_session(force=True)
+                    break
 
-            if nested.getter(self, "agent_finished", False):
                 await self.auto_save_session(force=True)
-                break
+        finally:
+            # Drain the queue once the current message/command has fully
+            # completed, so queued prompts are sent to the LLM (CLI-33).
+            if not is_management_command:
+                await self._drain_prompt_queue(preproc)
 
-            await self.auto_save_session(force=True)
+        if not await HookIntegration.call_end_hooks(self):
+            self.io.tool_warning("Execution stopped by end hook")
+            return
 
-        # 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
-        # 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
+    def _is_queue_management_command(self, user_message):
+        """Return True if user_message is a read-only queue-management command.
+
+        Only /list-queue and /remove-queue are excluded from auto-processing:
+        they inspect or mutate the queue without the user asking queued prompts
+        to run. /queue is deliberately NOT management here, so a prompt queued
+        while the coder is idle drains (and processes) immediately afterward.
+        """
+        if not self.commands or not self.commands.is_command(user_message):
+            return False
+        words = user_message.strip().split()
+        if not words:
+            return False
+        cmd_name = words[0][1:]
+        return cmd_name in ("list-queue", "remove-queue")
 
-            if item is not None:
-                self.io.tool_output(f"Processing queued prompt (id: {item['id']})...")
-                await self.run_one(item["text"], preproc)
+    async def _drain_prompt_queue(self, preproc):
+        """Process all queued prompts (FIFO) once the current message completes.
 
-        if not await HookIntegration.call_end_hooks(self):
-            self.io.tool_warning("Execution stopped by end hook")
+        Runs each queued prompt through ``run_one`` so it is sent to the LLM
+        exactly like a user-typed prompt. A single failing queued prompt is
+        logged and does not stop the remaining ones (``SwitchCoderSignal`` and
+        ``ReloadProgramSignal`` are BaseExceptions and propagate unchanged).
+        """
+        if self._processing_queue:
             return
+        self._processing_queue = True
+        try:
+            while True:
+                item = command_queue.dequeue_prompt(self)
+                if item is None:
+                    break
+                text = item["text"]
+                preview = text if len(text) <= 80 else text[:80] + "..."
+                self.io.tool_output(f"Processing queued prompt: {preview}")
+                try:
+                    await self.run_one(text, preproc)
+                except Exception as e:
+                    self.io.tool_error(f"Error processing queued prompt: {e}")
+        finally:
+            self._processing_queue = False
 
     def _is_url_allowed(self, url):
         allowed_domains = self.security_config.get("allowed-domains")
diff --git a/cecli/commands/__init__.py b/cecli/commands/__init__.py
index 0905610dc01..c76e4262dbc 100644
--- a/cecli/commands/__init__.py
+++ b/cecli/commands/__init__.py
@@ -38,6 +38,7 @@
 from .hooks import HooksCommand
 from .hot_reload import HotReloadCommand
 from .include_skill import IncludeSkillCommand
+from .insert_queue import InsertQueueCommand
 from .lint import LintCommand
 from .list_mcp import ListMcpCommand
 from .list_queue import ListQueueCommand
@@ -134,9 +135,11 @@
 CommandRegistry.register(SpawnAgentCommand)
 CommandRegistry.register(SwitchAgentCommand)
 CommandRegistry.register(IncludeSkillCommand)
+CommandRegistry.register(InsertQueueCommand)
 CommandRegistry.register(LintCommand)
 CommandRegistry.register(ListMcpCommand)
 CommandRegistry.register(ListQueueCommand)
+CommandRegistry.register(RemoveQueueCommand)
 CommandRegistry.register(ListSessionsCommand)
 CommandRegistry.register(ListSkillsCommand)
 CommandRegistry.register(LoadCommand)
@@ -221,11 +224,13 @@
     "HooksCommand",
     "HotReloadCommand",
     "IncludeSkillCommand",
+    "InsertQueueCommand",
     "ReapAgentCommand",
     "SpawnAgentCommand",
     "SwitchAgentCommand",
     "LintCommand",
     "ListSessionsCommand",
+    "ListQueueCommand",
     "ListSkillsCommand",
     "LoadCommand",
     "LoadHookCommand",
@@ -244,6 +249,7 @@
     "PasteCommand",
     "quote_filename",
     "QueueCommand",
+    "RemoveQueueCommand",
     "QuitCommand",
     "ReadOnlyCommand",
     "ReadOnlyStubCommand",
diff --git a/cecli/commands/core.py b/cecli/commands/core.py
index c8df65069af..116fc6b6f35 100644
--- a/cecli/commands/core.py
+++ b/cecli/commands/core.py
@@ -1,12 +1,9 @@
-import json
-import re
-import sys
+import asyncio
 import weakref
 from pathlib import Path
 
 from cecli.commands.utils.registry import CommandRegistry
-from cecli.helpers import nested, plugin_manager
-from cecli.helpers.file_searcher import handle_core_files
+from cecli.helpers import plugin_manager
 from cecli.helpers.threading import ThreadSafeEvent
 from cecli.signals import SwitchCoderSignal
 
@@ -14,35 +11,6 @@
 class Commands:
     scraper = None
 
-    def _get_coder(self):
-        """Return coder via weak reference, or None if collected."""
-        if self._coder_ref is not None:
-            return self._coder_ref()
-        return None
-
-    def _set_coder(self, value):
-        """Store coder as weakref to break circular reference chains."""
-        self._coder_ref = weakref.ref(value) if value is not None else None
-
-    coder = property(_get_coder, _set_coder)
-
-    def clone(self):
-        cloned = Commands(
-            self.io,
-            None,
-            voice_language=self.voice_language,
-            voice_input_device=self.voice_input_device,
-            voice_format=self.voice_format,
-            verify_ssl=self.verify_ssl,
-            args=self.args,
-            parser=self.parser,
-            verbose=self.verbose,
-            editor=self.editor,
-            original_read_only_fnames=self.original_read_only_fnames,
-        )
-        cloned.last_command_show_notification = self.last_command_show_notification
-        return cloned
-
     def __init__(
         self,
         io,
@@ -58,39 +26,29 @@ def __init__(
         original_read_only_fnames=None,
     ):
         self.io = io
-        # Use weak ref to avoid circular reference chains
-        self._coder_ref = weakref.ref(coder) if coder else None
-        self.parser = parser
+        self.coder = weakref.proxy(coder) if coder else None
         self.args = args
-        self.verbose = verbose
-        self.verify_ssl = verify_ssl
-        if voice_language == "auto":
-            voice_language = None
+        self.parser = parser
         self.voice_language = voice_language
-        self.voice_format = voice_format
         self.voice_input_device = voice_input_device
-        self.help = None
+        self.voice_format = voice_format
+        self.verify_ssl = verify_ssl
+        self.verbose = verbose
         self.editor = editor
-        self.original_read_only_fnames = set(original_read_only_fnames or [])
-
-        customizations = dict()
-        try:
-            if self.args:
-                customizations = nested.getter(self.args, "custom", "{}")
-                customizations = json.loads(customizations)
-        except (json.JSONDecodeError, TypeError):
-            customizations = dict()
-            pass
-
-        self.custom_commands = nested.getter(customizations, "command-paths", [])
-        self._load_custom_commands(self.custom_commands)
+        self.original_read_only_fnames = original_read_only_fnames
 
         self.cmd_running_event = ThreadSafeEvent()
         self.cmd_running_event.set()
         self.last_command_show_notification = True
 
+        # Prompt queue for CLI-33: in-memory FIFO queue
+        self.prompt_queue = []
+        self._queue_counter = 0
+        self._queue_lock = asyncio.Lock()
+        self._processing_queue = False
+
         # 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", "insert-queue"}
 
     # ── Queue Management Methods (CLI-33) ────────────────────────────── #
     #
@@ -101,41 +59,95 @@ def __init__(
     # instance, so each sub-agent's commands manage that sub-agent's own
     # queue.
 
-    @property
-    def prompt_queue(self):
-        """Proxy to the owning coder's prompt queue."""
-        coder = self.coder
-        return coder.prompt_queue if coder is not None else []
+    def _active_coder(self):
+        """Resolve the coder queue commands should target.
+
+        Prefers the foreground (sub-agent) coder via
+        ``command_queue.get_active_coder``, falling back to ``self.coder``
+        when no active coder can be resolved (e.g. no AgentService, or the
+        Commands instance is constructed without a coder in tests).
+        """
+        from cecli.helpers import command_queue
+
+        return command_queue.get_active_coder(self.coder) or self.coder
 
     def _enqueue_prompt(self, text: str) -> dict:
-        """Add a prompt to the owning coder's queue."""
+        """Add a prompt to the active (foreground) coder's queue."""
         from cecli.helpers import command_queue
 
-        return command_queue.enqueue_prompt(self.coder, text)
+        return command_queue.enqueue_prompt(self._active_coder(), text)
+
+    async def _process_queued_prompts(self, preproc):
+        """Process all queued prompts (FIFO) once the current message completes.
+
+        Runs each queued prompt through ``run_one`` so it is sent to the LLM
+        exactly like a user-typed prompt. A single failing queued prompt is
+        logged and does not stop the remaining ones (``SwitchCoderSignal`` and
+        ``ReloadProgramSignal`` are BaseExceptions and propagate unchanged).
+        """
+        if self._processing_queue:
+            return
+        self._processing_queue = True
+        try:
+            while True:
+                item = self._dequeue_prompt()
+                if item is None:
+                    break
+                text = item["text"]
+                preview = text if len(text) <= 80 else text[:80] + "..."
+                self.io.tool_output(f"Processing queued prompt: {preview}")
+                try:
+                    await self.coder.run_one(text, preproc)
+                except Exception as e:
+                    self.io.tool_error(f"Error processing queued prompt: {e}")
+        finally:
+            self._processing_queue = False
+
+    def _insert_prompt(self, text: str, index: int) -> dict:
+        """Insert a prompt at the given index in the active coder's queue."""
+        from cecli.helpers import command_queue
+
+        return command_queue.insert_prompt(self._active_coder(), text, index)
 
     def _dequeue_prompt(self) -> dict | None:
-        """Remove and return the first item from the owning coder's queue."""
+        """Remove and return the first item from the active coder's queue."""
         from cecli.helpers import command_queue
 
-        return command_queue.dequeue_prompt(self.coder)
+        return command_queue.dequeue_prompt(self._active_coder())
 
     def _get_queue_length(self) -> int:
-        """Return the current number of items in the owning coder's queue."""
+        """Return the current number of items in the active coder's queue."""
         from cecli.helpers import command_queue
 
-        return command_queue.get_queue_length(self.coder)
+        return command_queue.get_queue_length(self._active_coder())
 
     def _remove_from_queue(self, index: int) -> dict | None:
-        """Remove and return the item at the given index from the owning coder's queue."""
+        """Remove and return the item at the given index from the active coder's queue."""
         from cecli.helpers import command_queue
 
-        return command_queue.remove_from_queue(self.coder, index)
+        return command_queue.remove_from_queue(self._active_coder(), index)
 
     def _clear_queue(self) -> list:
-        """Remove all items from the owning coder's queue and return them."""
+        """Remove all items from the active coder's queue and return them."""
         from cecli.helpers import command_queue
 
-        return command_queue.clear_queue(self.coder)
+        return command_queue.clear_queue(self._active_coder())
+
+    def clone(self):
+        """Create a clone of this Commands instance with updated parameters."""
+        return Commands(
+            self.io,
+            None,
+            voice_language=self.voice_language,
+            voice_input_device=self.voice_input_device,
+            voice_format=self.voice_format,
+            verify_ssl=self.verify_ssl,
+            args=self.args,
+            parser=self.parser,
+            verbose=self.verbose,
+            editor=self.editor,
+            original_read_only_fnames=self.original_read_only_fnames,
+        )
 
     def _load_custom_commands(self, custom_commands):
         """
@@ -220,6 +232,46 @@ def get_commands(self):
         commands = [f"/{cmd}" for cmd in registry_commands]
         return sorted(commands)
 
+    def matching_commands(self, inp):
+        words = inp.strip().split()
+        if not words:
+            return
+        first_word = words[0]
+        rest_inp = inp[len(words[0]) :].strip()
+        all_commands = self.get_commands()
+        matching_commands = [cmd for cmd in all_commands if cmd.startswith(first_word)]
+        return matching_commands, first_word, rest_inp
+
+    async def run(self, inp, coder=None, **kwargs):
+        if inp.startswith("/"):
+            words = inp.strip().split()
+            cmd_name = words[0][1:]
+            rest_inp = inp[len(words[0]) :].strip()
+            return await self.execute(cmd_name, rest_inp, coder=coder, **kwargs)
+
+        if inp.startswith("!!!"):
+            return await self.execute(
+                "run", inp[3:], coder=coder, background=True, suppress_add=True
+            )
+        if inp.startswith("!!"):
+            return await self.execute("run", inp[2:], coder=coder, suppress_add=True)
+        if inp.startswith("!"):
+            return await self.execute("run", inp[1:], coder=coder)
+        res = self.matching_commands(inp)
+        if res is None:
+            return
+        matching_commands, first_word, rest_inp = res
+        if len(matching_commands) == 1:
+            command = matching_commands[0][1:]
+            return await self.execute(command, rest_inp, coder=coder, **kwargs)
+        elif first_word in matching_commands:
+            command = first_word[1:]
+            return await self.execute(command, rest_inp, coder=coder, **kwargs)
+        elif len(matching_commands) > 1:
+            self.io.tool_error(f"Ambiguous command: {', '.join(matching_commands)}")
+        else:
+            self.io.tool_error(f"Invalid command: {first_word}")
+
     async def execute(self, cmd_name, args, coder=None, **kwargs):
         from cecli.repo import ANY_GIT_ERROR
 
@@ -263,86 +315,10 @@ async def execute(self, cmd_name, args, coder=None, **kwargs):
             self.cmd_running_event.set()
             if self.coder.tui and self.coder.tui():
                 self.coder.tui().refresh()
-
-    def matching_commands(self, inp):
-        words = inp.strip().split()
-        if not words:
-            return
-        first_word = words[0]
-        rest_inp = inp[len(words[0]) :].strip()
-        all_commands = self.get_commands()
-        matching_commands = [cmd for cmd in all_commands if cmd.startswith(first_word)]
-        return matching_commands, first_word, rest_inp
-
-    async def run(self, inp, coder=None, **kwargs):
-        if inp.startswith("/"):
-            words = inp.strip().split()
-            cmd_name = words[0][1:]
-            rest_inp = inp[len(words[0]) :].strip()
-            return await self.execute(cmd_name, rest_inp, coder=coder, **kwargs)
-
-        if inp.startswith("!!!"):
-            return await self.execute(
-                "run", inp[3:], coder=coder, background=True, suppress_add=True
-            )
-        if inp.startswith("!!"):
-            return await self.execute("run", inp[2:], coder=coder, suppress_add=True)
-        if inp.startswith("!"):
-            return await self.execute("run", inp[1:], coder=coder)
-        res = self.matching_commands(inp)
-        if res is None:
-            return
-        matching_commands, first_word, rest_inp = res
-        if len(matching_commands) == 1:
-            command = matching_commands[0][1:]
-            return await self.execute(command, rest_inp, coder=coder, **kwargs)
-        elif first_word in matching_commands:
-            command = first_word[1:]
-            return await self.execute(command, rest_inp, coder=coder, **kwargs)
-        elif len(matching_commands) > 1:
-            self.io.tool_error(f"Ambiguous command: {', '.join(matching_commands)}")
-        else:
-            self.io.tool_error(f"Invalid command: {first_word}")
-
-    def get_help_md(self):
-        """Show help about all commands in markdown"""
-        res = "\n|Command|Description|\n|:------|:----------|\n"
-        commands = sorted(self.get_commands())
-        for cmd in commands:
-            cmd_name = cmd[1:]
-            command_class = CommandRegistry.get_command(cmd_name)
-            if command_class:
-                description = command_class.DESCRIPTION
-                res += f"| **{cmd}** | {description} |\n"
-            else:
-                res += f"| **{cmd}** | |\n"
-        res += "\n"
-        return res
-
-    def _get_session_directory(self):
-        """Get the session storage directory, creating it if needed"""
-        session_dir = handle_core_files(Path(self.coder.root) / ".cecli" / "sessions")
-        session_dir.mkdir(parents=True, exist_ok=True)
-        return session_dir
-
-    def _get_session_file_path(self, session_name):
-        """Get the full path for a session file"""
-        session_dir = self._get_session_directory()
-        safe_name = re.sub("[^a-zA-Z0-9_.-]", "_", session_name)
-        ext = "" if safe_name[-5:] == ".json" else ".json"
-        return session_dir / f"{safe_name}{ext}"
-
-
-def get_help_md():
-    md = Commands(None, None).get_help_md()
-    return md
-
-
-def main():
-    md = get_help_md()
-    print(md)
-
-
-if __name__ == "__main__":
-    status = main()
-    sys.exit(status)
+            # NEW: Queue processing integration
+            if (
+                getattr(self.coder, "prompt_queue", None)
+                and cmd_name not in self._MANAGEMENT_COMMANDS
+                and not getattr(self.coder, "_processing_queue", False)
+            ):
+                await self.coder._drain_prompt_queue(kwargs.get("preproc", True))
diff --git a/cecli/commands/insert_queue.py b/cecli/commands/insert_queue.py
new file mode 100644
index 00000000000..fec5b0cda17
--- /dev/null
+++ b/cecli/commands/insert_queue.py
@@ -0,0 +1,86 @@
+"""Insert-queue command for CLI-33: inserts a prompt at a specific queue position."""
+
+from typing import List
+
+from cecli.commands.utils.base_command import BaseCommand
+from cecli.commands.utils.helpers import format_command_result
+
+
+class InsertQueueCommand(BaseCommand):
+    NORM_NAME = "insert-queue"
+    DESCRIPTION = "Insert a prompt into the queue at a specific position"
+
+    @classmethod
+    async def execute(cls, io, coder, args, **kwargs):
+        """Execute the insert-queue command with given parameters.
+
+        Args:
+            io: InputOutput instance
+            coder: Coder instance (may be None for some commands)
+            args: Command arguments as string ("<index> <prompt text>")
+            **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 insert into queue.",
+            )
+
+        # Sad path: missing index or prompt text
+        parts = (args or "").strip().split(maxsplit=1)
+        if len(parts) != 2:
+            return format_command_result(
+                io,
+                cls.NORM_NAME,
+                "",
+                error="Usage: /insert-queue <index> <prompt text>",
+            )
+
+        # Sad path: non-integer index
+        try:
+            index = int(parts[0])
+        except ValueError:
+            return format_command_result(
+                io,
+                cls.NORM_NAME,
+                "",
+                error=f"Invalid index: '{parts[0]}'. Please provide a number.",
+            )
+
+        prompt_text = parts[1].strip()
+
+        # Happy path: insert the prompt
+        try:
+            item = coder.commands._insert_prompt(prompt_text, index)
+            io.tool_output(f"Prompt inserted at position {index + 1} (id: {item['id']})")
+            return f"Successfully executed {cls.NORM_NAME}."
+        except ValueError as e:
+            return format_command_result(io, cls.NORM_NAME, "", error=str(e))
+        except RuntimeError 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 insert-queue command."""
+        return []
+
+    @classmethod
+    def get_help(cls) -> str:
+        """Get help text for the insert-queue command."""
+        help_text = super().get_help()
+        help_text += "\nUsage:\n"
+        help_text += "  /insert-queue <index> <prompt text>  # Insert at a specific position\n"
+        help_text += "\nDescription:\n"
+        help_text += "  Inserts a prompt into the queue at the given 1-based position.\n"
+        help_text += "  Existing items shift down. Index is clamped to the queue bounds.\n"
+        help_text += "\nExamples:\n"
+        help_text += "  /insert-queue 1 Review the changes in src/main.py\n"
+        help_text += "  /insert-queue 3 Write unit tests for the new feature\n"
+        help_text += "\nSee also: /queue, /list-queue, /remove-queue\n"
+        return help_text
diff --git a/cecli/commands/list_queue.py b/cecli/commands/list_queue.py
index 72305b09d8c..85ab6d9d1f8 100644
--- a/cecli/commands/list_queue.py
+++ b/cecli/commands/list_queue.py
@@ -27,7 +27,7 @@ async def execute(cls, io, coder, args, **kwargs):
         # Sad path: coder.commands is None
         if not coder.commands:
             return format_command_result(
-                io, cls.NORM_NAME, error="Command system not available. Cannot list queue."
+                io, cls.NORM_NAME, "", error="Command system not available. Cannot list queue."
             )
 
         queue = coder.commands.prompt_queue
diff --git a/cecli/commands/queue.py b/cecli/commands/queue.py
index 7438ff42326..0d3113b4d97 100644
--- a/cecli/commands/queue.py
+++ b/cecli/commands/queue.py
@@ -26,7 +26,7 @@ async def execute(cls, io, coder, args, **kwargs):
         # Sad path: coder.commands is None
         if not coder.commands:
             return format_command_result(
-                io, cls.NORM_NAME, error="Command system not available. Cannot queue prompts."
+                io, cls.NORM_NAME, "", error="Command system not available. Cannot queue prompts."
             )
 
         # Sad path: no args (empty prompt text)
@@ -34,8 +34,11 @@ async def execute(cls, io, coder, args, **kwargs):
             return format_command_result(
                 io,
                 cls.NORM_NAME,
-                "Usage: /queue <prompt text>\n"
-                "Add a prompt to the queue for processing after current tasks complete.",
+                "",
+                error=(
+                    "Usage: /queue <prompt text> - Add a prompt to the queue "
+                    "for processing after current tasks complete."
+                ),
             )
 
         prompt_text = args.strip()
@@ -45,6 +48,7 @@ async def execute(cls, io, coder, args, **kwargs):
             return format_command_result(
                 io,
                 cls.NORM_NAME,
+                "",
                 error=f"Prompt exceeds maximum length of 10000 characters "
                 f"(got {len(prompt_text)}).",
             )
@@ -56,9 +60,9 @@ async def execute(cls, io, coder, args, **kwargs):
             io.tool_output(f"Prompt queued at position {position} (id: {item['id']})")
             return f"Successfully executed {cls.NORM_NAME}."
         except ValueError as e:
-            return format_command_result(io, cls.NORM_NAME, error=str(e))
+            return format_command_result(io, cls.NORM_NAME, "", error=str(e))
         except RuntimeError as e:
-            return format_command_result(io, cls.NORM_NAME, error=str(e))
+            return format_command_result(io, cls.NORM_NAME, "", error=str(e))
 
     @classmethod
     def get_completions(cls, io, coder, args) -> List[str]:
diff --git a/cecli/commands/remove_queue.py b/cecli/commands/remove_queue.py
index fc555869fe7..4ef3c9c6b68 100644
--- a/cecli/commands/remove_queue.py
+++ b/cecli/commands/remove_queue.py
@@ -26,13 +26,16 @@ async def execute(cls, io, coder, args, **kwargs):
         # Sad path: coder.commands is None
         if not coder.commands:
             return format_command_result(
-                io, cls.NORM_NAME, error="Command system not available. Cannot remove from queue."
+                io,
+                cls.NORM_NAME,
+                "",
+                error="Command system not available. Cannot remove from queue.",
             )
 
         # Sad path: empty queue
         if coder.commands._get_queue_length() == 0:
             return format_command_result(
-                io, cls.NORM_NAME, error="Queue is empty. Nothing to remove."
+                io, cls.NORM_NAME, "", error="Queue is empty. Nothing to remove."
             )
 
         # Handle wildcard: clear entire queue
@@ -50,6 +53,7 @@ async def execute(cls, io, coder, args, **kwargs):
                 return format_command_result(
                     io,
                     cls.NORM_NAME,
+                    "",
                     error=f"Invalid index: '{args.strip()}'. Please provide a number or '*'.",
                 )
 
@@ -59,6 +63,7 @@ async def execute(cls, io, coder, args, **kwargs):
                 return format_command_result(
                     io,
                     cls.NORM_NAME,
+                    "",
                     error=f"Index {args.strip()} is out of range. Queue has {queue_len} item(s).",
                 )
 
diff --git a/cecli/format_settings.py b/cecli/format_settings.py
index 0ad54aa51aa..e74290ae05c 100644
--- a/cecli/format_settings.py
+++ b/cecli/format_settings.py
@@ -1,3 +1,6 @@
+import os
+
+
 def scrub_sensitive_info(args, text):
     # Replace sensitive information with last 4 characters
     if text and args.openai_api_key:
@@ -23,4 +26,17 @@ def format_settings(parser, args):
         if val:
             val = scrub_sensitive_info(args, str(val))
         show += f"  - {arg}: {val}\n"  # noqa: E221
+    # Add environment variables that start with CECLI_
+    show += "\nEnvironment variables:\n"
+    for env_var, env_val in sorted(os.environ.items()):
+        if env_var.startswith("CECLI_"):
+            # Scrub sensitive env vars if needed
+            if (
+                env_var
+                in ["CECLI_OPENROUTER_API_KEY", "CECLI_OPENAI_API_KEY", "CECLI_ANTHROPIC_API_KEY"]
+                and env_val
+            ):
+                last_4 = env_val[-4:] if len(env_val) >= 4 else env_val
+                env_val = f"...{last_4}"
+            show += f"  - {env_var}: {env_val}\n"
     return show
diff --git a/cecli/helpers/command_queue.py b/cecli/helpers/command_queue.py
index 5f1a0ff8f7c..1463cffc18c 100644
--- a/cecli/helpers/command_queue.py
+++ b/cecli/helpers/command_queue.py
@@ -61,6 +61,40 @@ def enqueue_prompt(coder, text: str) -> dict:
     return item
 
 
+def insert_prompt(coder, text: str, index: int) -> dict:
+    """Insert a prompt at the given 0-based index and return the queued item.
+
+    Args:
+        coder: The coder whose queue should be modified.
+        text: The prompt text to insert.
+        index: 0-based position to insert at. Clamped to [0, len(queue)].
+
+    Returns:
+        dict with keys: id (str), text (str), timestamp (float).
+
+    Raises:
+        ValueError: If text is empty, None, or exceeds 10000 characters.
+        RuntimeError: If the queue is at max capacity (100 items).
+    """
+    if not text or not text.strip():
+        raise ValueError("Cannot enqueue empty prompt")
+    if len(text) > MAX_PROMPT_LENGTH:
+        raise ValueError("Prompt exceeds maximum length of 10000 characters")
+    if get_queue_length(coder) >= MAX_QUEUE_SIZE:
+        raise RuntimeError("Queue is full (max 100 items)")
+
+    index = max(0, min(index, get_queue_length(coder)))
+    with _get_lock(coder):
+        coder._queue_counter += 1
+        item = {
+            "id": str(coder._queue_counter),
+            "text": text,
+            "timestamp": time.time(),
+        }
+        coder.prompt_queue.insert(index, item)
+    return item
+
+
 def dequeue_prompt(coder) -> dict | None:
     """Remove and return the first item from the coder's queue (FIFO).
 
diff --git a/cecli/tests/test_queue_commands.py b/cecli/tests/test_queue_commands.py
new file mode 100644
index 00000000000..e0fc306d898
--- /dev/null
+++ b/cecli/tests/test_queue_commands.py
@@ -0,0 +1,792 @@
+"""
+Test suite for CLI-33 Queue Commands.
+
+This module contains comprehensive tests for:
+- Unit tests: Queue logic in Commands class (core.py)
+- Integration tests: QueueCommand, ListQueueCommand, RemoveQueueCommand
+- E2E tests: Full queue lifecycle and processing
+- Regression tests: Existing command integrity
+
+Test categories:
+- UTC-01 through UTC-20: Unit tests for queue methods
+- ITC-01 through ITC-20: Integration tests for commands
+- ETC-01 through ETC-10: E2E tests for full lifecycle
+- RTC-01 through RTC-05: Regression tests for existing functionality
+- TDS-01 through TDS-04: Test data setup and fixtures
+"""
+
+import asyncio
+import time
+from unittest.mock import MagicMock
+
+import pytest
+
+# Import the actual classes to test
+from cecli.commands.core import Commands
+from cecli.commands.list_queue import ListQueueCommand
+from cecli.commands.queue import QueueCommand
+from cecli.commands.remove_queue import RemoveQueueCommand
+from cecli.commands.utils.registry import CommandRegistry
+from cecli.signals import ReloadProgramSignal, SwitchCoderSignal
+
+
+def _make_coder():
+    """Build a minimal coder-like object with the queue attributes the
+    ``command_queue`` helpers require (``prompt_queue``, ``_queue_counter``,
+    ``_queue_lock``)."""
+    coder = MagicMock()
+    import uuid
+
+    coder.uuid = str(uuid.uuid4())
+    coder.prompt_queue = []
+    coder._queue_counter = 0
+    return coder
+
+
+@pytest.fixture(autouse=True)
+def _reset_agent_service():
+    """Reset the AgentService singleton between tests.
+
+    ``command_queue.get_active_coder`` resolves the foreground coder through
+    ``AgentService``, whose ``_instances`` and ``_primary_agent_uuid`` are
+    class-level state. Without a reset, the first test's coder becomes the
+    "primary" and every later test is routed to that coder's queue, causing
+    cross-test pollution.
+    """
+    from cecli.helpers.agents.service import AgentService
+
+    AgentService._instances = {}
+    AgentService._primary_agent_uuid = None
+    yield
+    AgentService._instances = {}
+    AgentService._primary_agent_uuid = None
+
+
+# ============================================================================
+# Test Fixtures (Section 10.6, TDS-01 through TDS-04)
+# ============================================================================
+
+
+@pytest.fixture
+def mock_io():
+    """Create a mock IO object with tool_output, tool_error, tool_warning methods."""
+    io = MagicMock()
+    io.tool_output = MagicMock()
+    io.tool_error = MagicMock()
+    io.tool_warning = MagicMock()
+    return io
+
+
+@pytest.fixture
+def mock_coder():
+    """Create a mock coder with commands attribute pointing to a Commands instance."""
+    coder = _make_coder()
+    commands = Commands(io=None, coder=coder)
+    coder.commands = commands
+    coder.io = None
+    coder.tui = None
+    return coder
+
+
+@pytest.fixture
+def clean_commands():
+    """Create a fresh Commands instance with empty queue for isolated testing."""
+    return Commands(io=None, coder=_make_coder())
+
+
+@pytest.fixture
+def populated_queue(clean_commands):
+    """Create Commands with pre-populated queue with known items."""
+    clean_commands._enqueue_prompt("alpha")
+    clean_commands._enqueue_prompt("beta")
+    clean_commands._enqueue_prompt("gamma")
+    return clean_commands
+
+
+@pytest.fixture
+def full_queue():
+    """Create Commands with queue filled to max capacity (100 items)."""
+    commands = Commands(io=None, coder=_make_coder())
+    for i in range(100):
+        commands._enqueue_prompt(f"prompt_{i}")
+    return commands
+
+
+@pytest.fixture
+def mock_coder_no_commands():
+    """Create a mock coder with commands set to None."""
+    coder = MagicMock()
+    coder.commands = None
+    coder.io = None
+    coder.tui = None
+    return coder
+
+
+# ============================================================================
+# Unit Tests - Queue Logic (Section 10.2, UTC-01 through UTC-20)
+# ============================================================================
+
+
+class TestEnqueuePrompt:
+    """Unit tests for _enqueue_prompt method."""
+
+    def test_utc_01_enqueue_single_prompt(self, clean_commands):
+        """UTC-01: Enqueue single prompt adds one item with correct structure."""
+        item = clean_commands._enqueue_prompt("test prompt")
+
+        assert len(clean_commands.prompt_queue) == 1
+        assert item["text"] == "test prompt"
+        assert "id" in item
+        assert "timestamp" in item
+        assert isinstance(item["id"], str)
+        assert isinstance(item["timestamp"], float)
+
+    def test_utc_02_enqueue_multiple_prompts_fifo_order(self, clean_commands):
+        """UTC-02: Enqueue multiple prompts maintains FIFO order and unique IDs."""
+        item1 = clean_commands._enqueue_prompt("first")
+        item2 = clean_commands._enqueue_prompt("second")
+        item3 = clean_commands._enqueue_prompt("third")
+
+        assert len(clean_commands.prompt_queue) == 3
+        assert clean_commands.prompt_queue[0]["text"] == "first"
+        assert clean_commands.prompt_queue[1]["text"] == "second"
+        assert clean_commands.prompt_queue[2]["text"] == "third"
+        assert item1["id"] != item2["id"]
+        assert item2["id"] != item3["id"]
+
+    def test_utc_13_max_queue_size_rejection(self, clean_commands):
+        """UTC-13: Enqueue rejected when queue already contains 100 items."""
+        for i in range(100):
+            clean_commands._enqueue_prompt(f"prompt_{i}")
+
+        with pytest.raises(RuntimeError, match="Queue is full"):
+            clean_commands._enqueue_prompt("overflow")
+
+    def test_utc_14_enqueue_empty_string_rejected(self, clean_commands):
+        """UTC-14: Enqueue rejects empty string with ValueError."""
+        with pytest.raises(ValueError, match="Cannot enqueue empty prompt"):
+            clean_commands._enqueue_prompt("")
+
+    def test_utc_14_enqueue_none_rejected(self, clean_commands):
+        """UTC-14: Enqueue rejects None with ValueError."""
+        with pytest.raises(ValueError, match="Cannot enqueue empty prompt"):
+            clean_commands._enqueue_prompt(None)
+
+    def test_utc_15_enqueue_extremely_long_prompt_rejected(self, clean_commands):
+        """UTC-15: Enqueue rejects prompt exceeding 10,000 characters."""
+        long_prompt = "x" * 10001
+        with pytest.raises(ValueError, match="exceeds maximum length"):
+            clean_commands._enqueue_prompt(long_prompt)
+
+    def test_utc_16_enqueue_exactly_10000_chars_accepted(self, clean_commands):
+        """UTC-16: Enqueue accepts prompt of exactly 10,000 characters (boundary)."""
+        boundary_prompt = "x" * 10000
+        item = clean_commands._enqueue_prompt(boundary_prompt)
+        assert item["text"] == boundary_prompt
+        assert len(clean_commands.prompt_queue) == 1
+
+    def test_utc_17_enqueue_9999_chars_accepted(self, clean_commands):
+        """UTC-17: Enqueue accepts prompt of 9,999 characters (boundary)."""
+        boundary_prompt = "x" * 9999
+        item = clean_commands._enqueue_prompt(boundary_prompt)
+        assert item["text"] == boundary_prompt
+        assert len(clean_commands.prompt_queue) == 1
+
+    def test_utc_18_counter_persistence(self, clean_commands):
+        """UTC-18: Internal counter increments across enqueue/remove cycles without reset."""
+        clean_commands._enqueue_prompt("first")
+        clean_commands._enqueue_prompt("second")
+        item = clean_commands._enqueue_prompt("third")
+
+        assert clean_commands._queue_counter == 3
+        assert item["id"] == "3"
+
+
+class TestDequeuePrompt:
+    """Unit tests for _dequeue_prompt method."""
+
+    def test_utc_03_dequeue_from_empty_queue(self, clean_commands):
+        """UTC-03: Dequeue from empty queue returns None without side effects."""
+        result = clean_commands._dequeue_prompt()
+        assert result is None
+        assert len(clean_commands.prompt_queue) == 0
+
+    def test_utc_04_dequeue_returns_fifo_first_item(self, clean_commands):
+        """UTC-04: Dequeue returns first item and shrinks queue by one."""
+        clean_commands._enqueue_prompt("first")
+        clean_commands._enqueue_prompt("second")
+
+        item = clean_commands._dequeue_prompt()
+        assert item["text"] == "first"
+        assert len(clean_commands.prompt_queue) == 1
+        assert clean_commands.prompt_queue[0]["text"] == "second"
+
+    def test_utc_05_dequeue_until_empty(self, clean_commands):
+        """UTC-05: Repeated dequeue eventually returns None after queue empties."""
+        clean_commands._enqueue_prompt("only_item")
+
+        item = clean_commands._dequeue_prompt()
+        assert item is not None
+        assert item["text"] == "only_item"
+
+        result = clean_commands._dequeue_prompt()
+        assert result is None
+
+
+class TestGetQueueLength:
+    """Unit tests for _get_queue_length method."""
+
+    def test_utc_06_queue_length_empty(self, clean_commands):
+        """UTC-06: Queue length returns correct count for empty queue."""
+        assert clean_commands._get_queue_length() == 0
+
+    def test_utc_07_queue_length_non_empty(self, clean_commands):
+        """UTC-07: Queue length returns correct count for populated queue."""
+        clean_commands._enqueue_prompt("item1")
+        clean_commands._enqueue_prompt("item2")
+        clean_commands._enqueue_prompt("item3")
+
+        assert clean_commands._get_queue_length() == 3
+
+
+class TestRemoveFromQueue:
+    """Unit tests for _remove_from_queue method."""
+
+    def test_utc_08_remove_by_valid_index(self, clean_commands):
+        """UTC-08: Remove by valid index returns item and shrinks queue by one."""
+        clean_commands._enqueue_prompt("first")
+        clean_commands._enqueue_prompt("second")
+        clean_commands._enqueue_prompt("third")
+
+        item = clean_commands._remove_from_queue(1)
+        assert item["text"] == "second"
+        assert len(clean_commands.prompt_queue) == 2
+        assert clean_commands.prompt_queue[0]["text"] == "first"
+        assert clean_commands.prompt_queue[1]["text"] == "third"
+
+    def test_utc_09_remove_out_of_bounds_high_index(self, clean_commands):
+        """UTC-09: Remove by out-of-bounds high index returns None without mutation."""
+        clean_commands._enqueue_prompt("only_item")
+
+        result = clean_commands._remove_from_queue(5)
+        assert result is None
+        assert len(clean_commands.prompt_queue) == 1
+
+    def test_utc_10_remove_negative_index(self, clean_commands):
+        """UTC-10: Remove by negative index returns None without mutation."""
+        clean_commands._enqueue_prompt("only_item")
+
+        result = clean_commands._remove_from_queue(-1)
+        assert result is None
+        assert len(clean_commands.prompt_queue) == 1
+
+
+class TestClearQueue:
+    """Unit tests for _clear_queue method."""
+
+    def test_utc_11_clear_queue_with_items(self, clean_commands):
+        """UTC-11: Clear queue returns all items and empties queue."""
+        clean_commands._enqueue_prompt("item1")
+        clean_commands._enqueue_prompt("item2")
+        clean_commands._enqueue_prompt("item3")
+
+        items = clean_commands._clear_queue()
+        assert len(items) == 3
+        assert len(clean_commands.prompt_queue) == 0
+
+    def test_utc_12_clear_empty_queue(self, clean_commands):
+        """UTC-12: Clear empty queue returns empty list and remains empty."""
+        items = clean_commands._clear_queue()
+        assert items == []
+        assert len(clean_commands.prompt_queue) == 0
+
+
+class TestTimestampBehavior:
+    """Unit tests for timestamp generation."""
+
+    def test_utc_19_timestamps_monotonic(self, clean_commands):
+        """UTC-19: Timestamps are monotonic non-decreasing across enqueues."""
+        item1 = clean_commands._enqueue_prompt("first")
+        time.sleep(0.01)
+        item2 = clean_commands._enqueue_prompt("second")
+
+        assert item1["timestamp"] <= item2["timestamp"]
+
+
+# ============================================================================
+# Integration Tests - Command Classes (Section 10.3, ITC-01 through ITC-20)
+# ============================================================================
+
+
+class TestQueueCommand:
+    """Integration tests for QueueCommand."""
+
+    @pytest.mark.asyncio
+    async def test_itc_01_queue_enqueues_and_confirms_position(self, mock_io, mock_coder):
+        """ITC-01: /queue "prompt" enqueues and confirms queue position."""
+        result = await QueueCommand.execute(mock_io, mock_coder, "test prompt")
+
+        assert result == "Successfully executed queue."
+        assert len(mock_coder.commands.prompt_queue) == 1
+        mock_io.tool_output.assert_called()
+
+    @pytest.mark.asyncio
+    async def test_itc_02_queue_empty_args_shows_usage(self, mock_io, mock_coder):
+        """ITC-02: /queue with empty args shows usage/help and does not enqueue."""
+        result = await QueueCommand.execute(mock_io, mock_coder, "")
+
+        assert "Error" in result
+        assert len(mock_coder.commands.prompt_queue) == 0
+
+    @pytest.mark.asyncio
+    async def test_itc_03_queue_no_args_shows_usage(self, mock_io, mock_coder):
+        """ITC-03: /queue with no args shows usage/help and does not enqueue."""
+        result = await QueueCommand.execute(mock_io, mock_coder, None)
+
+        assert "Error" in result
+        assert len(mock_coder.commands.prompt_queue) == 0
+
+    @pytest.mark.asyncio
+    async def test_itc_04_queue_rejects_long_prompt(self, mock_io, mock_coder):
+        """ITC-04: /queue rejects prompt >10,000 characters with warning."""
+        long_prompt = "x" * 10001
+        result = await QueueCommand.execute(mock_io, mock_coder, long_prompt)
+
+        assert "Error" in result or "exceeds" in result.lower()
+        assert len(mock_coder.commands.prompt_queue) == 0
+
+    @pytest.mark.asyncio
+    async def test_itc_05_queue_handles_coder_commands_none(self, mock_io, mock_coder_no_commands):
+        """ITC-05: /queue handles coder.commands is None with error message."""
+        result = await QueueCommand.execute(mock_io, mock_coder_no_commands, "test")
+
+        assert "Error" in result or "not available" in result.lower()
+
+    @pytest.mark.asyncio
+    async def test_itc_06_queue_at_max_capacity_rejects(self, mock_io, full_queue):
+        """ITC-06: /queue at max capacity (100) rejects new prompt."""
+        mock_coder = MagicMock()
+        mock_coder.commands = full_queue
+        mock_coder.io = mock_io
+
+        result = await QueueCommand.execute(mock_io, mock_coder, "overflow")
+
+        assert "Error" in result or "full" in result.lower()
+
+
+class TestListQueueCommand:
+    """Integration tests for ListQueueCommand."""
+
+    @pytest.mark.asyncio
+    async def test_itc_07_list_queue_shows_numbered_list(self, mock_io, populated_queue):
+        """ITC-07: /list-queue displays numbered list of queued prompts with timestamps."""
+        mock_coder = MagicMock()
+        mock_coder.commands = populated_queue
+        mock_coder.io = mock_io
+
+        result = await ListQueueCommand.execute(mock_io, mock_coder, "")
+
+        assert result == "Successfully executed list-queue."
+        mock_io.tool_output.assert_called()
+        calls = [str(call) for call in mock_io.tool_output.call_args_list]
+        output_text = " ".join(calls)
+        assert "[1]" in output_text or "alpha" in output_text
+
+    @pytest.mark.asyncio
+    async def test_itc_08_list_queue_empty_shows_message(self, mock_io, clean_commands):
+        """ITC-08: /list-queue on empty queue shows "Queue is empty" message."""
+        mock_coder = MagicMock()
+        mock_coder.commands = clean_commands
+        mock_coder.io = mock_io
+
+        result = await ListQueueCommand.execute(mock_io, mock_coder, "")
+
+        assert result == "Successfully executed list-queue."
+        mock_io.tool_output.assert_called()
+
+    @pytest.mark.asyncio
+    async def test_itc_09_list_queue_handles_coder_commands_none(
+        self, mock_io, mock_coder_no_commands
+    ):
+        """ITC-09: /list-queue handles coder.commands is None with error message."""
+        result = await ListQueueCommand.execute(mock_io, mock_coder_no_commands, "")
+
+        assert "Error" in result or "not available" in result.lower()
+
+    @pytest.mark.asyncio
+    async def test_itc_10_list_queue_truncates_long_prompts(self, mock_io):
+        """ITC-10: /list-queue truncates prompts longer than display threshold."""
+        commands = Commands(io=None, coder=None)
+        long_prompt = "x" * 120
+        commands._enqueue_prompt(long_prompt)
+
+        mock_coder = MagicMock()
+        mock_coder.commands = commands
+        mock_coder.io = mock_io
+
+        await ListQueueCommand.execute(mock_io, mock_coder, "")
+
+        calls = [str(call) for call in mock_io.tool_output.call_args_list]
+        output_text = " ".join(calls)
+        assert "..." in output_text or "x" * 80 in output_text
+
+
+class TestRemoveQueueCommand:
+    """Integration tests for RemoveQueueCommand."""
+
+    @pytest.mark.asyncio
+    async def test_itc_11_remove_by_index(self, mock_io, populated_queue):
+        """ITC-11: /remove-queue <index> removes exact item and confirms removal."""
+        mock_coder = MagicMock()
+        mock_coder.commands = populated_queue
+        mock_coder.io = mock_io
+
+        result = await RemoveQueueCommand.execute(mock_io, mock_coder, "2")
+
+        assert result == "Successfully executed remove-queue."
+        assert len(populated_queue.prompt_queue) == 2
+        mock_io.tool_output.assert_called()
+
+    @pytest.mark.asyncio
+    async def test_itc_12_remove_wildcard_clears_all(self, mock_io, populated_queue):
+        """ITC-12: /remove-queue * clears entire queue and confirms count removed."""
+        mock_coder = MagicMock()
+        mock_coder.commands = populated_queue
+        mock_coder.io = mock_io
+
+        result = await RemoveQueueCommand.execute(mock_io, mock_coder, "*")
+
+        assert result == "Successfully executed remove-queue."
+        assert len(populated_queue.prompt_queue) == 0
+        mock_io.tool_output.assert_called()
+
+    @pytest.mark.asyncio
+    async def test_itc_13_remove_interactive_mode(self, mock_io, populated_queue):
+        """ITC-13: /remove-queue with no args enters interactive selection."""
+        mock_coder = MagicMock()
+        mock_coder.commands = populated_queue
+        mock_coder.io = mock_io
+
+        result = await RemoveQueueCommand.execute(mock_io, mock_coder, "")
+
+        # Interactive mode shows queue list and prompt, returns success status
+        assert result == "Successfully executed remove-queue."
+        mock_io.tool_output.assert_called()
+        calls = [str(call) for call in mock_io.tool_output.call_args_list]
+        output_text = " ".join(calls)
+        assert "Queued prompts:" in output_text or "Enter index" in output_text
+
+    @pytest.mark.asyncio
+    async def test_itc_14_remove_invalid_index_non_integer(self, mock_io, populated_queue):
+        """ITC-14: /remove-queue with non-integer index shows invalid index error."""
+        mock_coder = MagicMock()
+        mock_coder.commands = populated_queue
+        mock_coder.io = mock_io
+
+        result = await RemoveQueueCommand.execute(mock_io, mock_coder, "abc")
+
+        assert "Error" in result or "Invalid index" in result
+
+    @pytest.mark.asyncio
+    async def test_itc_15_remove_out_of_bounds_index(self, mock_io, populated_queue):
+        """ITC-15: /remove-queue with out-of-bounds index shows error."""
+        mock_coder = MagicMock()
+        mock_coder.commands = populated_queue
+        mock_coder.io = mock_io
+
+        result = await RemoveQueueCommand.execute(mock_io, mock_coder, "99")
+
+        assert "Error" in result or "out of range" in result.lower()
+
+    @pytest.mark.asyncio
+    async def test_itc_16_remove_negative_index(self, mock_io, populated_queue):
+        """ITC-16: /remove-queue with negative index shows error."""
+        mock_coder = MagicMock()
+        mock_coder.commands = populated_queue
+        mock_coder.io = mock_io
+
+        result = await RemoveQueueCommand.execute(mock_io, mock_coder, "-1")
+
+        assert "Error" in result or "Invalid index" in result
+
+    @pytest.mark.asyncio
+    async def test_itc_17_remove_empty_queue(self, mock_io, clean_commands):
+        """ITC-17: /remove-queue on empty queue shows error."""
+        mock_coder = MagicMock()
+        mock_coder.commands = clean_commands
+        mock_coder.io = mock_io
+
+        result = await RemoveQueueCommand.execute(mock_io, mock_coder, "1")
+
+        assert "Error" in result or "empty" in result.lower()
+
+    @pytest.mark.asyncio
+    async def test_itc_18_remove_handles_coder_commands_none(self, mock_io, mock_coder_no_commands):
+        """ITC-18: /remove-queue handles coder.commands is None."""
+        result = await RemoveQueueCommand.execute(mock_io, mock_coder_no_commands, "1")
+
+        assert "Error" in result or "not available" in result.lower()
+
+    def test_itc_19_get_completions_returns_indices_and_wildcard(self, mock_coder, populated_queue):
+        """ITC-19: RemoveQueueCommand.get_completions() returns valid index completions and '*'."""
+        mock_coder.commands = populated_queue
+        completions = RemoveQueueCommand.get_completions(None, mock_coder, "")
+
+        assert "1" in completions
+        assert "2" in completions
+        assert "3" in completions
+        assert "*" in completions
+
+    def test_itc_20_commands_registered_in_registry(self):
+        """ITC-20: All three commands are registered and discoverable via help/registry lookup."""
+        assert CommandRegistry.get_command("queue") is not None
+        assert CommandRegistry.get_command("list-queue") is not None
+        assert CommandRegistry.get_command("remove-queue") is not None
+
+
+# ============================================================================
+# E2E Tests - Full Queue Lifecycle (Section 10.4, ETC-01 through ETC-10)
+# ============================================================================
+
+
+class TestQueueLifecycle:
+    """E2E tests for full queue lifecycle and processing."""
+
+    @pytest.mark.asyncio
+    async def test_etc_01_single_queued_prompt_auto_processes(self, mock_io, populated_queue):
+        """ETC-01: Single queued prompt auto-processes after system becomes idle."""
+        assert hasattr(populated_queue, "_process_queued_prompts")
+        assert callable(populated_queue._process_queued_prompts)
+
+    @pytest.mark.asyncio
+    async def test_etc_02_multiple_prompts_fifo_order(self, clean_commands):
+        """ETC-02: Multiple queued prompts execute in FIFO order with no reordering."""
+        clean_commands._enqueue_prompt("prompt_A")
+        clean_commands._enqueue_prompt("prompt_B")
+        clean_commands._enqueue_prompt("prompt_C")
+
+        assert clean_commands.prompt_queue[0]["text"] == "prompt_A"
+        assert clean_commands.prompt_queue[1]["text"] == "prompt_B"
+        assert clean_commands.prompt_queue[2]["text"] == "prompt_C"
+
+    @pytest.mark.asyncio
+    async def test_etc_03_queued_prompt_not_processed_while_running(self, mock_io, populated_queue):
+        """ETC-03: Queued prompt is not processed while another command is running."""
+        populated_queue.cmd_running_event.clear()
+
+        assert hasattr(populated_queue, "_MANAGEMENT_COMMANDS")
+        assert "queue" in populated_queue._MANAGEMENT_COMMANDS
+
+    @pytest.mark.asyncio
+    async def test_etc_06_management_commands_dont_trigger_processing(
+        self, mock_io, populated_queue
+    ):
+        """ETC-06: Management commands do not trigger auto-processing of queued items."""
+        assert populated_queue._MANAGEMENT_COMMANDS == {"queue", "list-queue", "remove-queue"}
+
+    @pytest.mark.asyncio
+    async def test_etc_05_prevent_infinite_loop(self, clean_commands):
+        """ETC-05: Queued command that queues additional items does not cause infinite loop."""
+        assert hasattr(clean_commands, "_processing_queue")
+        assert clean_commands._processing_queue is False
+
+    @pytest.mark.asyncio
+    async def test_etc_09_error_in_queued_prompt_continues(self, clean_commands):
+        """ETC-09: Exception in queued prompt is logged but doesn't stop later items."""
+        assert hasattr(clean_commands, "_process_queued_prompts")
+
+    @pytest.mark.asyncio
+    async def test_etc_10_full_lifecycle_sequence(self, mock_io, populated_queue):
+        """ETC-10: Full lifecycle sequence add -> list -> remove -> process."""
+        item = populated_queue._enqueue_prompt("new_prompt")
+        assert item is not None
+
+        assert populated_queue._get_queue_length() == 4
+
+        removed = populated_queue._remove_from_queue(0)
+        assert removed is not None
+
+        assert populated_queue._get_queue_length() == 3
+
+
+# ============================================================================
+# Regression Tests - Existing Command Integrity (Section 10.5, RTC-01 through RTC-05)
+# ============================================================================
+
+
+class TestRegression:
+    """Regression tests to ensure existing functionality is not broken."""
+
+    def test_rtc_01_existing_commands_still_registered(self):
+        """RTC-01: Existing commands still registered and functional after queue commands added."""
+        assert CommandRegistry.get_command("help") is not None
+        assert CommandRegistry.get_command("run") is not None
+        assert CommandRegistry.get_command("model") is not None
+
+    def test_rtc_02_commands_init_preserves_existing_attributes(self, clean_commands):
+        """RTC-02: Commands.__init__ preserves pre-existing attributes and adds new queue fields."""
+        assert hasattr(clean_commands, "io")
+        assert hasattr(clean_commands, "coder")
+        assert hasattr(clean_commands, "cmd_running_event")
+        assert hasattr(clean_commands, "last_command_show_notification")
+
+        assert hasattr(clean_commands, "prompt_queue")
+        assert hasattr(clean_commands, "_queue_counter")
+        assert hasattr(clean_commands, "_queue_lock")
+        assert hasattr(clean_commands, "_processing_queue")
+        assert hasattr(clean_commands, "_MANAGEMENT_COMMANDS")
+
+    def test_rtc_03_execute_preserves_existing_flow(self, mock_io, mock_coder):
+        """RTC-03: Commands.execute() preserves existing command behavior for non-queue commands."""
+        assert hasattr(mock_coder.commands, "execute")
+        assert callable(mock_coder.commands.execute)
+
+    def test_rtc_04_init_py_imports_all_commands(self):
+        """RTC-04: cecli/commands/__init__.py imports all commands without conflicts."""
+        from cecli.commands import (
+            CommandRegistry,
+            ListQueueCommand,
+            QueueCommand,
+            RemoveQueueCommand,
+        )
+
+        assert CommandRegistry.get_command("queue") is QueueCommand
+        assert CommandRegistry.get_command("list-queue") is ListQueueCommand
+        assert CommandRegistry.get_command("remove-queue") is RemoveQueueCommand
+
+    def test_rtc_05_thread_safety_under_concurrent_access(self, clean_commands):
+        """RTC-05: Simulated concurrent access patterns do not corrupt queue state."""
+        assert hasattr(clean_commands, "_queue_lock")
+        assert isinstance(clean_commands._queue_lock, asyncio.Lock)
+
+
+# ============================================================================
+# Additional Tests for Edge Cases and Error Handling
+# ============================================================================
+
+
+class TestEdgeCases:
+    """Additional edge case tests."""
+
+    def test_queue_with_whitespace_only_prompt(self, clean_commands):
+        """Test that whitespace-only prompts are rejected."""
+        with pytest.raises(ValueError):
+            clean_commands._enqueue_prompt("   ")
+
+    def test_queue_with_unicode_prompt(self, clean_commands):
+        """Test that unicode prompts are handled correctly."""
+        item = clean_commands._enqueue_prompt("Hello world")
+        assert item["text"] == "Hello world"
+
+    def test_remove_with_zero_index(self, clean_commands):
+        """Test removing with index 0 (first item)."""
+        clean_commands._enqueue_prompt("first")
+        clean_commands._enqueue_prompt("second")
+
+        item = clean_commands._remove_from_queue(0)
+        assert item["text"] == "first"
+
+    def test_remove_with_large_index(self, clean_commands):
+        """Test removing with a very large index."""
+        clean_commands._enqueue_prompt("only")
+
+        result = clean_commands._remove_from_queue(999999)
+        assert result is None
+
+
+# ============================================================================
+# Command Registration Tests
+# ============================================================================
+
+
+class TestCommandRegistration:
+    """Tests for command registration in CommandRegistry."""
+
+    def test_all_queue_commands_registered(self):
+        """Verify all queue commands are properly registered."""
+        commands = CommandRegistry.list_commands()
+
+        assert "queue" in commands
+        assert "list-queue" in commands
+        assert "remove-queue" in commands
+
+    def test_command_classes_have_required_attributes(self):
+        """Verify command classes have NORM_NAME and DESCRIPTION."""
+        assert QueueCommand.NORM_NAME == "queue"
+        assert QueueCommand.DESCRIPTION is not None
+
+        assert ListQueueCommand.NORM_NAME == "list-queue"
+        assert ListQueueCommand.DESCRIPTION is not None
+
+        assert RemoveQueueCommand.NORM_NAME == "remove-queue"
+        assert RemoveQueueCommand.DESCRIPTION is not None
+
+    def test_command_classes_have_execute_method(self):
+        """Verify command classes have async execute method."""
+        assert hasattr(QueueCommand, "execute")
+        assert hasattr(ListQueueCommand, "execute")
+        assert hasattr(RemoveQueueCommand, "execute")
+
+        assert callable(QueueCommand.execute)
+        assert callable(ListQueueCommand.execute)
+        assert callable(RemoveQueueCommand.execute)
+
+
+# ============================================================================
+# Helper Function Tests
+# ============================================================================
+
+
+class TestHelpers:
+    """Tests for helper functions used by queue commands."""
+
+    def test_format_command_result_success(self, mock_io):
+        """Test format_command_result for successful execution."""
+        from cecli.commands.utils.helpers import format_command_result
+
+        result = format_command_result(mock_io, "test", "Success message")
+        assert result == "Successfully executed test."
+        mock_io.tool_output.assert_called_once()
+
+    def test_format_command_result_error(self, mock_io):
+        """Test format_command_result for error case."""
+        from cecli.commands.utils.helpers import format_command_result
+
+        result = format_command_result(mock_io, "test", "Success", error="Something went wrong")
+        assert "Error" in result
+        mock_io.tool_error.assert_called_once()
+
+
+# ============================================================================
+# Signal Tests
+# ============================================================================
+
+
+class TestSignals:
+    """Tests for custom signals used in queue processing."""
+
+    def test_switch_coder_signal_attributes(self):
+        """Test SwitchCoderSignal has expected attributes."""
+        signal = SwitchCoderSignal(placeholder="test", custom_arg="value")
+
+        assert signal.placeholder == "test"
+        assert signal.kwargs == {"custom_arg": "value"}
+
+    def test_reload_program_signal_message(self):
+        """Test ReloadProgramSignal has message attribute."""
+        signal = ReloadProgramSignal(message="Custom message")
+
+        assert signal.message == "Custom message"
+
+
+# ============================================================================
+# Async Test Configuration
+# ============================================================================
+
+# Note: For pytest-asyncio to work, you may need to add to pyproject.toml or pytest.ini:
+# [pytest]
+# asyncio_mode = auto
+# asyncio_default_fixture_loop_scope = function
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-v"])
diff --git a/cecli/tui/app.py b/cecli/tui/app.py
index f33a40930f5..00feaa6f71d 100644
--- a/cecli/tui/app.py
+++ b/cecli/tui/app.py
@@ -894,15 +894,16 @@ def on_input_area_submit(self, message: InputArea.Submit):
                 self._handle_workspace_command(user_input, stripped)
                 return
 
-        # Intercept queue management commands (/queue, /list-queue, /remove-queue)
-        # to dispatch immediately without a full generation cycle - they only
-        # modify the active coder's prompt_queue.
+        # Intercept queue management commands (/queue, /list-queue, /remove-queue,
+        # /insert-queue) to dispatch immediately without a full generation cycle -
+        # they only modify the active coder's prompt_queue.
         if (
             stripped == "/queue"
             or stripped.startswith("/queue ")
             or stripped == "/list-queue"
             or stripped == "/remove-queue"
             or stripped.startswith("/remove-queue ")
+            or stripped.startswith("/insert-queue ")
         ):
             self._handle_queue_command(stripped)
             return
@@ -1019,6 +1020,20 @@ def _handle_queue_command(self, stripped: str) -> None:
                 f"Prompt queued at position {position} (id: {item['id']})"
             )
 
+            # Process the queued prompt immediately when the coder is idle (no
+            # active output task). The TUI dispatches queue commands directly
+            # here instead of through run_one(), so without this the queued
+            # prompt would only run after the user submits yet another prompt.
+            worker_loop = getattr(self.worker, "loop", None)
+            if (
+                worker_loop is not None
+                and not is_active(getattr(active_coder.io, "output_task", None))
+                and not getattr(active_coder, "_processing_queue", False)
+            ):
+                worker_loop.call_soon_threadsafe(
+                    lambda: worker_loop.create_task(active_coder._drain_prompt_queue(True))
+                )
+
         elif cmd == "/list-queue":
             items = command_queue.list_queue(active_coder)
             if not items:
@@ -1034,6 +1049,41 @@ def _handle_queue_command(self, stripped: str) -> None:
 
             self._get_visible_container().add_output("\n".join(lines))
 
+        elif cmd == "/insert-queue":
+            parts = args.split(maxsplit=1)
+            if len(parts) != 2:
+                self.show_error("Usage: /insert-queue <index> <prompt text>")
+                return
+
+            try:
+                index = int(parts[0])
+            except ValueError:
+                self.show_error(f"Invalid index: '{parts[0]}'. Please provide a number.")
+                return
+
+            prompt_text = parts[1].strip()
+            try:
+                item = command_queue.insert_prompt(active_coder, prompt_text, index)
+            except (ValueError, RuntimeError) as e:
+                self.show_error(str(e))
+                return
+
+            self._get_visible_container().add_output(
+                f"Prompt inserted at position {index + 1} (id: {item['id']})"
+            )
+
+            # Process the queued prompt immediately when the coder is idle,
+            # same as /queue.
+            worker_loop = getattr(self.worker, "loop", None)
+            if (
+                worker_loop is not None
+                and not is_active(getattr(active_coder.io, "output_task", None))
+                and not getattr(active_coder, "_processing_queue", False)
+            ):
+                worker_loop.call_soon_threadsafe(
+                    lambda: worker_loop.create_task(active_coder._drain_prompt_queue(True))
+                )
+
         elif cmd == "/remove-queue":
             if not args:
                 self.show_error("Usage: /remove-queue <index|*>")
diff --git a/cecli/website/docs/config.md b/cecli/website/docs/config.md
index 19a106e8a79..644161bcb2c 100644
--- a/cecli/website/docs/config.md
+++ b/cecli/website/docs/config.md
@@ -66,3 +66,15 @@ A directory containing custom tool packages (`.py` files exposing a `Tool` class
 
 > **Tip:**
 > See the [API key configuration docs](config/api-keys.html) for information on how to configure and store your API keys.
+
+## Prompt Queue Configuration
+
+- `max_queue_size`: Maximum number of prompts that can be queued in a single session (default: 100, range: 1-1000). Environment variable: `CECLI_MAX_QUEUE_SIZE`.
+- `max_prompt_length`: Maximum length of a single queued prompt (default: 10,000, range: 100-50,000). Environment variable: `CECLI_MAX_PROMPT_LENGTH`.
+
+## Prompt Queue Configuration
+
+- `max_queue_size`: Maximum number of prompts that can be queued in a single session (default: 100, range: 1-1000). Environment variable: `CECLI_MAX_QUEUE_SIZE`.
+- `max_prompt_length`: Maximum length of a single queued prompt (default: 10,000, range: 100-50,000). Environment variable: `CECLI_MAX_PROMPT_LENGTH`.
+
+{% include keys.md %}
diff --git a/cecli/website/docs/troubleshooting.md b/cecli/website/docs/troubleshooting.md
index 627f9c58f7a..02c63d97210 100644
--- a/cecli/website/docs/troubleshooting.md
+++ b/cecli/website/docs/troubleshooting.md
@@ -28,3 +28,19 @@ If the problem involves an LLM request, `--debug` may also create request logs u
 
 > **Tip:**
 > Use `/help <question>` to [ask for help about using cecli](troubleshooting/support.html), customizing settings, using LLMs, etc.
+
+## Queue Commands
+
+- **Queue not processing**: If queued prompts don't execute, ensure the system is idle (no active command running). Use `/list-queue` to verify prompts are in the queue.
+- **Prompt not queued**: If `/queue` fails, check that the prompt is not empty, does not exceed 10,000 characters, and the queue is not full (100 items).
+- **Cannot remove from queue**: Ensure you are using a valid positive integer index. Use `/list-queue` to verify current queue contents and valid indices.
+- **Queue seems corrupted**: The queue is in-memory and session-specific. Restarting the CLI session will clear the queue.
+
+## Queue Commands
+
+- **Queue not processing**: Ensure the system is idle (no active command running). Use `/list-queue` to verify prompts are in the queue.
+- **Prompt not queued**: Check if the prompt is empty, exceeds the 10,000 character limit, or if the queue is full (100 items).
+- **Cannot remove from queue**: Ensure the index is a valid positive integer. Use `/list-queue` to verify current queue contents and valid indices.
+- **Queue seems corrupted**: Restart the `cecli` session to clear the in-memory queue.
+
+{% include help.md %}
diff --git a/cecli/website/docs/usage/commands.md b/cecli/website/docs/usage/commands.md
index 9650275ebac..ccd2823b9b3 100644
--- a/cecli/website/docs/usage/commands.md
+++ b/cecli/website/docs/usage/commands.md
@@ -59,6 +59,138 @@ Cecli supports commands from within the chat, which all start with `/`.
 
 > **Tip:** You can easily re-send commands or messages. Use the up arrow ⬆ to scroll back or CONTROL-R to search your message history.
 
+## Prompt Queue Management
+
+| Command | Description |
+| :--- | :--- |
+| **/queue** | Queue a prompt for processing after current tasks complete |
+| **/list-queue** | List all prompts currently in the queue |
+| **/remove-queue** | Remove a prompt from the queue by index, or '*' to clear all |
+
+{: .tip }
+
+## Prompt Queue Management Commands
+
+The prompt queue management feature (`CLI-33`) adds three new commands for managing a first-in-first-out (FIFO) queue of prompts.
+
+### Queue Commands
+
+| Command | Description |
+|---------|-------------|
+| **/queue** | Queue a prompt for processing after current tasks complete |
+| **/insert-queue** | Insert a prompt at a specific position in the queue |
+| **/list-queue** | List all prompts currently in the queue |
+| **/remove-queue** | Remove a prompt from the queue by index, or '*' to clear all |
+
+#### `/queue` Command
+
+**Usage:** `/queue <prompt text>`
+
+**Description:** Adds a prompt to the queue for processing after the current command completes.
+
+**Arguments:**
+- `prompt text`: Required. The prompt text to queue (maximum 10,000 characters)
+
+**Returns:** Confirmation message with the queue position number
+
+**Examples:**
+```bash
+/queue "refactor database layer"
+/queue "add unit tests for user service"
+```
+
+**Implementation Details:**
+- `NORM_NAME = "queue"`
+- `DESCRIPTION = "Queue a prompt for processing after current tasks complete"`
+- `execute()`: Validates input, calls `coder.commands._enqueue_prompt()`, returns position confirmation
+- `get_help()`: Returns usage and examples
+
+#### `/list-queue` Command
+
+**Usage:** `/list-queue`
+
+**Description:** Displays all prompts currently in the queue with their position numbers and timestamps.
+
+**Arguments:** None
+
+**Returns:** Numbered list of queued prompts (`[index] text (timestamp)`) or "Queue is empty" message
+
+**Examples:**
+```bash
+/list-queue
+# Output: [1] refactor database layer (2026-08-01 10:30:00)
+#         [2] add unit tests for user service (2026-08-01 10:30:05)
+```
+
+**Implementation Details:**
+- `NORM_NAME = "list-queue"`
+- `DESCRIPTION = "List all prompts currently in the queue"`
+- `execute()`: Accesses queue, formats output with timestamps and truncated text, handles empty queue
+- `get_help()`: Returns usage and examples
+
+#### `/insert-queue` Command
+
+**Usage:** `/insert-queue <prompt text>`, `/insert-queue <index> <prompt text>`
+
+**Description:** Inserts a prompt at a specific position in the queue. When called without an index, the prompt is added at the front of the queue.
+
+**Arguments:**
+- `index`: Optional. Position at which to insert the prompt
+- `prompt text`: Required. The prompt text to insert
+
+**Returns:** Confirmation message with the queue position number
+
+**Examples:**
+```bash
+/insert-queue "add tests for login"
+/insert-queue 3 "refactor database layer"
+```
+
+**Implementation Details:**
+- `NORM_NAME = "insert-queue"`
+- `DESCRIPTION = "Insert a prompt at a specific position in the queue"`
+- `execute()`: Validates input, calls `coder.commands._insert_prompt()`, returns position confirmation
+- `get_help()`: Returns usage and examples
+
+#### `/remove-queue` Command
+
+**Usage:** `/remove-queue <index>`, `/remove-queue *`, or `/remove-queue` (interactive)
+
+**Description:** Removes a specific prompt from the queue by index, clears the entire queue with `*`, or provides interactive selection when called with no arguments.
+
+**Arguments:**
+- `index`: Optional. 0-based index of the prompt to remove, or `*` wildcard to clear all
+
+**Returns:** Confirmation of removal and updated queue state
+
+**Examples:**
+```bash
+/remove-queue 2          # Remove prompt at index 2
+/remove-queue *          # Clear entire queue
+/remove-queue            # Interactive selection mode
+```
+
+**Implementation Details:**
+- `NORM_NAME = "remove-queue"`
+- `DESCRIPTION = "Remove a prompt from the queue by index, or '*' to clear all"`
+- `execute()`: Handles `*` wildcard, numbered index, and interactive mode
+- `get_help()`: Returns usage and examples
+- `get_completions()`: Returns index numbers + `*` for tab completion
+
+#### Error Handling
+
+All queue commands follow consistent error handling patterns:
+- `ValueError`: Raised for empty prompts or None values in `/queue`
+- `IndexError`: Raised for out-of-bounds indices in `/remove-queue`
+- Usage errors: Non-integer indices, invalid arguments show user-friendly messages
+- Null checks: Handle `coder.commands` is None gracefully with error messages
+
+#### Thread Safety
+
+The queue uses an `asyncio.Lock` (`_queue_lock`) to protect all read and write operations, ensuring atomic updates in the single-threaded async event loop.
+You can easily re-send commands or messages.
+Use the up arrow ⬆ to scroll back
+or CONTROL-R to search your message history.
 
 ## Non-TUI Related Notes
 
diff --git a/cecli/website/docs/usage/prompt-queue.md b/cecli/website/docs/usage/prompt-queue.md
new file mode 100644
index 00000000000..6a918191b35
--- /dev/null
+++ b/cecli/website/docs/usage/prompt-queue.md
@@ -0,0 +1,339 @@
+---
+nav_order: 55
+parent: Usage
+description: Developer documentation for the prompt queue management feature
+---
+
+# Prompt Queue Management (Developer Guide)
+
+This document provides comprehensive developer documentation for the prompt queue management feature (`CLI-33`). The feature allows users to queue prompts for deferred processing, view the queue, and selectively remove items.
+
+## Architecture Overview
+
+### Queue Location and Data Structure
+
+The prompt queue is implemented as an instance variable on the `Commands` class in `cecli/commands/core.py`:
+
+```python
+# In Commands.__init__()
+self.prompt_queue = []  # List[Dict[str, Union[str, float]]]
+self._queue_counter = 0
+self._queue_lock = asyncio.Lock()
+self._processing_queue = False
+
+# Commands that should NOT trigger auto-processing of the queue
+self._MANAGEMENT_COMMANDS = {"queue", "list-queue", "remove-queue"}
+```
+
+Each queue item is a dictionary with the following structure:
+
+```python
+{
+    "id": str,           # Unique identifier (incrementing counter)
+    "text": str,         # The prompt text
+    "timestamp": float   # Unix timestamp when enqueued
+}
+```
+
+### Lifecycle
+
+- **Session-bound**: The queue is tied to the user's CLI session and does not persist across restarts
+- **In-memory only**: Stored as a Python list on the `Commands` instance
+- **FIFO ordering**: Prompts are processed in first-in-first-out order
+- **Auto-processing**: Triggered after the current command completes and the system is idle
+
+### Thread Safety
+
+The implementation uses a single-threaded async event loop architecture:
+
+- **`asyncio.Lock` (`_queue_lock`)**: Protects all read and write operations on `prompt_queue`
+- **Lock acquisition pattern**: `async with self._queue_lock:` for all queue modifications
+- **CPython GIL + async model**: Makes list operations naturally safe within the async loop
+- **No multi-threading**: The lock is a precaution for future concurrent access patterns
+
+## Commands Class Queue Management Methods
+
+### `_enqueue_prompt(self, text: str) -> dict`
+
+Adds a prompt to the end of the queue.
+
+**Parameters:**
+- `text`: The prompt text to enqueue
+
+**Returns:**
+- `dict` with keys: `id` (str), `text` (str), `timestamp` (float)
+
+**Raises:**
+- `ValueError`: If text is empty, None, or exceeds 10,000 characters
+- `RuntimeError`: If the queue is at max capacity (100 items)
+
+**Implementation:**
+```python
+async with self._queue_lock:
+    if not text or not text.strip():
+        raise ValueError("Cannot enqueue empty prompt")
+    if len(text) > 10000:
+        raise ValueError("Prompt exceeds maximum length of 10000 characters")
+    if len(self.prompt_queue) >= 100:
+        raise RuntimeError("Queue is full (max 100 items)")
+
+    self._queue_counter += 1
+    item = {
+        "id": str(self._queue_counter),
+        "text": text,
+        "timestamp": time.time(),
+    }
+    self.prompt_queue.append(item)
+    return item
+```
+
+### `_dequeue_prompt(self) -> dict | None`
+
+Removes and returns the first item from the queue (FIFO).
+
+**Returns:**
+- The dequeued item dict, or `None` if the queue is empty
+
+### `_get_queue_length(self) -> int`
+
+Returns the current number of items in the queue.
+
+**Returns:**
+- `int`: Current queue size
+
+### `_remove_from_queue(self, index: int) -> dict | None`
+
+Removes and returns the item at the given 0-based index.
+
+**Parameters:**
+- `index`: 0-based index of the item to remove
+
+**Returns:**
+- The removed item dict, or `None` if the index is out of bounds
+
+### `_clear_queue(self) -> list`
+
+Removes all items from the queue and returns them.
+
+**Returns:**
+- List of all items that were in the queue
+
+### `_process_queued_prompts(self)`
+
+Internal method that processes all prompts currently in the queue sequentially. Called from the `finally` block of `Commands.execute()` after `cmd_running_event.set()`.
+
+**Processing Logic:**
+1. Sets `self._processing_queue = True` (guard against re-entrant processing)
+2. While queue is non-empty:
+   a. Dequeues next item
+   b. Logs: `"Processing queued prompt (id: {id})..."`
+   c. Calls `await self.run(item["text"])`
+   d. Catches `SwitchCoderSignal` / `ReloadProgramSignal` → re-raises
+   e. Catches generic `Exception` → logs error, continues to next item
+3. Sets `self._processing_queue = False`
+
+## Queue Processing Integration
+
+### Integration Point
+
+The queue processing is triggered in the `finally` block of `Commands.execute()`:
+
+```python
+finally:
+    self.cmd_running_event.set()  # System is now idle
+    if self.coder.tui and self.coder.tui():
+        self.coder.tui().refresh()
+    # Queue processing integration
+    if (
+        self.prompt_queue
+        and cmd_name not in self._MANAGEMENT_COMMANDS
+        and not self._processing_queue
+    ):
+        await self._process_queued_prompts()
+```
+
+### Guard Conditions
+
+Queue processing only occurs when ALL of the following are true:
+1. `self.prompt_queue` is non-empty
+2. The command that just completed is NOT a management command (`queue`, `list-queue`, `remove-queue`)
+3. Not already processing the queue (`_processing_queue` flag is False)
+
+### Management Command Non-Interference
+
+Management commands (`/queue`, `/list-queue`, `/remove-queue`) are designed to execute immediately without interrupting ongoing prompt processing:
+
+- They do NOT clear `cmd_running_event`
+- They do NOT trigger auto-processing of queued items
+- Their execution is isolated so the current prompt continues uninterrupted
+- In `Commands.run()`, management commands starting with `/` are intercepted and executed immediately via `self.execute()`
+
+### Error Handling
+
+- **Signal propagation**: `SwitchCoderSignal` and `ReloadProgramSignal` from queued prompts are re-raised (not swallowed)
+- **Generic exceptions**: Caught, logged via `io.tool_error()`, and processing continues to the next item
+- **One bad prompt doesn't block the rest**: Error resilience is built into the processing loop
+
+## Command Registration Pattern
+
+Each queue command is implemented in a separate file following the `BaseCommand` pattern:
+
+### File Structure
+
+```
+cecli/commands/
+├── queue.py           # QueueCommand
+├── list_queue.py      # ListQueueCommand
+├── remove_queue.py    # RemoveQueueCommand
+└── __init__.py        # Registration
+```
+
+### Import Pattern (in `cecli/commands/__init__.py`)
+
+```python
+from .queue import QueueCommand
+from .list_queue import ListQueueCommand
+from .remove_queue import RemoveQueueCommand
+```
+
+### Registration
+
+```python
+CommandRegistry.register(QueueCommand)
+CommandRegistry.register(ListQueueCommand)
+CommandRegistry.register(RemoveQueueCommand)
+```
+
+### Module Exports (`__all__`)
+
+```python
+__all__ = [
+    # ... other commands ...
+    "QueueCommand",
+    "ListQueueCommand",
+    "RemoveQueueCommand",
+]
+```
+
+## BaseCommand Implementation for Queue Commands
+
+All three commands follow the `BaseCommand` interface:
+
+### Required Attributes
+
+- `NORM_NAME`: Normalized command name (e.g., `"queue"`, `"list-queue"`, `"remove-queue"`)
+- `DESCRIPTION`: Human-readable description for help output
+
+### Required Methods
+
+- `async execute(cls, io, coder, args, **kwargs)`: Main command logic
+- `get_help(cls) -> str`: Returns usage and examples
+- `get_completions(cls, io, coder, args) -> List[str]`: Tab completion (only `RemoveQueueCommand`)
+
+### QueueCommand
+
+```python
+class QueueCommand(BaseCommand):
+    NORM_NAME = "queue"
+    DESCRIPTION = "Queue a prompt for processing after current tasks complete"
+    
+    async def execute(cls, io, coder, args, **kwargs):
+        # Validates args, calls coder.commands._enqueue_prompt()
+        # Returns confirmation with queue position
+    
+    def get_help(cls) -> str:
+        # Returns usage and examples
+```
+
+### ListQueueCommand
+
+```python
+class ListQueueCommand(BaseCommand):
+    NORM_NAME = "list-queue"
+    DESCRIPTION = "List all prompts currently in the queue"
+    
+    async def execute(cls, io, coder, args, **kwargs):
+        # Accesses queue, displays numbered list, handles empty
+    
+    def get_help(cls) -> str:
+        # Returns usage and examples
+```
+
+### RemoveQueueCommand
+
+```python
+class RemoveQueueCommand(BaseCommand):
+    NORM_NAME = "remove-queue"
+    DESCRIPTION = "Remove a prompt from the queue by index, or '*' to clear all"
+    
+    async def execute(cls, io, coder, args, **kwargs):
+        # Handles '*' wildcard, numbered index, interactive mode
+    
+    def get_completions(cls, io, coder, args) -> List[str]:
+        # Returns index numbers + wildcard based on queue length
+    
+    def get_help(cls) -> str:
+        # Returns usage and examples
+```
+
+## Error Handling Patterns
+
+### ValueError
+
+Raised for:
+- Empty prompts or None values in `/queue`
+- Prompts exceeding 10,000 character limit
+
+### IndexError
+
+Raised for:
+- Out-of-bounds indices in `/remove-queue`
+
+### Usage Errors
+
+- Non-integer indices show user-friendly messages
+- Invalid arguments show usage/help
+
+### Null Checks
+
+All commands handle `coder.commands is None` gracefully with error messages instead of crashing.
+
+## Queue Limits
+
+| Limit | Value | Behavior |
+|-------|-------|----------|
+| Max Queue Size | 100 items | Rejects new prompts with warning when full |
+| Max Prompt Length | 10,000 characters | Rejects prompts exceeding this limit |
+| In-Memory Only | Session-bound | Lost on CLI restart |
+
+## Configuration (Future)
+
+The following configuration options are planned but not yet implemented:
+
+- `--max-queue-size` / `max_queue_size` (default: 100, range: 1-1000)
+- `--max-prompt-length` / `max_prompt_length` (default: 10000, range: 100-50000)
+- `--no-queue-auto-process` to disable auto-processing
+- `--queue-verbose` for verbose queue logging
+- Environment variables: `CECLI_MAX_QUEUE_SIZE`, `CECLI_MAX_PROMPT_LENGTH`
+
+## Testing
+
+See `cecli/tests/test_queue_commands.py` for:
+- Unit tests for queue logic in `core.py`
+- Integration tests for command classes
+- E2E tests for full queue lifecycle
+- Regression tests for existing command integrity
+- Test fixtures and data builders
+
+## Related Files
+
+- `cecli/commands/core.py` - Queue data structure and processing logic
+- `cecli/commands/queue.py` - `/queue` command implementation
+- `cecli/commands/list_queue.py` - `/list-queue` command implementation
+- `cecli/commands/remove_queue.py` - `/remove-queue` command implementation
+- `cecli/commands/__init__.py` - Command registration
+- `cecli/commands/utils/base_command.py` - BaseCommand interface
+- `cecli/tests/test_queue_commands.py` - Test suite
+- `cecli/website/docs/usage/commands.md` - User-facing command reference
+- `cecli/website/docs/troubleshooting.md` - Troubleshooting guide
+- `CHANGELOG.md` - Release notes
\ No newline at end of file
diff --git a/requirements.txt b/requirements.txt
index 56d75692d3a..b4a57cc1874 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -594,7 +594,7 @@ uvicorn[standard]==0.38.0
     #   -c requirements/common-constraints.txt
     #   chromadb
     #   mcp
-uvloop==0.22.1
+uvloop==0.22.1 ; platform_python_implementation != 'PyPy' and sys_platform != 'cygwin' and sys_platform != 'win32'
     # via
     #   -c requirements/common-constraints.txt
     #   uvicorn
diff --git a/tests/basic/test_commands.py b/tests/basic/test_commands.py
index cd9fedb374e..c367cafefd1 100644
--- a/tests/basic/test_commands.py
+++ b/tests/basic/test_commands.py
@@ -136,9 +136,8 @@ def test_command_paths_none_does_not_warn(self):
         args = SimpleNamespace(command_paths=None)
 
         with mock.patch.object(io, "tool_warning") as tool_warning:
-            commands = Commands(io, coder=None, args=args)
+            Commands(io, coder=None, args=args)
 
-        self.assertEqual(commands.custom_commands, [])
         tool_warning.assert_not_called()
 
     async def test_cmd_copy_pyperclip_exception(self):