diff --git a/cecli/coders/base_coder.py b/cecli/coders/base_coder.py
index 2633bc319e1..11b3890956d 100755
--- a/cecli/coders/base_coder.py
+++ b/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():
diff --git a/cecli/prompts/test.yml b/cecli/prompts/test.yml
new file mode 100644
index 00000000000..506b1d448f3
--- /dev/null
+++ b/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>
\ No newline at end of file
diff --git a/cecli/tui/io.py b/cecli/tui/io.py
index b17da16a2f4..c4ac2cfebc7 100644
--- a/cecli/tui/io.py
+++ b/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.
 
diff --git a/cecli/tui/worker.py b/cecli/tui/worker.py
index 4b28fcf7e89..63050765457 100644
--- a/cecli/tui/worker.py
+++ b/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()
diff --git a/tests/e2e/test_tui_interrupt.py b/tests/e2e/test_tui_interrupt.py
new file mode 100644
index 00000000000..24e4d708312
--- /dev/null
+++ b/tests/e2e/test_tui_interrupt.py
@@ -0,0 +1,263 @@
+"""E2E tests for TUI interrupt scenarios using Playwright.
+
+These tests require a running cecli TUI instance and Playwright.
+They are designed to be run manually or in a CI environment with
+appropriate setup.
+
+Test Cases Covered:
+  - TC-INTERRUPT-001: Single interrupt
+  - TC-INTERRUPT-002: Double interrupt (primary bug)
+  - TC-INTERRUPT-003: Triple+ interrupt
+  - TC-INTERRUPT-004: Sub-agent interrupt
+  - TC-INTERRUPT-006: Rapid message + interrupt sequence
+"""
+
+import os
+import subprocess
+import time
+
+import pytest
+
+# Skip E2E tests by default unless --run-e2e flag is passed
+pytestmark = pytest.mark.skipif(
+    not os.environ.get("RUN_E2E_TESTS"),
+    reason="E2E tests require running cecli TUI; set RUN_E2E_TESTS=1 to enable",
+)
+
+
+# ---------------------------------------------------------------------------
+# Helper: launch cecli in TUI mode and return the process handle
+# ---------------------------------------------------------------------------
+
+
+def _launch_cecli_tui() -> subprocess.Popen:
+    """Launch cecli in TUI mode with a mock LLM provider.
+
+    Returns a Popen process handle. The caller is responsible for
+    terminating the process.
+    """
+    env = os.environ.copy()
+    # Use a mock/slow model to give us time to send interrupts
+    env.setdefault("CE CLI_MODEL", "mock/slow")
+
+    proc = subprocess.Popen(
+        ["python", "-m", "cecli", "--gui"],
+        stdin=subprocess.PIPE,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        env=env,
+        text=True,
+    )
+    # Give the TUI a moment to start
+    time.sleep(2)
+    return proc
+
+
+def _send_interrupt(proc: subprocess.Popen) -> None:
+    """Send Ctrl+C (SIGINT) to the cecli process."""
+    import signal
+
+    if os.name == "nt":
+        # Windows: send CTRL_C_EVENT
+        proc.send_signal(signal.CTRL_C_EVENT)
+    else:
+        proc.send_signal(signal.SIGINT)
+
+
+def _send_message(proc: subprocess.Popen, message: str) -> None:
+    """Send a message to cecli via stdin."""
+    if proc.stdin:
+        proc.stdin.write(message + "\n")
+        proc.stdin.flush()
+
+
+def _wait_for_responsive(proc: subprocess.Popen, timeout: float = 5.0) -> bool:
+    """Wait for the TUI to become responsive after an interrupt.
+
+    Returns True if the process is still alive (responsive), False if it
+    appears hung or terminated.
+    """
+    deadline = time.time() + timeout
+    while time.time() < deadline:
+        poll = proc.poll()
+        if poll is not None:
+            # Process exited
+            return False
+        time.sleep(0.5)
+    # Still running after timeout → likely responsive
+    return proc.poll() is None
+
+
+# ---------------------------------------------------------------------------
+# Test Cases
+# ---------------------------------------------------------------------------
+
+
+class TestTUIInterruptE2E:
+    """End-to-end interrupt tests for the cecli TUI."""
+
+    def test_single_interrupt(self):
+        """TC-INTERRUPT-001: Single interrupt scenario.
+
+        Start cecli, send a message, press Ctrl+C once during processing.
+        Expected: Processing stops, TUI returns to responsive state.
+        """
+        proc = _launch_cecli_tui()
+        try:
+            # Send a message that triggers LLM processing
+            _send_message(proc, "Write a hello world function in Python")
+            time.sleep(1)  # Let processing start
+
+            # Send single interrupt
+            _send_interrupt(proc)
+
+            # Wait for TUI to become responsive
+            responsive = _wait_for_responsive(proc, timeout=5.0)
+            assert responsive, "TUI did not return to responsive state after single interrupt"
+
+            # Verify we can send another message
+            _send_message(proc, "What is 2+2?")
+            time.sleep(1)
+            responsive = _wait_for_responsive(proc, timeout=3.0)
+            assert responsive, "TUI not responsive after sending follow-up message"
+
+        finally:
+            proc.terminate()
+            proc.wait(timeout=5)
+
+    def test_double_interrupt(self):
+        """TC-INTERRUPT-002: Double interrupt scenario (primary bug).
+
+        Start cecli, send a message, press Ctrl+C twice during processing.
+        Expected: TUI does NOT hang, returns to responsive state.
+        """
+        proc = _launch_cecli_tui()
+        try:
+            # Send a message that triggers LLM processing
+            _send_message(proc, "Write a hello world function in Python")
+            time.sleep(1)  # Let processing start
+
+            # Send first interrupt
+            _send_interrupt(proc)
+            time.sleep(0.5)
+
+            # Send second interrupt immediately after
+            _send_interrupt(proc)
+
+            # Wait for TUI to become responsive
+            responsive = _wait_for_responsive(proc, timeout=10.0)
+            assert responsive, "TUI hung after double interrupt — primary bug not fixed"
+
+            # Verify we can send another message
+            _send_message(proc, "What is 2+2?")
+            time.sleep(1)
+            responsive = _wait_for_responsive(proc, timeout=3.0)
+            assert responsive, "TUI not responsive after sending follow-up message"
+
+        finally:
+            proc.terminate()
+            proc.wait(timeout=5)
+
+    def test_triple_interrupt(self):
+        """TC-INTERRUPT-003: Triple+ interrupt scenario.
+
+        Start cecli, send a message, press Ctrl+C three times rapidly.
+        Expected: TUI remains responsive, no hang.
+        """
+        proc = _launch_cecli_tui()
+        try:
+            # Send a message that triggers LLM processing
+            _send_message(proc, "Write a hello world function in Python")
+            time.sleep(1)  # Let processing start
+
+            # Send three interrupts in rapid succession
+            for _ in range(3):
+                _send_interrupt(proc)
+                time.sleep(0.3)
+
+            # Wait for TUI to become responsive
+            responsive = _wait_for_responsive(proc, timeout=5.0)
+            assert responsive, "TUI hung after triple interrupt"
+
+            # Verify we can send another message
+            _send_message(proc, "What is 2+2?")
+            time.sleep(1)
+            responsive = _wait_for_responsive(proc, timeout=3.0)
+            assert responsive, "TUI not responsive after sending follow-up message"
+
+        finally:
+            proc.terminate()
+            proc.wait(timeout=5)
+
+    def test_sub_agent_interrupt(self):
+        """TC-INTERRUPT-004: Sub-agent interrupt scenario.
+
+        Start cecli, activate a sub-agent, send a message, press Ctrl+C.
+        Expected: Sub-agent interrupt works correctly, no deadlock.
+        """
+        proc = _launch_cecli_tui()
+        try:
+            # Activate a sub-agent
+            _send_message(proc, "/agent researcher")
+            time.sleep(2)  # Wait for sub-agent to activate
+
+            # Send a message to the sub-agent
+            _send_message(proc, "Research the Python asyncio library")
+            time.sleep(1)  # Let processing start
+
+            # Send interrupt
+            _send_interrupt(proc)
+
+            # Wait for TUI to become responsive
+            responsive = _wait_for_responsive(proc, timeout=5.0)
+            assert responsive, "TUI hung after sub-agent interrupt"
+
+            # Verify we can send another message
+            _send_message(proc, "Continue")
+            time.sleep(1)
+            responsive = _wait_for_responsive(proc, timeout=3.0)
+            assert responsive, "TUI not responsive after sub-agent follow-up"
+
+        finally:
+            proc.terminate()
+            proc.wait(timeout=5)
+
+    def test_rapid_message_interrupt_sequence(self):
+        """TC-INTERRUPT-006: Rapid message + interrupt sequence.
+
+        Send multiple messages with interrupts between them.
+        Expected: Each cycle works correctly, no state leakage.
+        """
+        proc = _launch_cecli_tui()
+        try:
+            # Message 1 + single interrupt
+            _send_message(proc, "Write a hello world function")
+            time.sleep(1)
+            _send_interrupt(proc)
+            responsive = _wait_for_responsive(proc, timeout=5.0)
+            assert responsive, "TUI hung after message 1 + interrupt"
+
+            # Message 2 + double interrupt
+            _send_message(proc, "Write a factorial function")
+            time.sleep(1)
+            _send_interrupt(proc)
+            time.sleep(0.3)
+            _send_interrupt(proc)
+            responsive = _wait_for_responsive(proc, timeout=5.0)
+            assert responsive, "TUI hung after message 2 + double interrupt"
+
+            # Message 3 (normal, no interrupt)
+            _send_message(proc, "What is 2+2?")
+            time.sleep(2)
+            responsive = _wait_for_responsive(proc, timeout=5.0)
+            assert responsive, "TUI hung after message 3 (normal)"
+
+            # Message 4 (normal, no interrupt)
+            _send_message(proc, "What is the capital of France?")
+            time.sleep(2)
+            responsive = _wait_for_responsive(proc, timeout=5.0)
+            assert responsive, "TUI hung after message 4 (normal)"
+
+        finally:
+            proc.terminate()
+            proc.wait(timeout=5)
diff --git a/tests/e2e/test_tui_regression.py b/tests/e2e/test_tui_regression.py
new file mode 100644
index 00000000000..cceb0d80d4b
--- /dev/null
+++ b/tests/e2e/test_tui_regression.py
@@ -0,0 +1,144 @@
+"""E2E regression tests for normal (non-interrupt) TUI operation.
+
+These tests require a running cecli TUI instance and Playwright.
+They are designed to be run manually or in a CI environment with
+appropriate setup.
+
+Test Cases Covered:
+  - TC-INTERRUPT-005: Normal (non-interrupt) regression
+  - TC-INTERRUPT-010: Non-TUI (headless) regression
+"""
+
+import os
+import subprocess
+import time
+
+import pytest
+
+# Skip E2E tests by default unless --run-e2e flag is passed
+pytestmark = pytest.mark.skipif(
+    not os.environ.get("RUN_E2E_TESTS"),
+    reason="E2E tests require running cecli TUI; set RUN_E2E_TESTS=1 to enable",
+)
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _launch_cecli_tui() -> subprocess.Popen:
+    """Launch cecli in TUI mode with a mock LLM provider."""
+    env = os.environ.copy()
+    env.setdefault("CE CLI_MODEL", "mock/slow")
+
+    proc = subprocess.Popen(
+        ["python", "-m", "cecli", "--gui"],
+        stdin=subprocess.PIPE,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        env=env,
+        text=True,
+    )
+    time.sleep(2)
+    return proc
+
+
+def _send_message(proc: subprocess.Popen, message: str) -> None:
+    """Send a message to cecli via stdin."""
+    if proc.stdin:
+        proc.stdin.write(message + "\n")
+        proc.stdin.flush()
+
+
+def _wait_for_responsive(proc: subprocess.Popen, timeout: float = 5.0) -> bool:
+    """Wait for the TUI to become responsive."""
+    deadline = time.time() + timeout
+    while time.time() < deadline:
+        poll = proc.poll()
+        if poll is not None:
+            return False
+        time.sleep(0.5)
+    return proc.poll() is None
+
+
+def _launch_cecli_headless() -> subprocess.Popen:
+    """Launch cecli in non-TUI (headless) mode."""
+    env = os.environ.copy()
+    env.setdefault("CE CLI_MODEL", "mock/slow")
+
+    proc = subprocess.Popen(
+        ["python", "-m", "cecli", "--message", "test"],
+        stdin=subprocess.PIPE,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        env=env,
+        text=True,
+    )
+    time.sleep(1)
+    return proc
+
+
+# ---------------------------------------------------------------------------
+# Test Cases
+# ---------------------------------------------------------------------------
+
+
+class TestTUIRegressionE2E:
+    """End-to-end regression tests for the cecli TUI."""
+
+    def test_normal_operation_regression(self):
+        """TC-INTERRUPT-005: Normal (non-interrupt) regression.
+
+        Start cecli, send multiple messages without interrupts.
+        Expected: All messages process normally, no change in behavior.
+        """
+        proc = _launch_cecli_tui()
+        try:
+            # Send message 1
+            _send_message(proc, "Write a hello world function in Python")
+            time.sleep(2)
+            responsive = _wait_for_responsive(proc, timeout=5.0)
+            assert responsive, "TUI hung after message 1"
+
+            # Send message 2
+            _send_message(proc, "What is 2+2?")
+            time.sleep(2)
+            responsive = _wait_for_responsive(proc, timeout=5.0)
+            assert responsive, "TUI hung after message 2"
+
+            # Send message 3
+            _send_message(proc, "What is the capital of France?")
+            time.sleep(2)
+            responsive = _wait_for_responsive(proc, timeout=5.0)
+            assert responsive, "TUI hung after message 3"
+
+        finally:
+            proc.terminate()
+            proc.wait(timeout=5)
+
+    def test_non_tui_headless_regression(self):
+        """TC-INTERRUPT-010: Non-TUI (headless) regression.
+
+        Start cecli in non-TUI mode, send a message, verify it completes.
+        Expected: Non-TUI operation is unchanged; no regression.
+        """
+        proc = _launch_cecli_headless()
+        try:
+            # Wait for the process to complete
+            try:
+                stdout, stderr = proc.communicate(timeout=10)
+            except subprocess.TimeoutExpired:
+                proc.kill()
+                stdout, stderr = proc.communicate()
+                pytest.fail("Non-TUI cecli process timed out")
+
+            # Verify process exited cleanly
+            assert proc.returncode == 0, (
+                f"Non-TUI cecli exited with code {proc.returncode}:\n" f"stderr: {stderr}"
+            )
+
+        finally:
+            if proc.poll() is None:
+                proc.terminate()
+                proc.wait(timeout=5)
diff --git a/tests/fixtures/mock_llm.py b/tests/fixtures/mock_llm.py
new file mode 100644
index 00000000000..6fb10ce26f7
--- /dev/null
+++ b/tests/fixtures/mock_llm.py
@@ -0,0 +1,53 @@
+"""Mock LLM provider for interrupt testing.
+
+Provides a controllable LLM that simulates slow processing
+and supports cancellation via interrupt_event.
+"""
+
+import asyncio
+from typing import AsyncIterator, Optional
+
+
+class MockLLMProvider:
+    """Mock LLM with controllable delay and interrupt support."""
+
+    def __init__(self, delay: float = 5.0, response: str = "Mock response"):
+        self.delay = delay
+        self.response = response
+        self.interrupt_event: Optional[asyncio.Event] = None
+        self.call_count = 0
+
+    def set_interrupt_event(self, event: asyncio.Event) -> None:
+        """Set the interrupt event for cancellation support."""
+        self.interrupt_event = event
+
+    async def generate(self, prompt: str) -> str:
+        """Generate a response with controllable delay.
+
+        Checks interrupt_event periodically to support cancellation.
+        """
+        self.call_count += 1
+
+        # Simulate processing with interrupt checks
+        steps = 10
+        for i in range(steps):
+            if self.interrupt_event and self.interrupt_event.is_set():
+                raise asyncio.CancelledError("Interrupted by user")
+            await asyncio.sleep(self.delay / steps)
+
+        return self.response
+
+    async def stream(self, prompt: str) -> AsyncIterator[str]:
+        """Stream response in chunks with interrupt support."""
+        chunks = [self.response[i : i + 5] for i in range(0, len(self.response), 5)]
+
+        for chunk in chunks:
+            if self.interrupt_event and self.interrupt_event.is_set():
+                raise asyncio.CancelledError("Interrupted by user")
+            yield chunk
+            await asyncio.sleep(self.delay / len(chunks))
+
+
+def create_mock_llm(delay: float = 5.0, response: str = "Mock response") -> MockLLMProvider:
+    """Factory function to create a mock LLM provider."""
+    return MockLLMProvider(delay=delay, response=response)
diff --git a/tests/fixtures/test_coder.py b/tests/fixtures/test_coder.py
new file mode 100644
index 00000000000..e580d0a581d
--- /dev/null
+++ b/tests/fixtures/test_coder.py
@@ -0,0 +1,49 @@
+"""Test coder fixture for interrupt testing.
+
+Provides a BaseCoder subclass with instrumented state
+for testing interrupt behavior.
+"""
+
+import asyncio
+from unittest.mock import MagicMock
+
+from cecli.coders.base_coder import Coder
+
+
+class TestCoder(Coder):
+    """Test coder with controllable state for interrupt testing."""
+
+    prompt_format = "test"
+
+    def __init__(self, main_model=None, io=None, **kwargs):
+        if main_model is None:
+            main_model = MagicMock()
+        if io is None:
+            io = MagicMock()
+        super().__init__(main_model, io, **kwargs)
+        self.input_running = False
+        self.output_running = False
+        self.interrupt_event = asyncio.Event()
+        self.io = MagicMock()
+        self.io.tool_error = MagicMock()
+        self.io.tool_output = MagicMock()
+
+    async def generate(self, prompt: str) -> str:
+        """Mock generate that respects interrupt_event."""
+        self.interrupt_event.clear()
+        for _ in range(10):
+            if self.interrupt_event.is_set():
+                raise asyncio.CancelledError("Interrupted")
+            await asyncio.sleep(0.1)
+        return "Test response"
+
+    async def get_input(self) -> str:
+        """Mock get_input that respects input_running."""
+        while self.input_running:
+            await asyncio.sleep(0.1)
+        return ""
+
+
+def create_test_coder(main_model=None, io=None):
+    """Factory function to create a test coder instance."""
+    return TestCoder(main_model=main_model, io=io)
diff --git a/tests/fixtures/test_worker.py b/tests/fixtures/test_worker.py
new file mode 100644
index 00000000000..3be06f4d8d9
--- /dev/null
+++ b/tests/fixtures/test_worker.py
@@ -0,0 +1,28 @@
+"""Test worker fixture for interrupt testing.
+
+Provides a CoderWorker with mocked dependencies
+for testing interrupt behavior.
+"""
+
+from unittest.mock import MagicMock
+
+from cecli.tui.worker import CoderWorker
+
+
+def create_test_worker() -> CoderWorker:
+    """Factory function to create a test worker instance.
+
+    Returns a CoderWorker with mocked AgentService and IO.
+    """
+    worker = CoderWorker()
+
+    # Mock the agent service
+    worker.agent_service = MagicMock()
+    worker.agent_service.get_foreground_coder.return_value = MagicMock()
+
+    # Mock IO
+    worker.io = MagicMock()
+    worker.io.output_task = MagicMock()
+    worker.io.input_task = MagicMock()
+
+    return worker
diff --git a/tests/integration/test_interrupt_flow.py b/tests/integration/test_interrupt_flow.py
new file mode 100644
index 00000000000..c5a49eb428f
--- /dev/null
+++ b/tests/integration/test_interrupt_flow.py
@@ -0,0 +1,250 @@
+"""Integration tests for interrupt flow scenarios."""
+
+import asyncio
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from cecli.tui.worker import CoderWorker
+from tests.fixtures.test_coder import create_test_coder
+
+
+@pytest.mark.asyncio
+async def test_single_interrupt_scenario():
+    """Test single interrupt scenario (TC-INTERRUPT-001)."""
+    # Setup
+    coder = create_test_coder()
+    worker = CoderWorker(coder, MagicMock(), MagicMock())
+
+    # Mock the io object and its tasks
+    coder.io = MagicMock()
+    coder.io.output_task = AsyncMock()
+    coder.io.input_task = AsyncMock()
+
+    # Set initial state
+    coder.input_running = True
+    coder.output_running = True
+    coder.interrupt_event.clear()
+
+    # Mock the generate method to simulate processing
+    async def mock_generate():
+        await asyncio.sleep(0.1)  # Simulate processing time
+        return "response"
+
+    coder.generate = mock_generate
+
+    # Mock AgentService to return our test coder as foreground, and patch _run_parallel
+    with (
+        patch("cecli.helpers.agents.service.AgentService") as mock_agent_service,
+        patch.object(coder, "_run_parallel") as mock_run,
+    ):
+        mock_instance = MagicMock()
+        mock_instance.foreground_coder = coder
+        mock_agent_service.get_instance.return_value = mock_instance
+
+        # Simulate first interrupt
+        worker.interrupt()
+
+        # Verify both flags are set to False
+        assert coder.input_running is False
+        assert coder.output_running is False
+
+        # Verify interrupt_event is set
+        assert coder.interrupt_event.is_set()
+
+        # Verify _run_parallel was called
+        mock_run.assert_called()
+
+
+@pytest.mark.asyncio
+async def test_double_interrupt_scenario():
+    """Test double interrupt scenario (primary bug) (TC-INTERRUPT-002)."""
+    # Setup
+    coder = create_test_coder()
+    worker = CoderWorker(coder, MagicMock(), MagicMock())
+
+    # Mock the io object and its tasks
+    coder.io = MagicMock()
+    coder.io.output_task = AsyncMock()
+    coder.io.input_task = AsyncMock()
+
+    # Set initial state
+    coder.input_running = True
+    coder.output_running = True
+    coder.interrupt_event.clear()
+
+    # Mock the generate method to simulate processing
+    async def mock_generate():
+        await asyncio.sleep(0.1)  # Simulate processing time
+        return "response"
+
+    coder.generate = mock_generate
+
+    # Mock _run_parallel to return normally
+    with patch.object(coder, "_run_parallel", return_value="response") as mock_run:
+        # Mock AgentService to return our test coder as foreground
+        with patch("cecli.helpers.agents.service.AgentService") as mock_agent_service:
+            mock_instance = MagicMock()
+            mock_instance.foreground_coder = coder
+            mock_agent_service.get_instance.return_value = mock_instance
+
+            # Simulate first interrupt
+            worker.interrupt()
+
+            # Verify both flags are set to False after first interrupt
+            assert coder.input_running is False
+            assert coder.output_running is False
+
+            # Simulate second interrupt immediately after
+            worker.interrupt()
+
+            # Verify both flags remain False (no regression)
+            assert coder.input_running is False
+            assert coder.output_running is False
+
+            # Verify interrupt_event is set both times
+            assert coder.interrupt_event.is_set()
+
+            # Verify _run_parallel was called
+            mock_run.assert_called()
+
+
+@pytest.mark.asyncio
+async def test_triple_interrupt_scenario():
+    """Test triple+ interrupt scenario (TC-INTERRUPT-003)."""
+    # Setup
+    coder = create_test_coder()
+    worker = CoderWorker(coder, MagicMock(), MagicMock())
+
+    # Mock the io object and its tasks
+    coder.io = MagicMock()
+    coder.io.output_task = AsyncMock()
+    coder.io.input_task = AsyncMock()
+
+    # Set initial state
+    coder.input_running = True
+    coder.output_running = True
+    coder.interrupt_event.clear()
+
+    # Mock the generate method to simulate processing
+    async def mock_generate():
+        await asyncio.sleep(0.1)  # Simulate processing time
+        return "response"
+
+    coder.generate = mock_generate
+
+    # Mock _run_parallel to return normally
+    with patch.object(coder, "_run_parallel", return_value="response") as mock_run:
+        # Mock AgentService to return our test coder as foreground
+        with patch("cecli.helpers.agents.service.AgentService") as mock_agent_service:
+            mock_instance = MagicMock()
+            mock_instance.foreground_coder = coder
+            mock_agent_service.get_instance.return_value = mock_instance
+
+            # Simulate three interrupts in rapid succession
+            for i in range(3):
+                worker.interrupt()
+
+                # Verify both flags are set to False after each interrupt
+                assert coder.input_running is False
+                assert coder.output_running is False
+
+            # Verify interrupt_event is set
+            assert coder.interrupt_event.is_set()
+
+            # Verify _run_parallel was called
+            mock_run.assert_called()
+
+
+@pytest.mark.asyncio
+async def test_normal_operation_regression():
+    """Test normal (non-interrupt) regression (TC-INTERRUPT-005)."""
+    # Setup
+    coder = create_test_coder()
+
+    # Mock the io object and its tasks
+    coder.io = MagicMock()
+    coder.io.output_task = AsyncMock()
+    coder.io.input_task = AsyncMock()
+
+    # Set initial state
+    coder.input_running = True
+    coder.output_running = True
+    coder.interrupt_event.clear()
+
+    # Mock the generate method to simulate processing
+    async def mock_generate():
+        await asyncio.sleep(0.01)  # Short processing time
+        return "response"
+
+    coder.generate = mock_generate
+
+    # Mock _run_parallel to return normally
+    with patch.object(coder, "_run_parallel", return_value="response") as mock_run:
+        # Simulate normal operation without interrupts
+        result = await coder.generate()
+
+        # Verify processing completed normally
+        assert result == "response"
+
+        # Verify _run_parallel was called
+        mock_run.assert_called()
+
+
+@pytest.mark.asyncio
+async def test_rapid_message_interrupt_sequence():
+    """Test rapid message + interrupt sequence (TC-INTERRUPT-006)."""
+    # Setup
+    coder = create_test_coder()
+    worker = CoderWorker(coder, MagicMock(), MagicMock())
+
+    # Mock the io object and its tasks
+    coder.io = MagicMock()
+    coder.io.output_task = AsyncMock()
+    coder.io.input_task = AsyncMock()
+
+    # Set initial state
+    coder.input_running = True
+    coder.output_running = True
+    coder.interrupt_event.clear()
+
+    # Mock the generate method to simulate processing
+    async def mock_generate():
+        await asyncio.sleep(0.01)  # Short processing time
+        return "response"
+
+    coder.generate = mock_generate
+
+    # Mock _run_parallel to return normally
+    with patch.object(coder, "_run_parallel", return_value="response") as mock_run:
+        # Mock AgentService to return our test coder as foreground
+        with patch("cecli.helpers.agents.service.AgentService") as mock_agent_service:
+            mock_instance = MagicMock()
+            mock_instance.foreground_coder = coder
+            mock_agent_service.get_instance.return_value = mock_instance
+
+            # Simulate message 1 + interrupt
+            await coder.generate()  # Message 1
+            worker.interrupt()  # Interrupt 1
+            assert coder.input_running is False
+            assert coder.output_running is False
+
+            # Simulate message 2 + double interrupt
+            await coder.generate()  # Message 2
+            worker.interrupt()  # Interrupt 2a
+            worker.interrupt()  # Interrupt 2b
+            assert coder.input_running is False
+            assert coder.output_running is False
+
+            # Simulate message 3 (normal)
+            await coder.generate()  # Message 3
+
+            # Simulate message 4 (normal)
+            await coder.generate()  # Message 4
+
+            # Verify _run_parallel was called for each message
+            assert mock_run.call_count == 4
+
+
+if __name__ == "__main__":
+    pytest.main([__file__])
diff --git a/tests/integration/test_normal_regression.py b/tests/integration/test_normal_regression.py
new file mode 100644
index 00000000000..177d9aeafab
--- /dev/null
+++ b/tests/integration/test_normal_regression.py
@@ -0,0 +1,95 @@
+"""Integration tests for normal (non-interrupt) regression."""
+
+import asyncio
+from unittest.mock import MagicMock
+
+import pytest
+
+from cecli.coders.base_coder import Coder
+
+
+@pytest.mark.asyncio
+async def test_normal_operation_multiple_messages():
+    """Test that multiple messages process normally without interruption (TC-INTERRUPT-005)."""
+    mock_io = MagicMock()
+    coder = Coder(main_model=MagicMock(), io=mock_io)
+
+    # Set initial state
+    coder.input_running = True
+    coder.output_running = True
+    coder.interrupt_event.clear()
+
+    # Mock the generate method to simulate quick processing
+    async def mock_generate():
+        await asyncio.sleep(0.01)
+        return "response"
+
+    coder.generate = mock_generate
+
+    # Simulate three normal message completions
+    for i in range(3):
+        result = await coder.generate()
+        assert result == "response"
+
+    # Verify no interrupt state was triggered
+    assert not coder.interrupt_event.is_set()
+
+
+@pytest.mark.asyncio
+async def test_normal_operation_no_premature_unblocking():
+    """Test that _run_parallel does not unblock prematurely under normal operation."""
+    mock_io = MagicMock()
+    coder = Coder(main_model=MagicMock(), io=mock_io)
+
+    # Set initial state simulating a run
+    coder.input_running = True
+    coder.output_running = True
+    coder.interrupt_event.clear()
+
+    # Mock the generate method
+    async def mock_generate():
+        await asyncio.sleep(0.05)
+        return "response"
+
+    coder.generate = mock_generate
+
+    # Simulate normal completion
+    result = await coder.generate()
+    assert result == "response"
+
+    # Verify no cancellation or premature unblocking occurred
+    assert not coder.interrupt_event.is_set()
+    # In normal flow, both running flags should be reset by _run_parallel's finally
+    # (We simulate this by manually checking the expected behavior)
+
+
+@pytest.mark.asyncio
+async def test_normal_operation_spinner_behavior():
+    """Test that spinner starts and stops correctly during normal operation."""
+    mock_io = MagicMock()
+    coder = Coder(main_model=MagicMock(), io=mock_io)
+
+    # Set initial state
+    coder.input_running = True
+    coder.output_running = True
+    coder.interrupt_event.clear()
+
+    # Verify spinner operations are available
+    assert hasattr(mock_io, "tool_output")
+    assert hasattr(mock_io, "tool_error")
+
+    # Simulate processing and verify no errors triggered
+    async def mock_generate():
+        await asyncio.sleep(0.01)
+        return "response"
+
+    coder.generate = mock_generate
+    result = await coder.generate()
+    assert result == "response"
+
+    # Verify no tool_error calls during normal operation
+    mock_io.tool_error.assert_not_called()
+
+
+if __name__ == "__main__":
+    pytest.main([__file__])
diff --git a/tests/integration/test_sub_agent_interrupt.py b/tests/integration/test_sub_agent_interrupt.py
new file mode 100644
index 00000000000..6efb8edd224
--- /dev/null
+++ b/tests/integration/test_sub_agent_interrupt.py
@@ -0,0 +1,147 @@
+"""Integration tests for sub-agent interrupt scenarios."""
+
+import asyncio
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from cecli.tui.worker import CoderWorker
+from tests.fixtures.test_coder import create_test_coder
+
+
+@pytest.mark.asyncio
+async def test_sub_agent_interrupt_scenario():
+    """Test sub-agent interrupt scenario (TC-INTERRUPT-004)."""
+    # Setup
+    coder = create_test_coder()
+    worker = CoderWorker(coder, MagicMock(), MagicMock())
+
+    # Mock the io object and its tasks
+    coder.io = MagicMock()
+    coder.io.output_task = AsyncMock()
+    coder.io.input_task = AsyncMock()
+
+    # Set initial state
+    coder.input_running = True
+    coder.output_running = True
+    coder.interrupt_event.clear()
+
+    # Mock the generate method to simulate processing
+    async def mock_generate():
+        await asyncio.sleep(0.1)  # Simulate processing time
+        return "response"
+
+    coder.generate = mock_generate
+
+    # Mock _run_parallel to return normally
+    with patch.object(coder, "_run_parallel", return_value="response") as mock_run:
+        # Mock AgentService to return our test coder as foreground
+        with patch("cecli.helpers.agents.service.AgentService") as mock_agent_service:
+            mock_instance = MagicMock()
+            mock_instance.foreground_coder = coder
+            mock_agent_service.get_instance.return_value = mock_instance
+
+            # Simulate interrupt
+            worker.interrupt()
+
+            # Verify both flags are set to False
+            assert coder.input_running is False
+            assert coder.output_running is False
+
+            # Verify interrupt_event is set
+            assert coder.interrupt_event.is_set()
+
+            # Verify _run_parallel was called
+            mock_run.assert_called()
+
+            # Verify no AttributeError from missing input_running attribute
+            # (This is tested implicitly by the hasattr check in worker.interrupt)
+
+
+@pytest.mark.asyncio
+async def test_sub_agent_interrupt_with_layers_2_and_3():
+    """Test sub-agent interrupt with Layers 2 and 3 applied."""
+    # Setup
+    coder = create_test_coder()
+    worker = CoderWorker(coder, MagicMock(), MagicMock())
+
+    # Mock the io object and its tasks
+    coder.io = MagicMock()
+    coder.io.output_task = AsyncMock()
+    coder.io.input_task = AsyncMock()
+
+    # Set initial state
+    coder.input_running = True
+    coder.output_running = True
+    coder.interrupt_event.clear()
+
+    # Mock the generate method to simulate processing
+    async def mock_generate():
+        await asyncio.sleep(0.1)  # Simulate processing time
+        return "response"
+
+    coder.generate = mock_generate
+
+    # Mock _run_parallel to return normally
+    with patch.object(coder, "_run_parallel", return_value="response") as mock_run:
+        # Mock AgentService to return our test coder as foreground
+        with patch("cecli.helpers.agents.service.AgentService") as mock_agent_service:
+            mock_instance = MagicMock()
+            mock_instance.foreground_coder = coder
+            mock_agent_service.get_instance.return_value = mock_instance
+
+            # Simulate interrupt
+            worker.interrupt()
+
+            # Verify both flags are set to False
+            assert coder.input_running is False
+            assert coder.output_running is False
+
+            # Verify interrupt_event is set
+            assert coder.interrupt_event.is_set()
+
+            # Verify _run_parallel was called
+            mock_run.assert_called()
+
+
+@pytest.mark.asyncio
+async def test_sub_agent_interrupt_no_attribute_error():
+    """Test that sub-agent interrupt does not raise AttributeError when input_running is missing."""
+    # Setup
+    coder = create_test_coder()
+    worker = CoderWorker(coder, MagicMock(), MagicMock())
+
+    # Mock the io object and its tasks
+    coder.io = MagicMock()
+    coder.io.output_task = AsyncMock()
+    coder.io.input_task = AsyncMock()
+
+    # Set initial state - but remove input_running to simulate sub-agent
+    coder.output_running = True
+    coder.interrupt_event.clear()
+    if hasattr(coder, "input_running"):
+        delattr(coder, "input_running")
+
+    # Mock _run_parallel to return normally
+    with patch.object(coder, "_run_parallel", return_value="response") as mock_run:
+        # Mock AgentService to return our test coder as foreground
+        with patch("cecli.helpers.agents.service.AgentService") as mock_agent_service:
+            mock_instance = MagicMock()
+            mock_instance.foreground_coder = coder
+            mock_agent_service.get_instance.return_value = mock_instance
+
+            # Simulate interrupt - should not raise AttributeError
+            worker.interrupt()
+
+            # Verify output_running is set to False
+            assert coder.output_running is False
+
+            # Verify interrupt_event is set
+            assert coder.interrupt_event.is_set()
+
+            # Verify _run_parallel was called
+            mock_run.assert_called()
+
+
+if __name__ == "__main__":
+    pytest.main([__file__])
diff --git a/tests/unit/test_base_coder_interrupt.py b/tests/unit/test_base_coder_interrupt.py
new file mode 100644
index 00000000000..f608d09f391
--- /dev/null
+++ b/tests/unit/test_base_coder_interrupt.py
@@ -0,0 +1,70 @@
+"""Unit tests for BaseCoder interrupt handling."""
+
+import asyncio
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from tests.fixtures.test_coder import create_test_coder
+
+
+def test_base_coder_initial_state():
+    """Test that BaseCoder initializes with correct interrupt state."""
+    coder = create_test_coder()
+
+    # Initial state should be False for both running flags
+    assert coder.input_running is False
+    assert coder.output_running is False
+    assert not coder.interrupt_event.is_set()
+
+
+def test_base_coder_keyboard_interrupt():
+    """Test that keyboard_interrupt sets interrupt_event and calls tool_warning."""
+    coder = create_test_coder()
+
+    # Call keyboard_interrupt
+    coder.keyboard_interrupt()
+
+    # Verify interrupt_event is set
+    assert coder.interrupt_event.is_set()
+
+    # Verify tool_warning was called (not stop_task_streams)
+    coder.io.tool_warning.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_base_coder_run_parallel_sets_flags():
+    """Test that _run_parallel sets running flags to True on start."""
+    coder = create_test_coder()
+
+    # Set initial state
+    coder.input_running = False
+    coder.output_running = False
+    coder.interrupt_event.clear()
+
+    # Mock the tasks to complete quickly
+    async def quick_task():
+        await asyncio.sleep(0.01)
+        return "result"
+
+    input_task = asyncio.create_task(quick_task())
+    output_task = asyncio.create_task(quick_task())
+
+    # Mock run_one to avoid actual processing
+    coder.run_one = AsyncMock(return_value=None)
+
+    # Call _run_parallel
+    with patch("asyncio.wait") as mock_wait:
+        mock_wait.return_value = ({input_task}, {output_task})
+        await coder._run_parallel(with_message="test message")
+
+    # Verify flags were reset to False in finally block
+    assert coder.input_running is False
+    assert coder.output_running is False
+
+    # Verify interrupt_event was cleared
+    assert not coder.interrupt_event.is_set()
+
+
+if __name__ == "__main__":
+    pytest.main([__file__])
diff --git a/tests/unit/test_interrupt_event.py b/tests/unit/test_interrupt_event.py
new file mode 100644
index 00000000000..5c88244d6d5
--- /dev/null
+++ b/tests/unit/test_interrupt_event.py
@@ -0,0 +1,61 @@
+"""Unit tests for interrupt_event clearing."""
+
+import asyncio
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from tests.fixtures.test_coder import create_test_coder
+
+
+@pytest.mark.asyncio
+async def test_interrupt_event_cleared_in_run_parallel():
+    """Verify that interrupt_event is cleared in _run_parallel's finally block."""
+    # Create a test Coder instance
+    coder = create_test_coder()
+    coder.interrupt_event.set()  # Pre-set the event
+
+    # Mock run_one to avoid actual processing
+    coder.run_one = AsyncMock(return_value=None)
+
+    # Mock the tasks to complete quickly
+    input_task = asyncio.create_task(asyncio.sleep(0.01))
+    output_task = asyncio.create_task(asyncio.sleep(0.01))
+
+    # Run _run_parallel and let it complete
+    with patch("asyncio.wait") as mock_wait:
+        mock_wait.return_value = ({input_task}, {output_task})
+        await coder._run_parallel(with_message="test message")
+
+    # Assert that the interrupt_event is cleared
+    assert not coder.interrupt_event.is_set()
+
+
+@pytest.mark.asyncio
+async def test_interrupt_event_cleared_even_when_tasks_cancelled():
+    """Test that interrupt_event is cleared even when tasks are cancelled."""
+    # Create a test Coder instance
+    coder = create_test_coder()
+    coder.interrupt_event.set()  # Pre-set the event
+
+    # Mock run_one to avoid actual processing
+    coder.run_one = AsyncMock(return_value=None)
+
+    # Mock the tasks that will be cancelled
+    input_task = asyncio.create_task(asyncio.sleep(10))  # Long running
+    output_task = asyncio.create_task(asyncio.sleep(10))  # Long running
+
+    # Mock asyncio.wait to return when first task completes (but we'll cancel)
+    with patch("asyncio.wait") as mock_wait:
+        # Return one task as done, one as pending
+        mock_wait.return_value = ({input_task}, {output_task})
+
+        # Call _run_parallel
+        await coder._run_parallel(with_message="test message")
+
+        # Assert that the interrupt_event is cleared
+        assert not coder.interrupt_event.is_set()
+
+
+if __name__ == "__main__":
+    pytest.main([__file__])
diff --git a/tests/unit/test_run_parallel.py b/tests/unit/test_run_parallel.py
new file mode 100644
index 00000000000..0361b2c94f6
--- /dev/null
+++ b/tests/unit/test_run_parallel.py
@@ -0,0 +1,118 @@
+"""Unit tests for _run_parallel with FIRST_COMPLETED."""
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from tests.fixtures.test_coder import create_test_coder
+
+
+@pytest.mark.asyncio
+async def test_run_parallel_first_completed():
+    """Test that _run_parallel returns when first task completes (FIRST_COMPLETED)."""
+    # Create a test Coder instance
+    coder = create_test_coder()
+    coder.input_running = True
+    coder.output_running = True
+    coder.interrupt_event = MagicMock()
+
+    # Create mock tasks
+    input_task = AsyncMock()
+    output_task = AsyncMock()
+
+    # Configure input_task to complete quickly, output_task to hang
+    input_task.return_value = "input_result"
+    output_task.done.return_value = False  # Simulate task that hasn't completed yet
+
+    # Mock run_one to return a simple value (not awaitable)
+    coder.run_one = AsyncMock(return_value="test_result")
+
+    # Mock asyncio.wait to return when first task completes
+    with patch("asyncio.wait") as mock_wait:
+        mock_wait.return_value = ({input_task}, {output_task})
+
+        # Call _run_parallel
+        result = await coder._run_parallel(with_message="test message")
+
+        # Verify that _run_parallel returned when input_task completed
+        assert result == "test_result"
+        # Verify that both tasks were cancelled in finally block
+        input_task.cancel.assert_called_once()
+        output_task.cancel.assert_called_once()
+
+        # Verify interrupt_event was cleared (Layer 3)
+        coder.interrupt_event.clear.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_run_parallel_interrupt_event_cleared():
+    """Test that interrupt_event is cleared in _run_parallel finally block."""
+    # Create a test Coder instance
+    coder = create_test_coder()
+    coder.input_running = True
+    coder.output_running = True
+    coder.interrupt_event = MagicMock()
+    coder.interrupt_event.clear.return_value = None
+
+    # Create mock tasks that both complete quickly
+    input_task = AsyncMock()
+    output_task = AsyncMock()
+    input_task.return_value = "input_result"
+    output_task.return_value = "output_result"
+
+    # Mock run_one to return a simple value
+    coder.run_one = AsyncMock(return_value="test_result")
+
+    # Mock asyncio.wait to return when first task completes
+    with patch("asyncio.wait") as mock_wait:
+        mock_wait.return_value = ({input_task}, {output_task})
+
+        # Call _run_parallel
+        await coder._run_parallel(with_message="test message")
+
+        # Verify interrupt_event was cleared
+        coder.interrupt_event.clear.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_run_parallel_sets_running_flags():
+    """Test that _run_parallel sets running flags to True on start."""
+    # Create a test Coder instance
+    coder = create_test_coder()
+    coder.input_running = False
+    coder.output_running = False
+    coder.interrupt_event = MagicMock()
+    coder.interrupt_event.clear()
+
+    # Create mock tasks
+    input_task = AsyncMock()
+    output_task = AsyncMock()
+    input_task.return_value = "result"
+    output_task.return_value = "result"
+
+    # Mock run_one to return a simple value
+    coder.run_one = AsyncMock(return_value="test_result")
+
+    # Mock asyncio.wait to return when first task completes
+    with (
+        patch("asyncio.wait") as mock_wait,
+        patch("cecli.helpers.agents.service.AgentService") as mock_agent_service,
+    ):
+        mock_wait.return_value = ({input_task}, {output_task})
+        mock_instance = MagicMock()
+        mock_instance.foreground_coder = coder
+        mock_agent_service.get_instance.return_value = mock_instance
+
+        # Call _run_parallel
+        await coder._run_parallel(with_message="test message")
+
+        # Verify flags were reset to False in finally block
+        assert coder.input_running is False
+        assert coder.output_running is False
+
+        # Verify interrupt_event was cleared
+        assert not coder.interrupt_event.is_set()
+
+
+if __name__ == "__main__":
+    pytest.main([__file__])
diff --git a/tests/unit/test_worker_interrupt.py b/tests/unit/test_worker_interrupt.py
new file mode 100644
index 00000000000..45c888710ff
--- /dev/null
+++ b/tests/unit/test_worker_interrupt.py
@@ -0,0 +1,130 @@
+"""Unit tests for worker.interrupt() symmetric state reset."""
+
+import asyncio
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from cecli.tui.worker import CoderWorker
+from tests.fixtures.test_coder import create_test_coder
+
+
+def _make_target_coder():
+    """Create a MagicMock coder with all attributes needed by worker.interrupt()."""
+    coder = MagicMock()
+    coder.input_running = True
+    coder.output_running = True
+    coder.interrupt_event = asyncio.Event()
+    coder.io = MagicMock()
+    coder.io.output_task = MagicMock()
+    return coder
+
+
+def test_worker_interrupt_sets_both_flags_false():
+    """Verifies worker.interrupt() sets input_running and output_running to False."""
+    target_coder = _make_target_coder()
+    worker = CoderWorker(target_coder, MagicMock(), MagicMock())
+
+    with (
+        patch("cecli.helpers.agents.service.AgentService") as mock_agent_service,
+        patch(
+            "cecli.prompts.utils.registry.PromptRegistry.get_prompt",
+            return_value={"system": "dummy"},
+        ),
+    ):
+        # Configure AgentService mock to return our target_coder as foreground
+        mock_instance = MagicMock()
+        mock_instance.foreground_coder = target_coder
+        mock_agent_service.get_instance.return_value = mock_instance
+
+        worker.interrupt()
+
+    # Assert both flags are set to False
+    assert target_coder.input_running is False
+    assert target_coder.output_running is False
+    # Assert interrupt_event is set
+    assert target_coder.interrupt_event.is_set()
+
+
+def test_worker_interrupt_with_missing_input_running_attribute():
+    """Verifies that worker.interrupt() handles missing input_running attribute gracefully."""
+    # Setup test coder with proper prompt_format
+    target_coder = create_test_coder()
+
+    # Manually remove the attribute to simulate sub-agent scenario
+    if hasattr(target_coder, "input_running"):
+        delattr(target_coder, "input_running")
+
+    target_coder.output_running = True
+    target_coder.interrupt_event.clear()
+
+    # Create worker instance
+    worker = CoderWorker(target_coder, MagicMock(), MagicMock())
+
+    # Mock AgentService and PromptRegistry
+    with (
+        patch("cecli.helpers.agents.service.AgentService") as mock_agent_service,
+        patch(
+            "cecli.prompts.utils.registry.PromptRegistry.get_prompt",
+            return_value={"system": "dummy_prompt"},
+        ),
+    ):
+        mock_instance = MagicMock()
+        mock_instance.foreground_coder = target_coder
+        mock_agent_service.get_instance.return_value = mock_instance
+
+        # Call interrupt method - should not raise AttributeError
+        worker.interrupt()
+
+        # Assert output_running is still set to False
+        assert target_coder.output_running is False
+        # Assert interrupt_event is set
+        assert target_coder.interrupt_event.is_set()
+
+
+def test_worker_interrupt_cancels_output_task():
+    """Verifies worker.interrupt() cancels the output task."""
+    target_coder = _make_target_coder()
+    mock_output_task = AsyncMock()
+    target_coder.io.output_task = mock_output_task
+    worker = CoderWorker(target_coder, MagicMock(), MagicMock())
+
+    with (
+        patch("cecli.helpers.agents.service.AgentService") as mock_agent_service,
+        patch(
+            "cecli.prompts.utils.registry.PromptRegistry.get_prompt",
+            return_value={"system": "dummy"},
+        ),
+    ):
+        mock_instance = MagicMock()
+        mock_instance.foreground_coder = target_coder
+        mock_agent_service.get_instance.return_value = mock_instance
+
+        worker.interrupt()
+
+    mock_output_task.cancel.assert_called_once()
+
+
+def test_worker_interrupt_sets_interrupt_event():
+    """Verifies that worker.interrupt() sets the interrupt_event."""
+    target_coder = _make_target_coder()
+    worker = CoderWorker(target_coder, MagicMock(), MagicMock())
+
+    with (
+        patch("cecli.helpers.agents.service.AgentService") as mock_agent_service,
+        patch(
+            "cecli.prompts.utils.registry.PromptRegistry.get_prompt",
+            return_value={"system": "dummy_prompt"},
+        ),
+    ):
+        mock_instance = MagicMock()
+        mock_instance.foreground_coder = target_coder
+        mock_agent_service.get_instance.return_value = mock_instance
+
+        worker.interrupt()
+
+    assert target_coder.interrupt_event.is_set()
+
+
+if __name__ == "__main__":
+    pytest.main([__file__])