From 4bf8efc4a200eb54947a7fa83e47e4d459a36294 Mon Sep 17 00:00:00 2001 From: Your Name Date: Sun, 11 Oct 2026 14:42:12 -0700 Subject: [PATCH 1/6] test: add run-queue/idle-immediate queue tests, fix RTC-05 lock check --- cecli/tests/test_queue_commands.py | 181 ++++++++++++++++++++++++++++- 1 file changed, 178 insertions(+), 3 deletions(-) diff --git a/cecli/tests/test_queue_commands.py b/cecli/tests/test_queue_commands.py index 2449c55e35c..524d728315f 100644 --- a/cecli/tests/test_queue_commands.py +++ b/cecli/tests/test_queue_commands.py @@ -23,9 +23,11 @@ # Import the actual classes to test from cecli.commands.core import Commands +from cecli.commands.insert_queue import InsertQueueCommand from cecli.commands.list_queue import ListQueueCommand from cecli.commands.queue import QueueCommand from cecli.commands.remove_queue import RemoveQueueCommand +from cecli.commands.run_queue import RunQueueCommand from cecli.commands.utils.registry import CommandRegistry from cecli.helpers import command_queue from cecli.signals import ReloadProgramSignal, SwitchCoderSignal @@ -503,10 +505,173 @@ def test_itc_19_get_completions_returns_indices_and_wildcard(self, populated_que 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.""" + """ITC-20: All four queue commands are registered and discoverable via 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 + assert CommandRegistry.get_command("run-queue") is not None + + +class TestRunQueueCommand: + """Integration tests for RunQueueCommand.""" + + @pytest.mark.asyncio + async def test_itc_21_run_queue_drains_multiple_items_fifo(self, mock_io): + """ITC-21: /run-queue drains every queued item in FIFO order.""" + coder = _make_coder_with_commands(("first", "second", "third")) + processed = [] + + async def fake_drain(): + while coder.prompt_queue: + item = command_queue.dequeue_prompt(coder) + processed.append(item["text"]) + + coder.process_prompt_queue = fake_drain + + result = await RunQueueCommand.execute(mock_io, coder, "") + + assert result == "Successfully executed run-queue." + assert processed == ["first", "second", "third"] + assert command_queue.get_queue_length(coder) == 0 + + @pytest.mark.asyncio + async def test_itc_22_run_queue_exception_returns_formatted_error(self, mock_io): + """ITC-22: An exception in process_prompt_queue returns a formatted error.""" + coder = _make_coder_with_commands(("boom",)) + + async def failing_drain(): + raise RuntimeError("drain exploded") + + coder.process_prompt_queue = failing_drain + + result = await RunQueueCommand.execute(mock_io, coder, "") + + assert "Error" in result + assert "drain exploded" in result + mock_io.tool_error.assert_called_once() + + @pytest.mark.asyncio + async def test_itc_23_run_queue_empty_queue_informs_user(self, mock_io, mock_coder): + """ITC-23: /run-queue on an empty queue informs the user without error.""" + result = await RunQueueCommand.execute(mock_io, mock_coder, "") + + assert result == "Successfully executed run-queue." + calls = [str(call) for call in mock_io.tool_output.call_args_list] + assert any("Queue is empty" in call for call in calls) + + @pytest.mark.asyncio + async def test_itc_24_run_queue_handles_coder_commands_none( + self, mock_io, mock_coder_no_commands + ): + """ITC-24: /run-queue handles coder.commands is None with error message.""" + result = await RunQueueCommand.execute(mock_io, mock_coder_no_commands, "") + + assert "Error" in result or "not available" in result.lower() + + +class TestQueueIdleImmediate: + """Integration tests for idle-immediate queue draining (plan step 9.8).""" + + @pytest.mark.asyncio + async def test_itc_25_queue_idle_triggers_immediate_processing(self, mock_io): + """ITC-25: /queue while idle (run_one_completed=True) drains the queue now.""" + coder = _make_coder_with_commands() + coder.run_one_completed = True + coder._processing_queue = False + processed = [] + + async def fake_drain(): + while coder.prompt_queue: + item = command_queue.dequeue_prompt(coder) + processed.append(item["text"]) + + coder.process_prompt_queue = fake_drain + + result = await QueueCommand.execute(mock_io, coder, "idle prompt") + + assert result == "Successfully executed queue." + assert processed == ["idle prompt"] + assert command_queue.get_queue_length(coder) == 0 + + @pytest.mark.asyncio + async def test_itc_26_queue_busy_does_not_trigger_processing(self, mock_io): + """ITC-26: /queue while a request is in flight defers to the auto-drain.""" + coder = _make_coder_with_commands() + coder.run_one_completed = False + coder._processing_queue = False + drain_calls = [] + + async def fake_drain(): + drain_calls.append(True) + + coder.process_prompt_queue = fake_drain + + result = await QueueCommand.execute(mock_io, coder, "busy prompt") + + assert result == "Successfully executed queue." + assert drain_calls == [] + assert command_queue.get_queue_length(coder) == 1 + assert coder.prompt_queue[0]["text"] == "busy prompt" + + @pytest.mark.asyncio + async def test_itc_27_queue_guard_prevents_double_drain(self, mock_io): + """ITC-27: _processing_queue=True suppresses the idle drain (re-entrancy guard).""" + coder = _make_coder_with_commands() + coder.run_one_completed = True + coder._processing_queue = True + drain_calls = [] + + async def fake_drain(): + drain_calls.append(True) + + coder.process_prompt_queue = fake_drain + + result = await QueueCommand.execute(mock_io, coder, "guarded prompt") + + assert result == "Successfully executed queue." + assert drain_calls == [] + assert command_queue.get_queue_length(coder) == 1 + + @pytest.mark.asyncio + async def test_itc_28_insert_queue_idle_triggers_immediate_processing(self, mock_io): + """ITC-28: /insert-queue while idle drains the queue immediately.""" + coder = _make_coder_with_commands() + coder.run_one_completed = True + coder._processing_queue = False + processed = [] + + async def fake_drain(): + while coder.prompt_queue: + item = command_queue.dequeue_prompt(coder) + processed.append(item["text"]) + + coder.process_prompt_queue = fake_drain + + result = await InsertQueueCommand.execute(mock_io, coder, "1 idle insert") + + assert result == "Successfully executed insert-queue." + assert processed == ["idle insert"] + assert command_queue.get_queue_length(coder) == 0 + + @pytest.mark.asyncio + async def test_itc_29_insert_queue_busy_does_not_trigger_processing(self, mock_io): + """ITC-29: /insert-queue while a request is in flight defers to auto-drain.""" + coder = _make_coder_with_commands() + coder.run_one_completed = False + coder._processing_queue = False + drain_calls = [] + + async def fake_drain(): + drain_calls.append(True) + + coder.process_prompt_queue = fake_drain + + result = await InsertQueueCommand.execute(mock_io, coder, "1 busy insert") + + assert result == "Successfully executed insert-queue." + assert drain_calls == [] + assert command_queue.get_queue_length(coder) == 1 + assert coder.prompt_queue[0]["text"] == "busy insert" # ============================================================================ @@ -550,7 +715,12 @@ async def test_etc_06_management_commands_dont_trigger_processing( self, mock_io, clean_commands ): """ETC-06: Management commands do not trigger auto-processing of queued items.""" - assert clean_commands._MANAGEMENT_COMMANDS == {"queue", "list-queue", "remove-queue"} + assert clean_commands._MANAGEMENT_COMMANDS == { + "queue", + "list-queue", + "remove-queue", + "run-queue", + } @pytest.mark.asyncio async def test_etc_05_prevent_infinite_loop(self, populated_queue): @@ -639,7 +809,7 @@ def test_rtc_05_thread_safety_under_concurrent_access(self, clean_commands): coder = clean_commands.coder command_queue.enqueue_prompt(coder, "item") - assert isinstance(coder._queue_lock, threading.Lock) + assert isinstance(coder._queue_lock, type(threading.Lock())) # ============================================================================ @@ -691,6 +861,7 @@ def test_all_queue_commands_registered(self): assert "queue" in commands assert "list-queue" in commands assert "remove-queue" in commands + assert "run-queue" in commands def test_command_classes_have_required_attributes(self): """Verify command classes have NORM_NAME and DESCRIPTION.""" @@ -703,6 +874,9 @@ def test_command_classes_have_required_attributes(self): assert RemoveQueueCommand.NORM_NAME == "remove-queue" assert RemoveQueueCommand.DESCRIPTION is not None + assert RunQueueCommand.NORM_NAME == "run-queue" + assert RunQueueCommand.DESCRIPTION is not None + def test_command_classes_have_execute_method(self): """Verify command classes have async execute method.""" assert hasattr(QueueCommand, "execute") @@ -712,6 +886,7 @@ def test_command_classes_have_execute_method(self): assert callable(QueueCommand.execute) assert callable(ListQueueCommand.execute) assert callable(RemoveQueueCommand.execute) + assert callable(RunQueueCommand.execute) # ============================================================================ From fb04f197de7818dff865f0ed896145f7937835a6 Mon Sep 17 00:00:00 2001 From: Your Name Date: Mon, 12 Oct 2026 06:30:50 -0700 Subject: [PATCH 2/6] fix: drain idle prompt queue in run_one command path and run_stream --- cecli/coders/base_coder.py | 64 ++++++++++++++++++++++++------ cecli/tests/test_queue_commands.py | 48 +++++++++++++++++++++- 2 files changed, 99 insertions(+), 13 deletions(-) diff --git a/cecli/coders/base_coder.py b/cecli/coders/base_coder.py index 126781384fd..dba693474ff 100755 --- a/cecli/coders/base_coder.py +++ b/cecli/coders/base_coder.py @@ -1615,6 +1615,10 @@ async def run_stream(self, user_message): async for chunk in self.send_message(user_message): yield chunk + # Stream-driven interfaces bypass run_one(), so drain the queue here + # to keep queued prompts processing after the request completes. + await self.process_prompt_queue() + def init_before_message(self): self.coder_edited_files = set() self.reflected_message = None @@ -2014,6 +2018,47 @@ async def preproc_user_input(self, inp): def wrap_user_input(self, inp): return inp + async def process_prompt_queue(self, preproc=True): + """Drain the coder's prompt queue in FIFO order. + + Re-entrancy safe: a nested call (e.g. a queued prompt that itself + triggers a drain) returns immediately. Each item is isolated so one + failing prompt does not abort the rest of the queue. An interrupt + (Ctrl+C / cancellation) re-queues the in-flight item at the front and + re-raises so the caller can resume later via /run-queue. + """ + if self._processing_queue: + return + + self._processing_queue = True + try: + while self.prompt_queue: + item = command_queue.dequeue_prompt(self) + if item is None: + break + + self.io.tool_output(f"Processing queued prompt (id: {item['id']})...") + try: + await self.run_one(item["text"], preproc) + except (asyncio.CancelledError, KeyboardInterrupt): + # Put the in-flight prompt back at the front so it is not + # lost, then let the interrupt propagate to the caller. + command_queue.insert_prompt(self, item["text"], 0) + remaining = command_queue.get_queue_length(self) + self.io.tool_warning( + f"Queue processing interrupted. {remaining} prompt(s) remain" + " queued; use /run-queue to resume." + ) + raise + except (SwitchCoderSignal, SystemExit): + raise + except Exception as e: + # Isolate failures: log and continue with the next item. + self.io.tool_error(f"Error processing queued prompt (id: {item['id']}): {e}") + continue + finally: + self._processing_queue = False + async def run_one(self, user_message, preproc): self.init_before_message() @@ -2029,6 +2074,10 @@ async def run_one(self, user_message, preproc): if self.commands.is_command(user_message) and not self.commands.is_test_command( user_message ): + # Command path: may have queued prompts (e.g. /queue); drain so + # idle-queued prompts start immediately. The _processing_queue + # guard prevents double-drain when this runs inside a drain loop. + await self.process_prompt_queue(preproc) return if not self.commands.is_command(user_message): @@ -2100,19 +2149,10 @@ async def run_one(self, user_message, preproc): await self.auto_save_session(force=True) - # Move to the next queued prompt (CLI-33) only after the current message - # has fully completed, so the queue drains within run_one() instead of + # Drain any queued prompts (CLI-33) only after the current message has + # fully completed, so the queue is processed within run_one() instead of # being watched by the generation loops. - if self.prompt_queue and not self._processing_queue: - self._processing_queue = True - try: - item = command_queue.dequeue_prompt(self) - finally: - self._processing_queue = False - - if item is not None: - self.io.tool_output(f"Processing queued prompt (id: {item['id']})...") - await self.run_one(item["text"], preproc) + await self.process_prompt_queue(preproc) if not await HookIntegration.call_end_hooks(self): self.io.tool_warning("Execution stopped by end hook") diff --git a/cecli/tests/test_queue_commands.py b/cecli/tests/test_queue_commands.py index 524d728315f..f19108e1448 100644 --- a/cecli/tests/test_queue_commands.py +++ b/cecli/tests/test_queue_commands.py @@ -11,7 +11,7 @@ - UTC-01 through UTC-20: Unit tests for queue helpers - 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 +- RTC-01 through RTC-06: Regression tests for existing functionality - TDS-01 through TDS-04: Test data setup and fixtures """ @@ -674,6 +674,51 @@ async def fake_drain(): assert coder.prompt_queue[0]["text"] == "busy insert" +class TestRunOneCommandPathDrain: + """Regression tests for the run_one() command-path drain (plan step 9.9).""" + + @pytest.mark.asyncio + async def test_itc_30_run_one_command_path_drains_idle_queue(self, mock_io): + """ITC-30: A user-typed /queue while idle drains via run_one()'s command + early return even though run_one_completed is False while the command + executes inside generate() (so the command-side idle drain is skipped). + """ + from cecli.coders.base_coder import Coder + + coder = _make_coder_with_commands() + coder.io = mock_io + # Bug scenario: the typed /queue runs inside generate(), which has + # already set run_one_completed=False, so the command-side idle drain + # in queue.py is skipped; only run_one()'s early return can drain. + coder.run_one_completed = False + coder._processing_queue = False + + calls = [] + processed = [] + + async def fake_preproc(inp): + calls.append("preproc") + return await coder.commands.run(inp, coder=coder) + + async def fake_drain(preproc=True): + calls.append("drain") + while coder.prompt_queue: + item = command_queue.dequeue_prompt(coder) + processed.append(item["text"]) + + coder.preproc_user_input = fake_preproc + coder.process_prompt_queue = fake_drain + + await Coder.run_one(coder, "/queue tell me the color of the sun", preproc=True) + + # The command ran first, then exactly one drain fired from run_one()'s + # command early return. A command-side idle drain would have added a + # second "drain" entry while still inside "preproc". + assert calls == ["preproc", "drain"] + assert processed == ["tell me the color of the sun"] + assert command_queue.get_queue_length(coder) == 0 + + # ============================================================================ # E2E Tests - Full Queue Lifecycle (Section 10.4, ETC-01 through ETC-10) # ============================================================================ @@ -812,6 +857,7 @@ def test_rtc_05_thread_safety_under_concurrent_access(self, clean_commands): assert isinstance(coder._queue_lock, type(threading.Lock())) + # ============================================================================ # Additional Tests for Edge Cases and Error Handling # ============================================================================ From e750db0490af409d6c35cafb40cf84a117648294 Mon Sep 17 00:00:00 2001 From: Your Name Date: Thu, 15 Oct 2026 16:32:35 -0700 Subject: [PATCH 3/6] feat: drain idle queue in TUI and fix nested command completions --- cecli/commands/core.py | 8 +- cecli/commands/insert_queue.py | 34 ++- cecli/commands/queue.py | 27 ++- cecli/io.py | 15 +- cecli/tests/test_queue_commands.py | 332 +++++++++++++++++++++++++++++ cecli/tui/app.py | 94 +++++++- 6 files changed, 501 insertions(+), 9 deletions(-) diff --git a/cecli/commands/core.py b/cecli/commands/core.py index 312a1802625..e2df42af02e 100644 --- a/cecli/commands/core.py +++ b/cecli/commands/core.py @@ -14,6 +14,12 @@ class Commands: scraper = None + # Queue commands need the FULL text after the command name (including the + # trailing space) passed as completion args, so nested slash-command + # completion (e.g. "/queue /add ") can see the inner command and its + # arguments (plan 9.13). Both completers (io.py, tui/app.py) consult this. + FULL_ARGS_COMPLETION_COMMANDS = ("/queue", "/insert-queue") + def _get_coder(self): """Return coder via weak reference, or None if collected.""" if self._coder_ref is not None: @@ -90,7 +96,7 @@ def __init__( self.last_command_show_notification = True # Commands that should NOT trigger auto-processing of the queue - self._MANAGEMENT_COMMANDS = {"queue", "list-queue", "remove-queue"} + self._MANAGEMENT_COMMANDS = {"queue", "list-queue", "remove-queue", "run-queue"} def _load_custom_commands(self, custom_commands): """ diff --git a/cecli/commands/insert_queue.py b/cecli/commands/insert_queue.py index f4cc1922c9f..78d8951dc07 100644 --- a/cecli/commands/insert_queue.py +++ b/cecli/commands/insert_queue.py @@ -1,10 +1,12 @@ """Insert-queue command for CLI-33: inserts a prompt at a specific queue position.""" +import asyncio from typing import List from cecli.commands.utils.base_command import BaseCommand from cecli.commands.utils.helpers import format_command_result from cecli.helpers import command_queue +from cecli.signals import SwitchCoderSignal class InsertQueueCommand(BaseCommand): @@ -60,6 +62,22 @@ async def execute(cls, io, coder, args, **kwargs): try: item = command_queue.insert_prompt(coder, prompt_text, index) io.tool_output(f"Prompt inserted at position {index + 1} (id: {item['id']})") + + # Idle-immediate: when no request is in flight there is nothing to + # wait for, so drain the queue right away. When a request IS in + # flight, the queue auto-drains when run_one() completes. + target = command_queue.get_active_coder(coder) + if command_queue.idle_drain_needed(target): + try: + await target.process_prompt_queue() + except (KeyboardInterrupt, asyncio.CancelledError): + # The interrupted item was re-queued; let the user know the queue paused. + io.tool_output("Queue processing paused. Use /run-queue to resume.") + except (SwitchCoderSignal, SystemExit): + raise + except Exception as e: + return format_command_result(io, cls.NORM_NAME, "", error=str(e)) + return f"Successfully executed {cls.NORM_NAME}." except ValueError as e: return format_command_result(io, cls.NORM_NAME, "", error=str(e)) @@ -68,8 +86,20 @@ async def execute(cls, io, coder, args, **kwargs): @classmethod def get_completions(cls, io, coder, args) -> List[str]: - """Get completion options for insert-queue command.""" - return [] + """Get completion options for insert-queue command. + + Args are " "; once the index token and a space + are present, mirror the nested prompt's completions (e.g. a nested + slash command delegates to the original command's completions). + """ + text = args or "" + first_space = text.find(" ") + if first_space == -1: + # Still typing the index; suggest valid positions. + if not coder.commands: + return [] + return [str(i) for i in range(1, command_queue.get_queue_length(coder) + 2)] + return command_queue.nested_command_completions(coder, text[first_space + 1 :]) @classmethod def get_help(cls) -> str: diff --git a/cecli/commands/queue.py b/cecli/commands/queue.py index 9c4a858377c..f936d89e93b 100644 --- a/cecli/commands/queue.py +++ b/cecli/commands/queue.py @@ -2,9 +2,12 @@ from typing import List +import asyncio + from cecli.commands.utils.base_command import BaseCommand from cecli.commands.utils.helpers import format_command_result from cecli.helpers import command_queue +from cecli.signals import SwitchCoderSignal class QueueCommand(BaseCommand): @@ -59,6 +62,22 @@ async def execute(cls, io, coder, args, **kwargs): item = command_queue.enqueue_prompt(coder, prompt_text) position = command_queue.get_queue_length(coder) io.tool_output(f"Prompt queued at position {position} (id: {item['id']})") + + # Idle-immediate: when no request is in flight there is nothing to + # wait for, so drain the queue right away. When a request IS in + # flight, the queue auto-drains when run_one() completes. + target = command_queue.get_active_coder(coder) + if command_queue.idle_drain_needed(target): + try: + await target.process_prompt_queue() + except (KeyboardInterrupt, asyncio.CancelledError): + # The interrupted item was re-queued; let the user know the queue paused. + io.tool_output("Queue processing paused. Use /run-queue to resume.") + except (SwitchCoderSignal, SystemExit): + raise + except Exception as e: + return format_command_result(io, cls.NORM_NAME, "", error=str(e)) + return f"Successfully executed {cls.NORM_NAME}." except ValueError as e: return format_command_result(io, cls.NORM_NAME, "", error=str(e)) @@ -67,8 +86,12 @@ async def execute(cls, io, coder, args, **kwargs): @classmethod def get_completions(cls, io, coder, args) -> List[str]: - """Get completion options for queue command.""" - return [] + """Get completion options for queue command. + + When the queued prompt starts with a slash command, mirror the + original command's completions (e.g. /queue /add offers files). + """ + return command_queue.nested_command_completions(coder, args) @classmethod def get_help(cls) -> str: diff --git a/cecli/io.py b/cecli/io.py index a3bf286315f..2c61b669ae7 100644 --- a/cecli/io.py +++ b/cecli/io.py @@ -265,7 +265,20 @@ def get_command_completions(self, document, complete_event, text, words): yield from raw_completer(document, complete_event) return - candidates = self.commands.get_completions(cmd, partial) + # Nested-host commands (/queue, /insert-queue) need the FULL text after + # the command name (trailing space preserved) so inner slash-command + # argument completion works (e.g. "/queue /add ") - plan 9.13. + if cmd in self.commands.FULL_ARGS_COMPLETION_COMMANDS: + nested_text = text[len(words[0]) :].lstrip() + candidates = self.commands.get_completions(cmd, nested_text) + # The substring filter and start_position below operate on the last + # word; an empty/trailing-space nested text means "no partial yet". + if not nested_text or nested_text[-1].isspace(): + partial = "" + else: + partial = nested_text.split()[-1].lower() + else: + candidates = self.commands.get_completions(cmd, partial) if candidates is None: return diff --git a/cecli/tests/test_queue_commands.py b/cecli/tests/test_queue_commands.py index f19108e1448..f5ff2eae3b6 100644 --- a/cecli/tests/test_queue_commands.py +++ b/cecli/tests/test_queue_commands.py @@ -719,6 +719,92 @@ async def fake_drain(preproc=True): assert command_queue.get_queue_length(coder) == 0 +# ============================================================================ +# Nested Completions Tests (Section 10.6, plan step 9.11) +# ============================================================================ + + +class TestNestedCompletions: + """Unit tests for nested slash-command completions in /queue and /insert-queue.""" + + def test_ntc_01_queue_prefix_matches_registry_commands(self, mock_io): + """NTC-01: /queue /sa offers inner commands whose names start with "sa".""" + coder = _make_coder_with_commands() + + result = QueueCommand.get_completions(mock_io, coder, "/sa") + + assert "/save-session" in result + assert all(name.startswith("/sa") for name in result) + assert result == sorted(result) + + def test_ntc_02_queue_full_inner_command_delegates_to_original(self, mock_io): + """NTC-02: /queue /add delegates to the original command via + Commands.get_completions so file names are offered like /add .""" + coder = _make_coder_with_commands() + file_completions = ["file_a.py", "file_b.py"] + coder.commands.get_completions = MagicMock(return_value=file_completions) + + result = QueueCommand.get_completions(mock_io, coder, "/add ") + + assert result == file_completions + coder.commands.get_completions.assert_called_once_with("/add", "", coder=coder) + + def test_ntc_03_queue_save_session_delegates_session_names(self, mock_io): + """NTC-03: /queue /save-session delegates to SaveSessionCommand so + existing session names are offered exactly like /save-session .""" + coder = _make_coder_with_commands() + session_names = ["auto-save", "feature-x"] + coder.commands.get_completions = MagicMock(return_value=session_names) + + result = QueueCommand.get_completions(mock_io, coder, "/save-session ") + + assert result == session_names + coder.commands.get_completions.assert_called_once_with( + "/save-session", "", coder=coder + ) + + def test_ntc_04_queue_plain_text_returns_empty(self, mock_io): + """NTC-04: non-command queued text (and empty args) have no completions.""" + coder = _make_coder_with_commands() + + assert QueueCommand.get_completions(mock_io, coder, "plain text") == [] + assert QueueCommand.get_completions(mock_io, coder, "") == [] + + def test_ntc_05_queue_unknown_inner_command_returns_empty(self, mock_io): + """NTC-05: delegation errors (e.g. unknown inner command) degrade to [].""" + coder = _make_coder_with_commands() + + def boom(cmd, args="", coder=None): + raise RuntimeError("unknown command") + + coder.commands.get_completions = boom + + assert QueueCommand.get_completions(mock_io, coder, "/no-such-cmd arg ") == [] + + def test_ntc_06_insert_queue_delegates_after_index(self, mock_io): + """NTC-06: /insert-queue 1 /save-session delegates after the index.""" + coder = _make_coder_with_commands() + session_names = ["auto-save"] + coder.commands.get_completions = MagicMock(return_value=session_names) + + result = InsertQueueCommand.get_completions(mock_io, coder, "1 /save-session ") + + assert result == session_names + coder.commands.get_completions.assert_called_once_with( + "/save-session", "", coder=coder + ) + + def test_ntc_07_insert_queue_index_only_suggests_positions(self, mock_io): + """NTC-07: while typing the index, /insert-queue offers clamped positions.""" + coder = _make_coder_with_commands() + command_queue.enqueue_prompt(coder, "only item") + + result = InsertQueueCommand.get_completions(mock_io, coder, "1") + + # Queue holds 1 item, so positions 1..2 (append at end) are valid. + assert result == ["1", "2"] + + # ============================================================================ # E2E Tests - Full Queue Lifecycle (Section 10.4, ETC-01 through ETC-10) # ============================================================================ @@ -982,6 +1068,252 @@ def test_reload_program_signal_message(self): assert signal.message == "Custom message" +# ============================================================================ +# Idle-drain gate and TUI scheduling (plan 10.8) +# ============================================================================ + + +class TestIdleDrainGate: + """Tests for command_queue.idle_drain_needed and TUI drain scheduling.""" + + def test_idg_01_idle_drain_needed_true_when_idle(self): + """IDG-01: idle when run_one_completed=True and _processing_queue=False.""" + coder = _make_coder_with_commands() + coder.run_one_completed = True + coder._processing_queue = False + + assert command_queue.idle_drain_needed(coder) is True + + def test_idg_02_idle_drain_needed_false_when_busy(self): + """IDG-02: run_one_completed=False blocks the idle drain.""" + coder = _make_coder_with_commands() + coder.run_one_completed = False + coder._processing_queue = False + + assert command_queue.idle_drain_needed(coder) is False + + def test_idg_03_idle_drain_needed_false_when_draining(self): + """IDG-03: _processing_queue=True blocks the idle drain (re-entrancy).""" + coder = _make_coder_with_commands() + coder.run_one_completed = True + coder._processing_queue = True + + assert command_queue.idle_drain_needed(coder) is False + + @pytest.mark.asyncio + async def test_idg_04_scheduled_drain_processes_idle_queue(self): + """IDG-04: scheduling the drain for an idle coder drains its queue. + + Exercises the TUI scheduling helper against a fake worker whose + ``loop`` is the running asyncio loop (the real TUI schedules the same + call on the worker loop). + """ + import asyncio + + from cecli.tui.app import TUI + + coder = _make_coder_with_commands(items=("queued prompt",)) + coder.run_one_completed = True + coder._processing_queue = False + processed = [] + + async def fake_drain(): + while coder.prompt_queue: + processed.append(command_queue.dequeue_prompt(coder)["text"]) + + coder.process_prompt_queue = fake_drain + + app = TUI.__new__(TUI) + fake_worker = MagicMock() + fake_worker.loop = asyncio.get_running_loop() + app.worker = fake_worker + + TUI._schedule_idle_queue_drain(app, coder) + # Let the scheduled callback run (1 tick) and the drain task finish + # (a second tick) before asserting. + await asyncio.sleep(0) + await asyncio.sleep(0) + + assert processed == ["queued prompt"] + assert command_queue.get_queue_length(coder) == 0 + + @pytest.mark.asyncio + async def test_idg_05_scheduled_drain_skips_when_busy(self): + """IDG-05: a busy coder is not drained by the scheduler.""" + import asyncio + + from cecli.tui.app import TUI + + coder = _make_coder_with_commands(items=("queued prompt",)) + coder.run_one_completed = False + drain_calls = [] + + async def fake_drain(): + drain_calls.append(True) + + coder.process_prompt_queue = fake_drain + + app = TUI.__new__(TUI) + fake_worker = MagicMock() + fake_worker.loop = asyncio.get_running_loop() + app.worker = fake_worker + + TUI._schedule_idle_queue_drain(app, coder) + await asyncio.sleep(0) + + assert drain_calls == [] + assert command_queue.get_queue_length(coder) == 1 + + +# ============================================================================ +# Completer plumbing - full-args semantics (plan 10.9) +# ============================================================================ + + +class TestCompleterPlumbing: + """Plan 10.9: full-args completion plumbing through Commands.get_completions. + + These assert the completer-level call path (``/queue`` receives the FULL + text after the command name), not just the inner command class (which is + already covered by TestNestedCompletions). + """ + + def test_tcp_01_queue_add_trailing_space_delegates_empty_inner_args(self, monkeypatch): + """TCP-01: /queue /add reaches AddCommand with "" (trailing space).""" + from cecli.commands.add import AddCommand + + coder = _make_coder_with_commands() + seen = [] + + def spy(cls, io, coder_, args): + seen.append(args) + return ["file_a.py", "file_b.py"] + + monkeypatch.setattr(AddCommand, "get_completions", classmethod(spy)) + + result = coder.commands.get_completions("/queue", args="/add ") + + assert result == ["file_a.py", "file_b.py"] + assert seen == [""] + + def test_tcp_02_queue_add_partial_word_delegates_partial(self, monkeypatch): + """TCP-02: /queue /add s reaches AddCommand with "s".""" + from cecli.commands.add import AddCommand + + coder = _make_coder_with_commands() + seen = [] + + def spy(cls, io, coder_, args): + seen.append(args) + return ["src/"] + + monkeypatch.setattr(AddCommand, "get_completions", classmethod(spy)) + + result = coder.commands.get_completions("/queue", args="/add s") + + assert result == ["src/"] + assert seen == ["s"] + + def test_tcp_03_queue_inner_command_names(self): + """TCP-03: /queue /sa surfaces matching inner command names.""" + coder = _make_coder_with_commands() + + result = coder.commands.get_completions("/queue", args="/sa") + + assert "/save-session" in result + assert all(name.startswith("/sa") for name in result) + + def test_tcp_04_queue_save_session_delegates_session_names(self, monkeypatch): + """TCP-04: /queue /save-session pre reaches SaveSessionCommand with "pre".""" + from cecli.commands.save_session import SaveSessionCommand + + coder = _make_coder_with_commands() + seen = [] + + def spy(cls, io, coder_, args): + seen.append(args) + return ["presentation"] + + monkeypatch.setattr(SaveSessionCommand, "get_completions", classmethod(spy)) + + result = coder.commands.get_completions("/queue", args="/save-session pre") + + assert result == ["presentation"] + assert seen == ["pre"] + + def test_tcp_05_queue_plain_text_returns_empty(self): + """TCP-05: /queue plain text has no completions.""" + coder = _make_coder_with_commands() + + assert coder.commands.get_completions("/queue", args="plain text") == [] + + def test_tcp_06_insert_queue_delegates_after_index(self, monkeypatch): + """TCP-06: /insert-queue 1 /add delegates to AddCommand after index.""" + from cecli.commands.add import AddCommand + + coder = _make_coder_with_commands() + seen = [] + + def spy(cls, io, coder_, args): + seen.append(args) + return ["file_a.py"] + + monkeypatch.setattr(AddCommand, "get_completions", classmethod(spy)) + + result = coder.commands.get_completions("/insert-queue", args="1 /add ") + + assert result == ["file_a.py"] + assert seen == [""] + + def test_tcp_07_io_partial_derivation(self): + """TCP-07: io.get_command_completions derives the last-word partial from + the FULL nested text ("" after a trailing space, "s" for "/add s") and + forwards the full nested text (trailing space preserved) to the command. + """ + from cecli.io import AutoCompleter + + completer = AutoCompleter.__new__(AutoCompleter) + captured = [] + commands = MagicMock() + commands.FULL_ARGS_COMPLETION_COMMANDS = ("/queue", "/insert-queue") + # Mirror Commands.matching_commands: match the input's first word. + commands.matching_commands = lambda text: ( + ["/queue"] if text.startswith("/queue") else (["/add"] if text.startswith("/add") else None), + text.split()[0], + "", + ) + commands.get_raw_completions = lambda cmd: None + + def fake_get_completions(cmd, args="", coder=None): + captured.append((cmd, args)) + return ["file_a.py"] + + commands.get_completions = fake_get_completions + completer.commands = commands + + # Trailing space: nested text keeps the space; partial is empty. + results = list( + completer.get_command_completions(None, None, "/queue /add ", ["/queue", "/add"]) + ) + assert captured[-1] == ("/queue", "/add ") + assert all(c.start_position == 0 for c in results) + + # Partial word: nested text is forwarded verbatim; partial is the word. + results = list( + completer.get_command_completions( + None, None, "/queue /add s", ["/queue", "/add", "s"] + ) + ) + assert captured[-1] == ("/queue", "/add s") + assert all(c.start_position == -1 for c in results) + + # Non-nested commands keep the legacy last-word partial. + commands.FULL_ARGS_COMPLETION_COMMANDS = () + captured.clear() + list(completer.get_command_completions(None, None, "/add s", ["/add", "s"])) + assert captured[-1] == ("/add", "s") + + # ============================================================================ # Async Test Configuration # ============================================================================ diff --git a/cecli/tui/app.py b/cecli/tui/app.py index d5ca44c35a8..7ed46ff9b4b 100644 --- a/cecli/tui/app.py +++ b/cecli/tui/app.py @@ -1,7 +1,9 @@ """Main Textual application for cecli TUI.""" +import asyncio import concurrent.futures import json +import logging import platform import queue import time @@ -41,6 +43,8 @@ if platform.system() == "Windows": IS_WINDOWS = True +logger = logging.getLogger(__name__) + class TUI(App): """Main Textual application for cecli TUI.""" @@ -1059,6 +1063,7 @@ def _handle_queue_command(self, stripped: str) -> None: self._get_visible_container().add_output( f"Prompt queued at position {position} (id: {item['id']})" ) + self._schedule_idle_queue_drain(active_coder) elif cmd == "/list-queue": items = command_queue.list_queue(active_coder) @@ -1097,6 +1102,7 @@ def _handle_queue_command(self, stripped: str) -> None: self._get_visible_container().add_output( f"Prompt inserted at position {index + 1} (id: {item['id']})" ) + self._schedule_idle_queue_drain(active_coder) elif cmd == "/remove-queue": if not args: @@ -1125,6 +1131,79 @@ def _handle_queue_command(self, stripped: str) -> None: f"Removed queued prompt {index}: {removed['text'][:80]}" ) + _queue_drain_task = None + _queue_drain_lock = None + + def _schedule_idle_queue_drain(self, active_coder) -> None: + """Schedule immediate queue processing on the worker loop when idle. + + The TUI runs on the textual thread, so coder coroutines must never be + awaited here; work is scheduled on the worker's event loop instead. + + The first queued item is dequeued and processed through + ``coder.generate()`` — the same lifecycle as a user-typed prompt + (it sets ``run_one_completed``, clears the interrupt event, and its + trailing auto-drain in run_one() processes any remaining items). + Calling ``run_one()``/``process_prompt_queue()`` directly here would + run outside that lifecycle and stall silently (live bug 2026-10-15). + + When a request is in flight or a drain is already running, do + nothing — the post-run_one auto-drain covers the queue. + """ + from cecli.helpers import command_queue + from cecli.signals import SwitchCoderSignal + + loop = self.worker.loop + if loop is None or loop.is_closed(): + return + if not command_queue.idle_drain_needed(active_coder): + return + + async def _drain(): + # Serialize: a second scheduled drain must never run generate() + # concurrently with the first one. The lock is created lazily on + # the worker loop; the None check cannot interleave because the + # worker loop runs this task step atomically up to the acquire. + if self._queue_drain_lock is None: + self._queue_drain_lock = asyncio.Lock() + async with self._queue_drain_lock: + # Re-check just before draining: a generation cycle may have + # started between scheduling and execution. + if not command_queue.idle_drain_needed(active_coder): + return + item = command_queue.dequeue_prompt(active_coder) + if item is None: + return + try: + await active_coder.generate(item["text"], True) + except (asyncio.CancelledError, KeyboardInterrupt): + # The interrupted prompt is re-queued so nothing is lost; + # tell the user how to resume. Swallow the interrupt here + # so a bare task ending on KeyboardInterrupt cannot escape + # into the worker loop and crash it. + command_queue.insert_prompt(active_coder, item["text"], 0) + remaining = command_queue.get_queue_length(active_coder) + active_coder.io.tool_warning( + f"Queue processing interrupted. {remaining} prompt(s) remain" + " queued; use /run-queue to resume." + ) + except (SwitchCoderSignal, SystemExit): + raise + except Exception as e: + # Match process_prompt_queue per-item isolation: surface + # the failure to the user instead of dying silently. + active_coder.io.tool_warning( + f"Error processing queued prompt (id: {item['id']}): {e}" + ) + logger.warning("Idle queue drain failed", exc_info=True) + + def _start(): + # Keep a strong reference so the asyncio task is not garbage + # collected before it completes. + self._queue_drain_task = loop.create_task(_drain()) + + loop.call_soon_threadsafe(_start) + def _handle_spawn_agent_command(self, user_input: str, stripped: str) -> None: """Dispatch /spawn-agent immediately without a full generation cycle. @@ -1938,9 +2017,18 @@ def _get_suggestions(self, text: str) -> list[str]: else: # Use standard command completions (no file fallback) try: - cmd_completions = commands.get_completions( - cmd_name, args=arg_prefix, coder=active_coder - ) + if cmd_name in commands.FULL_ARGS_COMPLETION_COMMANDS: + # Nested-host commands receive the FULL text after + # the command name so inner slash-command argument + # completion works (e.g. "/queue /add "). + nested_text = text[len(cmd_name) :].lstrip() + cmd_completions = commands.get_completions( + cmd_name, args=nested_text, coder=active_coder + ) + else: + cmd_completions = commands.get_completions( + cmd_name, args=arg_prefix, coder=active_coder + ) if cmd_completions: exempt_from_substring_matching = { "/model", From accf42f945fd0077c97bdf93c2e2e33ac5f87705 Mon Sep 17 00:00:00 2001 From: Your Name Date: Thu, 15 Oct 2026 20:50:52 -0700 Subject: [PATCH 4/6] cli-69: fix queue bugs --- cecli/coders/base_coder.py | 11 +++- cecli/commands/__init__.py | 3 ++ cecli/commands/run_queue.py | 79 +++++++++++++++++++++++++++++ cecli/helpers/command_queue.py | 42 +++++++++++++++ cecli/tui/app.py | 66 ++++++++---------------- requirements.txt | 4 +- requirements/common-constraints.txt | 2 +- 7 files changed, 157 insertions(+), 50 deletions(-) create mode 100644 cecli/commands/run_queue.py diff --git a/cecli/coders/base_coder.py b/cecli/coders/base_coder.py index dba693474ff..8953ad5a7b4 100755 --- a/cecli/coders/base_coder.py +++ b/cecli/coders/base_coder.py @@ -2026,6 +2026,15 @@ async def process_prompt_queue(self, preproc=True): failing prompt does not abort the rest of the queue. An interrupt (Ctrl+C / cancellation) re-queues the in-flight item at the front and re-raises so the caller can resume later via /run-queue. + + Each item runs through ``generate()`` rather than a bare ``run_one()``. + ``generate()`` owns the generation lifecycle: it clears the interrupt + event, sets ``run_one_completed`` False for the duration, drives the + spinner, and wires ``io.output_task``. A bare ``run_one()`` skips all + of that, so a drain that called it directly (e.g. the TUI idle drain on + the worker loop) started the turn but stalled silently. ``_processing_queue`` + still guards against the nested drain that ``run_one()`` performs at its + end. """ if self._processing_queue: return @@ -2039,7 +2048,7 @@ async def process_prompt_queue(self, preproc=True): self.io.tool_output(f"Processing queued prompt (id: {item['id']})...") try: - await self.run_one(item["text"], preproc) + await self.generate(item["text"], preproc) except (asyncio.CancelledError, KeyboardInterrupt): # Put the in-flight prompt back at the front so it is not # lost, then let the interrupt propagate to the caller. diff --git a/cecli/commands/__init__.py b/cecli/commands/__init__.py index f081e306672..1a6aa6e4a8e 100644 --- a/cecli/commands/__init__.py +++ b/cecli/commands/__init__.py @@ -73,6 +73,7 @@ from .reset import ResetCommand from .rules import RulesCommand from .run import RunCommand +from .run_queue import RunQueueCommand from .save import SaveCommand from .save_session import SaveSessionCommand from .search_memory import SearchMemoryCommand @@ -168,6 +169,7 @@ CommandRegistry.register(ResetCommand) CommandRegistry.register(RulesCommand) CommandRegistry.register(RunCommand) +CommandRegistry.register(RunQueueCommand) CommandRegistry.register(SaveCommand) CommandRegistry.register(SaveSessionCommand) CommandRegistry.register(SearchMemoryCommand) @@ -263,6 +265,7 @@ "ResetCommand", "RulesCommand", "RunCommand", + "RunQueueCommand", "SaveCommand", "SaveSessionCommand", "SearchMemoryCommand", diff --git a/cecli/commands/run_queue.py b/cecli/commands/run_queue.py new file mode 100644 index 00000000000..3b71d0cfe3f --- /dev/null +++ b/cecli/commands/run_queue.py @@ -0,0 +1,79 @@ +"""Run-queue command for CLI-33: processes queued prompts on demand. + +Primary use: after cancelling an in-flight response (Ctrl+C), the in-flight +prompt is re-queued at the front. /run-queue resumes processing the queue. +""" + +import asyncio +from typing import List + +from cecli.commands.utils.base_command import BaseCommand +from cecli.commands.utils.helpers import format_command_result +from cecli.helpers import command_queue +from cecli.signals import SwitchCoderSignal + + +class RunQueueCommand(BaseCommand): + NORM_NAME = "run-queue" + DESCRIPTION = "Process queued prompts in order" + + @classmethod + async def execute(cls, io, coder, args, **kwargs): + """Execute the run-queue command with given parameters. + + Args: + io: InputOutput instance + coder: Coder instance (may be None for some commands) + args: Command arguments (unused for run-queue) + **kwargs: Additional context + + Returns: + Formatted result string + """ + # Sad path: coder.commands is None + if not coder.commands: + return format_command_result( + io, cls.NORM_NAME, "", error="Command system not available. Cannot run queue." + ) + + # Resolve the target coder (sub-agent aware: use the foreground agent's queue) + target = command_queue.get_active_coder(coder) + + # Sad path: empty queue + if command_queue.get_queue_length(target) == 0: + io.tool_output("Queue is empty. Use /queue to add prompts first.") + return f"Successfully executed {cls.NORM_NAME}." + + # Happy path: drain the queue + try: + await target.process_prompt_queue() + return f"Successfully executed {cls.NORM_NAME}." + except (KeyboardInterrupt, asyncio.CancelledError): + # The interrupted item was re-queued; let the user know the queue paused. + io.tool_output("Queue processing paused. Use /run-queue to resume.") + return f"Successfully executed {cls.NORM_NAME}." + except (SwitchCoderSignal, SystemExit): + raise + except Exception as e: + return format_command_result(io, cls.NORM_NAME, "", error=str(e)) + + @classmethod + def get_completions(cls, io, coder, args) -> List[str]: + """Get completion options for run-queue command.""" + return [] + + @classmethod + def get_help(cls) -> str: + """Get help text for the run-queue command.""" + help_text = super().get_help() + help_text += "\nUsage:\n" + help_text += " /run-queue # Process all queued prompts in FIFO order\n" + help_text += "\nDescription:\n" + help_text += " Processes every prompt currently in the queue, in order.\n" + help_text += " Queued prompts normally run automatically after the current\n" + help_text += " response finishes. Use /run-queue to resume a queue that was\n" + help_text += " paused (e.g. after cancelling an in-flight response with Ctrl+C).\n" + help_text += "\nExamples:\n" + help_text += " /run-queue # Drains the queue, processing each prompt in order\n" + help_text += "\nSee also: /queue, /list-queue, /remove-queue\n" + return help_text \ No newline at end of file diff --git a/cecli/helpers/command_queue.py b/cecli/helpers/command_queue.py index 1463cffc18c..9be01864112 100644 --- a/cecli/helpers/command_queue.py +++ b/cecli/helpers/command_queue.py @@ -151,6 +151,48 @@ def clear_queue(coder) -> list: return items +def idle_drain_needed(coder) -> bool: + """True when the coder is idle and not already draining its queue. + + Used by the command-side drains (queue.py / insert_queue.py) and the TUI + submit handler to decide whether an enqueue should trigger an immediate + drain. When a request is in flight (``run_one_completed`` False) or a + drain is already running (``_processing_queue`` True), the post-run_one + auto-drain covers the queue instead. + """ + return bool(getattr(coder, "run_one_completed", True)) and not getattr( + coder, "_processing_queue", False + ) + + +def nested_command_completions(coder, text: str) -> list: + """Completions for a nested slash-command inside queued prompt text (CLI-33). + + While the inner command is still being typed (no space yet), suggest + matching command names from the registry (e.g. "/sa" -> "/save-session"). + Once the inner command is complete, delegate to the original command's + ``get_completions`` so ``/queue /add `` behaves exactly like ``/add ``. + """ + if not text or not text.startswith("/"): + return [] + + from cecli.commands.utils.registry import CommandRegistry + + first_space = text.find(" ") + if first_space == -1: + prefix = text[1:].lower() + return sorted( + f"/{cmd}" for cmd in CommandRegistry.list_commands() if cmd.startswith(prefix) + ) + + inner_cmd = text[:first_space] + inner_args = text[first_space + 1 :] + try: + return list(coder.commands.get_completions(inner_cmd, inner_args, coder=coder)) + except Exception: + return [] + + def _get_lock(coder) -> threading.Lock: """Return the coder's queue lock, creating one if missing.""" lock = getattr(coder, "_queue_lock", None) diff --git a/cecli/tui/app.py b/cecli/tui/app.py index 7ed46ff9b4b..fa2d60c59eb 100644 --- a/cecli/tui/app.py +++ b/cecli/tui/app.py @@ -1132,7 +1132,6 @@ def _handle_queue_command(self, stripped: str) -> None: ) _queue_drain_task = None - _queue_drain_lock = None def _schedule_idle_queue_drain(self, active_coder) -> None: """Schedule immediate queue processing on the worker loop when idle. @@ -1140,15 +1139,10 @@ def _schedule_idle_queue_drain(self, active_coder) -> None: The TUI runs on the textual thread, so coder coroutines must never be awaited here; work is scheduled on the worker's event loop instead. - The first queued item is dequeued and processed through - ``coder.generate()`` — the same lifecycle as a user-typed prompt - (it sets ``run_one_completed``, clears the interrupt event, and its - trailing auto-drain in run_one() processes any remaining items). - Calling ``run_one()``/``process_prompt_queue()`` directly here would - run outside that lifecycle and stall silently (live bug 2026-10-15). - - When a request is in flight or a drain is already running, do - nothing — the post-run_one auto-drain covers the queue. + Delegates to ``process_prompt_queue()``, which runs each queued item + through ``generate()`` (the same lifecycle as a user-typed prompt) and +ows the re-entrancy guard, per-item isolation, and interrupt re-queueing. + Calling ``run_one()`` directly here would skip that lifecycle and stall. """ from cecli.helpers import command_queue from cecli.signals import SwitchCoderSignal @@ -1160,42 +1154,22 @@ def _schedule_idle_queue_drain(self, active_coder) -> None: return async def _drain(): - # Serialize: a second scheduled drain must never run generate() - # concurrently with the first one. The lock is created lazily on - # the worker loop; the None check cannot interleave because the - # worker loop runs this task step atomically up to the acquire. - if self._queue_drain_lock is None: - self._queue_drain_lock = asyncio.Lock() - async with self._queue_drain_lock: - # Re-check just before draining: a generation cycle may have - # started between scheduling and execution. - if not command_queue.idle_drain_needed(active_coder): - return - item = command_queue.dequeue_prompt(active_coder) - if item is None: - return - try: - await active_coder.generate(item["text"], True) - except (asyncio.CancelledError, KeyboardInterrupt): - # The interrupted prompt is re-queued so nothing is lost; - # tell the user how to resume. Swallow the interrupt here - # so a bare task ending on KeyboardInterrupt cannot escape - # into the worker loop and crash it. - command_queue.insert_prompt(active_coder, item["text"], 0) - remaining = command_queue.get_queue_length(active_coder) - active_coder.io.tool_warning( - f"Queue processing interrupted. {remaining} prompt(s) remain" - " queued; use /run-queue to resume." - ) - except (SwitchCoderSignal, SystemExit): - raise - except Exception as e: - # Match process_prompt_queue per-item isolation: surface - # the failure to the user instead of dying silently. - active_coder.io.tool_warning( - f"Error processing queued prompt (id: {item['id']}): {e}" - ) - logger.warning("Idle queue drain failed", exc_info=True) + # Re-check just before draining: a generation cycle may have + # started between scheduling and execution. + if not command_queue.idle_drain_needed(active_coder): + return + try: + await active_coder.process_prompt_queue() + except (asyncio.CancelledError, KeyboardInterrupt): + # process_prompt_queue already re-queued the in-flight prompt + # and told the user how to resume; swallow the interrupt here + # so a bare task ending on KeyboardInterrupt cannot escape into + # the worker loop and crash it. + logger.warning("Idle queue drain interrupted") + except (SwitchCoderSignal, SystemExit): + raise + except Exception: + logger.warning("Idle queue drain failed", exc_info=True) def _start(): # Keep a strong reference so the asyncio task is not garbage diff --git a/requirements.txt b/requirements.txt index 64d185820fb..9e182322f7d 100644 --- a/requirements.txt +++ b/requirements.txt @@ -608,7 +608,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 @@ -642,6 +642,6 @@ zipp==3.23.0 # via # -c requirements/common-constraints.txt # importlib-metadata - + tree-sitter==0.23.2; python_version < "3.10" tree-sitter>=0.25.1; python_version >= "3.10" diff --git a/requirements/common-constraints.txt b/requirements/common-constraints.txt index 5f79887e8e4..53f2548b1f7 100644 --- a/requirements/common-constraints.txt +++ b/requirements/common-constraints.txt @@ -514,7 +514,7 @@ uvicorn[standard]==0.38.0 # via # chromadb # mcp -uvloop==0.22.1 +uvloop==0.22.1 ; platform_python_implementation != 'PyPy' and sys_platform != 'cygwin' and sys_platform != 'win32' # via uvicorn virtualenv==20.35.4 # via pre-commit From b8aa5e56310c5bbf75fb312223f82b0d2620bec3 Mon Sep 17 00:00:00 2001 From: Your Name Date: Fri, 16 Oct 2026 00:10:55 -0700 Subject: [PATCH 5/6] cli-69: updated tests --- cecli/tests/test_queue_commands.py | 40 ++++++++++++++++++++++-------- 1 file changed, 29 insertions(+), 11 deletions(-) diff --git a/cecli/tests/test_queue_commands.py b/cecli/tests/test_queue_commands.py index f5ff2eae3b6..e17b01923d1 100644 --- a/cecli/tests/test_queue_commands.py +++ b/cecli/tests/test_queue_commands.py @@ -1101,27 +1101,35 @@ def test_idg_03_idle_drain_needed_false_when_draining(self): assert command_queue.idle_drain_needed(coder) is False @pytest.mark.asyncio - async def test_idg_04_scheduled_drain_processes_idle_queue(self): + async def test_idg_04_scheduled_drain_processes_idle_queue(self, mock_io): """IDG-04: scheduling the drain for an idle coder drains its queue. Exercises the TUI scheduling helper against a fake worker whose ``loop`` is the running asyncio loop (the real TUI schedules the same - call on the worker loop). + call on the worker loop). Spies the ``generate`` seam (plan 9.14): the + real ``process_prompt_queue`` dequeues each item and runs it through + ``generate()``, so the fake only records the prompt text. """ import asyncio + from cecli.coders.base_coder import Coder from cecli.tui.app import TUI coder = _make_coder_with_commands(items=("queued prompt",)) + coder.io = mock_io coder.run_one_completed = True coder._processing_queue = False processed = [] - async def fake_drain(): - while coder.prompt_queue: - processed.append(command_queue.dequeue_prompt(coder)["text"]) + async def fake_generate(text, preproc=True): + processed.append(text) - coder.process_prompt_queue = fake_drain + coder.generate = fake_generate + # Drive the real drain loop on the mock coder: it dequeues each item + # and awaits coder.generate, which the fake above records. + coder.process_prompt_queue = lambda preproc=True: Coder.process_prompt_queue( + coder, preproc + ) app = TUI.__new__(TUI) fake_worker = MagicMock() @@ -1138,20 +1146,30 @@ async def fake_drain(): assert command_queue.get_queue_length(coder) == 0 @pytest.mark.asyncio - async def test_idg_05_scheduled_drain_skips_when_busy(self): - """IDG-05: a busy coder is not drained by the scheduler.""" + async def test_idg_05_scheduled_drain_skips_when_busy(self, mock_io): + """IDG-05: a busy coder is not drained by the scheduler. + + Spies the same ``generate`` seam as IDG-04: with run_one_completed=False + the scheduler must return before touching the coder, so no generate call + is recorded and the queued prompt stays in place. + """ import asyncio + from cecli.coders.base_coder import Coder from cecli.tui.app import TUI coder = _make_coder_with_commands(items=("queued prompt",)) + coder.io = mock_io coder.run_one_completed = False drain_calls = [] - async def fake_drain(): - drain_calls.append(True) + async def fake_generate(text, preproc=True): + drain_calls.append(text) - coder.process_prompt_queue = fake_drain + coder.generate = fake_generate + coder.process_prompt_queue = lambda preproc=True: Coder.process_prompt_queue( + coder, preproc + ) app = TUI.__new__(TUI) fake_worker = MagicMock() From f7e788631736ce82cedcbea2cfb864ef716cd795 Mon Sep 17 00:00:00 2001 From: Your Name Date: Fri, 16 Oct 2026 14:03:10 -0700 Subject: [PATCH 6/6] cli-69: updated tests --- cecli/commands/queue.py | 3 +-- cecli/commands/run_queue.py | 2 +- cecli/tests/test_queue_commands.py | 27 ++++++++++----------------- cecli/tui/app.py | 5 +++-- 4 files changed, 15 insertions(+), 22 deletions(-) diff --git a/cecli/commands/queue.py b/cecli/commands/queue.py index f936d89e93b..721721dbf8a 100644 --- a/cecli/commands/queue.py +++ b/cecli/commands/queue.py @@ -1,8 +1,7 @@ """Queue command for CLI-33: adds a prompt to the processing queue.""" -from typing import List - import asyncio +from typing import List from cecli.commands.utils.base_command import BaseCommand from cecli.commands.utils.helpers import format_command_result diff --git a/cecli/commands/run_queue.py b/cecli/commands/run_queue.py index 3b71d0cfe3f..f84767b88bc 100644 --- a/cecli/commands/run_queue.py +++ b/cecli/commands/run_queue.py @@ -76,4 +76,4 @@ def get_help(cls) -> str: help_text += "\nExamples:\n" help_text += " /run-queue # Drains the queue, processing each prompt in order\n" help_text += "\nSee also: /queue, /list-queue, /remove-queue\n" - return help_text \ No newline at end of file + return help_text diff --git a/cecli/tests/test_queue_commands.py b/cecli/tests/test_queue_commands.py index e17b01923d1..5a05fc9f193 100644 --- a/cecli/tests/test_queue_commands.py +++ b/cecli/tests/test_queue_commands.py @@ -759,9 +759,7 @@ def test_ntc_03_queue_save_session_delegates_session_names(self, mock_io): result = QueueCommand.get_completions(mock_io, coder, "/save-session ") assert result == session_names - coder.commands.get_completions.assert_called_once_with( - "/save-session", "", coder=coder - ) + coder.commands.get_completions.assert_called_once_with("/save-session", "", coder=coder) def test_ntc_04_queue_plain_text_returns_empty(self, mock_io): """NTC-04: non-command queued text (and empty args) have no completions.""" @@ -790,9 +788,7 @@ def test_ntc_06_insert_queue_delegates_after_index(self, mock_io): result = InsertQueueCommand.get_completions(mock_io, coder, "1 /save-session ") assert result == session_names - coder.commands.get_completions.assert_called_once_with( - "/save-session", "", coder=coder - ) + coder.commands.get_completions.assert_called_once_with("/save-session", "", coder=coder) def test_ntc_07_insert_queue_index_only_suggests_positions(self, mock_io): """NTC-07: while typing the index, /insert-queue offers clamped positions.""" @@ -943,7 +939,6 @@ def test_rtc_05_thread_safety_under_concurrent_access(self, clean_commands): assert isinstance(coder._queue_lock, type(threading.Lock())) - # ============================================================================ # Additional Tests for Edge Cases and Error Handling # ============================================================================ @@ -1127,9 +1122,7 @@ async def fake_generate(text, preproc=True): coder.generate = fake_generate # Drive the real drain loop on the mock coder: it dequeues each item # and awaits coder.generate, which the fake above records. - coder.process_prompt_queue = lambda preproc=True: Coder.process_prompt_queue( - coder, preproc - ) + coder.process_prompt_queue = lambda preproc=True: Coder.process_prompt_queue(coder, preproc) app = TUI.__new__(TUI) fake_worker = MagicMock() @@ -1167,9 +1160,7 @@ async def fake_generate(text, preproc=True): drain_calls.append(text) coder.generate = fake_generate - coder.process_prompt_queue = lambda preproc=True: Coder.process_prompt_queue( - coder, preproc - ) + coder.process_prompt_queue = lambda preproc=True: Coder.process_prompt_queue(coder, preproc) app = TUI.__new__(TUI) fake_worker = MagicMock() @@ -1296,7 +1287,11 @@ def test_tcp_07_io_partial_derivation(self): commands.FULL_ARGS_COMPLETION_COMMANDS = ("/queue", "/insert-queue") # Mirror Commands.matching_commands: match the input's first word. commands.matching_commands = lambda text: ( - ["/queue"] if text.startswith("/queue") else (["/add"] if text.startswith("/add") else None), + ( + ["/queue"] + if text.startswith("/queue") + else (["/add"] if text.startswith("/add") else None) + ), text.split()[0], "", ) @@ -1318,9 +1313,7 @@ def fake_get_completions(cmd, args="", coder=None): # Partial word: nested text is forwarded verbatim; partial is the word. results = list( - completer.get_command_completions( - None, None, "/queue /add s", ["/queue", "/add", "s"] - ) + completer.get_command_completions(None, None, "/queue /add s", ["/queue", "/add", "s"]) ) assert captured[-1] == ("/queue", "/add s") assert all(c.start_position == -1 for c in results) diff --git a/cecli/tui/app.py b/cecli/tui/app.py index fa2d60c59eb..16cef8fe09b 100644 --- a/cecli/tui/app.py +++ b/cecli/tui/app.py @@ -1141,8 +1141,9 @@ def _schedule_idle_queue_drain(self, active_coder) -> None: Delegates to ``process_prompt_queue()``, which runs each queued item through ``generate()`` (the same lifecycle as a user-typed prompt) and -ows the re-entrancy guard, per-item isolation, and interrupt re-queueing. - Calling ``run_one()`` directly here would skip that lifecycle and stall. + follows the re-entrancy guard, per-item isolation, and interrupt + re-queueing. Calling ``run_one()`` directly here would skip that + lifecycle and stall. """ from cecli.helpers import command_queue from cecli.signals import SwitchCoderSignal