CLI-52: Fix hang on interrupt during LLM processing

cecli/coders/base_coder.py

@@ -6,6 +6,7 @@
 import hashlib
 import json
 import locale
+import logging
 import math
 import mimetypes
 import os
@@ -68,6 +69,7 @@
 from ..prompts.utils.registry import PromptObject, PromptRegistry
 
 GLOBAL_DATE = date.today().isoformat()
+logger = logging.getLogger(__name__)
 
 
 class UnknownEditFormat(ValueError):
@@ -719,7 +721,8 @@ def __init__(
             customizations = dict()
             pass
 
-        self.custom = customizations
+        self.custom = customizations or {}
+        self.prompt_format = nested.getter(self.custom, "prompt_format", "default")
         self.file_diffs = nested.getter(self.args, "file_diffs", True)
 
         if nested.getter(self.custom, "prompt_map.all", None):
@@ -1560,8 +1563,18 @@ async def _run_parallel(self, with_message=None, preproc=True):
 
             try:
                 # Wait for both tasks to complete or for one to raise an exception
+                # Wait for either task to complete. FIRST_COMPLETED provides
+                # faster interrupt response and robustness against asymmetric
+                # state bugs (e.g., if only one running flag is set to False).
+                # In normal operation both tasks run indefinitely (while-loops),
+                # so neither completes first — behavior is identical to
+                # FIRST_EXCEPTION in the absence of exceptions.
+                logger.debug(
+                    "_run_parallel: Waiting for tasks "
+                    f"(input_task.done={input_task.done()}, output_task.done={output_task.done()})"
+                )
                 done, pending = await asyncio.wait(
-                    [input_task, output_task], return_when=asyncio.FIRST_EXCEPTION
+                    [input_task, output_task], return_when=asyncio.FIRST_COMPLETED
                 )
 
                 # Check for exceptions
@@ -1576,6 +1589,17 @@ async def _run_parallel(self, with_message=None, preproc=True):
                 # Signal tasks to stop
                 self.input_running = False
                 self.output_running = False
+                # Clear interrupt_event to prevent state leakage between
+                # interrupt cycles. Without this, a stale interrupt_event can
+                # cause the next generate() call to immediately abort.
+                # ThreadSafeEvent.clear() is thread-safe and idempotent.
+                self.interrupt_event.clear()
+
+                # Log state before cancelling tasks
+                logger.debug(
+                    "_run_parallel finally: "
+                    f"input_running={self.input_running}, output_running={self.output_running}"
+                )
 
                 # Cancel tasks
                 input_task.cancel()
@@ -1602,11 +1626,22 @@ async def input_task(self, preproc):
         This task manages the input loop and coordinates with output_task.
         """
         while self.input_running:
+            logger.debug(f"input_task loop: input_running={self.input_running}")
             try:
                 # Wait for commands to finish
                 if not self.commands.cmd_running_event.is_set():
-                    await self.commands.cmd_running_event.wait()
+                    logger.debug("input_task: Waiting for commands to finish")
+                    # Use interruptible wait with timeout to allow checking input_running
+                    while not self.commands.cmd_running_event.is_set() and self.input_running:
+                        try:
+                            await asyncio.wait_for(
+                                self.commands.cmd_running_event.wait(), timeout=0.1
+                            )
+                        except asyncio.TimeoutError:
+                            continue
                     continue
+                # This duplicate check is now handled by the interruptible wait above
+                pass
 
                 # Wait for input task completion
                 if self.io.input_task and self.io.input_task.done():
@@ -1668,11 +1703,22 @@ async def output_task(self, preproc):
         This task manages the output loop and coordinates with input_task.
         """
         while self.output_running:
+            logger.debug(f"output_task loop: output_running={self.output_running}")
             try:
                 # Wait for commands to finish
                 if not self.commands.cmd_running_event.is_set():
-                    await self.commands.cmd_running_event.wait()
+                    logger.debug("output_task: Waiting for commands to finish")
+                    # Use interruptible wait with timeout to allow checking output_running
+                    while not self.commands.cmd_running_event.is_set() and self.output_running:
+                        try:
+                            await asyncio.wait_for(
+                                self.commands.cmd_running_event.wait(), timeout=0.1
+                            )
+                        except asyncio.TimeoutError:
+                            continue
                     continue
+                    # This duplicate check is now handled by the interruptible wait above
+                    pass
 
                 # Check if we have a user message to process
                 if self.user_message and not self.io.get_confirmation_acknowledgement():

cecli/prompts/test.yml

@@ -0,0 +1,14 @@
+# Test prompts - inherits from base.yaml
+_inherits: [base]
+
+# Minimal system prompt for testing purposes
+main_system: |
+  You are a test assistant. Your only goal is to assist in running tests.
+  {final_reminders}
+
+system_reminder: |
+  <context name="critical_reminders">
+  - You are in test mode. Focus on test execution and reporting.
+  {lazy_prompt}
+  {shell_cmd_reminder}
+  </context>

cecli/tui/io.py

@@ -646,21 +646,6 @@ async def confirm_ask(
         except asyncio.CancelledError:
             return False
 
-    async def stop_task_streams(self):
-        """Override to avoid asyncio issues in worker thread.
-
-        TUI doesn't use the same parallel streaming, so this is a no-op.
-        """
-        pass
-
-    async def stop_input_task(self):
-        """Override to avoid asyncio issues in worker thread."""
-        pass
-
-    async def stop_output_task(self):
-        """Override to avoid asyncio issues in worker thread."""
-        pass
-
     def request_exit(self):
         """Request the TUI to exit gracefully.
 

cecli/tui/worker.py

@@ -221,6 +221,17 @@ def interrupt(self):
                 if hasattr(target_coder, "output_running"):
                     target_coder.output_running = False
 
+                # Also set input_running to False to prevent input_task deadlock.
+                # _run_parallel's asyncio.wait(FIRST_EXCEPTION) hangs because
+                # input_task keeps looping while input_running stays True.
+                if hasattr(target_coder, "input_running"):
+                    target_coder.input_running = False
+
+            # Also cancel the input task to unblock it from get_input()
+            if self.loop and self.loop.is_running():
+                if hasattr(target_coder.io, "input_task") and target_coder.io.input_task:
+                    self.loop.call_soon_threadsafe(target_coder.io.input_task.cancel)
+
             # Cancel any tracked generate task on the coder directly
             if hasattr(target_coder, "interrupt_event") and target_coder.interrupt_event:
                 target_coder.interrupt_event.set()