cli/interrupt: route SIGINT through a session handler

A SIGINT delivered while the event loop is parked in its selector is raised in
the loop's own frame: it escapes run_until_complete, the Runner cancels the main
task, and the REPL is left half-shut-down (no prompt, and the next line the
human types is eaten by an abandoned prompt reader) until the following Ctrl+C
kills the process. Off-loop prompts (ask_user_* / wait / tool confirms run in a
worker thread via asyncio.to_thread) and the auto-retry countdown are the usual
windows.

Keep one asyncio-level SIGINT handler installed for the whole session instead:
it cancels the innermost registered target (run_cancellable registers the turn
task, sleep_cancellable the retry countdown, off_loop_prompt the waiting prompt)
and never raises. Reads that block on the main thread inside a coroutine (the
REPL entry prompt, synchronous click prompts) suspend the router and take SIGINT
as a KeyboardInterrupt. Off-loop readline prompts cannot be interrupted without
abandoning their reader thread on stdin, so they ignore SIGINT with a one-line
hint.

src/plyngent/cli/app.py

@@ -17,7 +17,11 @@ from plyngent.cli.editor import (
     resolve_config_path,
 )
 from plyngent.cli.exit_codes import EXIT_CANCELLED, EXIT_OK, EXIT_TURN_FAILED
-from plyngent.cli.interrupt import install_keyboard_interrupt_sigint
+from plyngent.cli.interrupt import (
+    install_keyboard_interrupt_sigint,
+    install_sigint_router,
+    uninstall_sigint_router,
+)
 from plyngent.cli.limits import install_cli_limit_hooks
 from plyngent.cli.repl import run_repl
 from plyngent.cli.retry import run_user_text_with_retries
@@ -251,6 +255,11 @@ async def _run_chat(  # noqa: C901, PLR0912, PLR0915 — chat orchestration
     yolo: YoloMode | None = "on" if yes else None
 
     memory = await MemoryStore.open(_database_config(store, quiet=quiet or oneshot))
+    # A SIGINT must never surface as a KeyboardInterrupt raised in the event
+    # loop's own frame (it would tear this run — and the shutdown below — down
+    # half-finished). The router cancels the innermost registered target instead
+    # and stays installed through cleanup.
+    _ = install_sigint_router()
     try:
         from plyngent.cli.provider_recovery import ensure_provider_ready
 
@@ -402,6 +411,7 @@ async def _run_chat(  # noqa: C901, PLR0912, PLR0915 — chat orchestration
             from plyngent.tools.process.pty_session import PtyManager
 
             PtyManager.close_all()
+        uninstall_sigint_router()
 
 
 def _configure_logging(level: str) -> None:

src/plyngent/cli/interrupt.py

@@ -4,7 +4,10 @@ import asyncio
 import contextlib
 import signal
 from contextlib import contextmanager
-from typing import TYPE_CHECKING
+from dataclasses import dataclass, field
+from typing import TYPE_CHECKING, Protocol
+
+import click
 
 if TYPE_CHECKING:
     from collections.abc import Callable, Generator
@@ -12,21 +15,17 @@ if TYPE_CHECKING:
 
     type SigHandler = Callable[[int, FrameType | None], None] | int | None
 
-# Depth of nested pause_task_cancel_for_prompt. Must be a plain int (not ContextVar):
-# asyncio.add_signal_handler freezes the ContextVar snapshot at install time, so a
-# ContextVar reset after reinstall would leave SIGINT permanently non-cancelling.
-_prompt_depth: int = 0
-_reinstall_holder: list[Callable[[], None] | None] = [None]
 
+class InterruptTarget(Protocol):
+    """Something a pending SIGINT can cancel (a task, a prompt read, a hint)."""
 
-def allow_task_cancel() -> bool:
+    def cancel(self) -> object:
-    """Whether the CLI SIGINT handler should cancel the in-flight turn task."""
+        """Cancel the pending work; the return value is ignored.
-    return _prompt_depth == 0
 
-
+        Declared loosely (``object``) so ``asyncio.Task.cancel`` — which returns
-def set_sigint_reinstall(callback: Callable[[], None] | None) -> None:
+        ``bool`` and takes an optional message — satisfies it.
-    """Register how to re-bind SIGINT after a TTY prompt (set by run_cancellable)."""
+        """
-    _reinstall_holder[0] = callback
+        ...
 
 
 def raise_keyboard_interrupt_sigint(signum: int, frame: FrameType | None) -> None:
@@ -56,57 +55,172 @@ def install_keyboard_interrupt_sigint() -> None:
         _ = signal.signal(signal.SIGINT, raise_keyboard_interrupt_sigint)
 
 
+# --------------------------------------------------------------------------- #
+# SIGINT routing (session scope)
+# --------------------------------------------------------------------------- #
+
+
+@dataclass(slots=True)
+class _RouterState:
+    """Session SIGINT routing state (one instance, kept in ``_router``)."""
+
+    # Cancel targets, innermost last: a SIGINT cancels the innermost one.
+    targets: list[InterruptTarget] = field(default_factory=list)
+    installed: bool = False
+    # Nesting depth of main-thread blocking reads. A plain counter (not a
+    # ContextVar): asyncio.add_signal_handler freezes the ContextVar snapshot at
+    # install time, so a ContextVar reset after reinstall would leave SIGINT
+    # permanently routed to the wrong place.
+    blocking_reads: int = 0
+
+
+_router = _RouterState()
+
+
+def sigint_routed() -> bool:
+    """Whether the session SIGINT router currently owns SIGINT."""
+    return _router.installed
+
+
+def install_sigint_router() -> bool:
+    """Install the session SIGINT handler; False when the loop cannot take it.
+
+    A SIGINT delivered while the event loop is parked in its selector is raised
+    in the *loop's* frame: it escapes ``run_until_complete``, the Runner cancels
+    the main task and the REPL (or a half-finished shutdown) is left broken —
+    the next Ctrl+C then kills the process. Keeps an asyncio-level SIGINT
+    handler installed for the whole session instead: it cancels the innermost
+    target registered with :func:`sigint_cancels` and never raises.
+
+    Blocking main-thread reads suspend the router via
+    :func:`pause_task_cancel_for_prompt`. Platforms without asyncio signal
+    handlers (Windows proactor) return False and keep the plain
+    KeyboardInterrupt behaviour. Idempotent.
+    """
+    if _router.installed:
+        return True
+    if not _add_router_handler():
+        return False
+    _router.installed = True
+    return True
+
+
+def uninstall_sigint_router() -> None:
+    """Drop the session SIGINT handler and restore the KeyboardInterrupt one."""
+    if not _router.installed:
+        return
+    _router.installed = False
+    _remove_router_handler()
+    _router.targets.clear()
+    install_keyboard_interrupt_sigint()
+
+
+def _add_router_handler() -> bool:
+    try:
+        asyncio.get_running_loop().add_signal_handler(signal.SIGINT, _route_sigint)
+    except RuntimeError, NotImplementedError, ValueError:
+        return False
+    return True
+
+
+def _remove_router_handler() -> None:
+    with contextlib.suppress(NotImplementedError, RuntimeError, ValueError):
+        _ = asyncio.get_running_loop().remove_signal_handler(signal.SIGINT)
+
+
+def _route_sigint() -> None:
+    """Cancel the innermost target (none → ignore: never escape the loop)."""
+    if _router.targets:
+        _ = _router.targets[-1].cancel()
+
+
+@contextmanager
+def sigint_cancels(target: InterruptTarget) -> Generator[None]:
+    """Route SIGINT to ``target.cancel()`` while inside (innermost wins)."""
+    _router.targets.append(target)
+    try:
+        yield
+    finally:
+        for index in range(len(_router.targets) - 1, -1, -1):
+            if _router.targets[index] is target:
+                del _router.targets[index]
+                break
+
+
 @contextmanager
 def pause_task_cancel_for_prompt() -> Generator[None]:
-    """Disable turn-task cancel during blocking TTY prompts (confirm, etc.).
+    """Give SIGINT to a blocking *main-thread* read instead of routing it.
 
-    Must run on the main thread (signal handlers are main-thread only).
+    The read (REPL entry prompt, synchronous click confirm/prompt) runs on the
-    Restores the default SIGINT handler so ``click.confirm`` can receive
+    main thread inside a coroutine, so the resulting KeyboardInterrupt is raised
-    KeyboardInterrupt / Abort instead of the asyncio turn being cancelled.
+    in the reader's own frame: its handlers see it, and neither the in-flight
+    turn nor the event loop is cancelled. Use it only around blocking reads — an
+    ``await`` that parks the loop in its selector must keep the router installed
+    (:func:`install_sigint_router`).
 
-    Nested prompts are supported via a depth counter. The asyncio SIGINT
+    Suspends the router; nested calls restore it when the outermost exits.
-    handler is reinstalled only after depth returns to 0, and only after
-    cancel is re-enabled, so the handler never freezes ``allow=False``.
     """
-
+    _router.blocking_reads += 1
-    global _prompt_depth  # noqa: PLW0603
+    resuming = _router.blocking_reads == 1 and _router.installed
-    _prompt_depth += 1
+    if resuming:
-    loop_handler_removed = False
+        _remove_router_handler()
     previous: SigHandler = signal.SIG_DFL
-    try:
-        try:
-            loop = asyncio.get_running_loop()
-            _ = loop.remove_signal_handler(signal.SIGINT)
-            loop_handler_removed = True
-        except RuntimeError, NotImplementedError, ValueError:
-            loop_handler_removed = False
-
     try:
         previous = signal.getsignal(signal.SIGINT)
-            _ = signal.signal(signal.SIGINT, signal.default_int_handler)
+        _ = signal.signal(signal.SIGINT, raise_keyboard_interrupt_sigint)
     except ValueError:
-            # Not on the main thread — skip OS signal rebinding.
+        # Not on the main thread — signal handlers cannot be rebound here.
         previous = signal.SIG_DFL
+    try:
         yield
     finally:
         with contextlib.suppress(ValueError):
             _ = signal.signal(signal.SIGINT, previous)
-        _prompt_depth = max(0, _prompt_depth - 1)
+        _router.blocking_reads = max(0, _router.blocking_reads - 1)
-        # Reinstall only when fully out of prompts and cancel is allowed again.
+        if resuming:
-        # Order matters: decrement depth first so allow_task_cancel() is True
+            _ = _add_router_handler()
-        # before reinstall (and handlers never freeze allow=False).
-        if _prompt_depth == 0 and loop_handler_removed:
-            reinstall = _reinstall_holder[0]
-            if reinstall is not None:
-                with contextlib.suppress(RuntimeError, NotImplementedError, ValueError):
-                    reinstall()
 
 
-async def run_in_prompt_thread[**P, R](func: Callable[P, R], *args: P.args, **kwargs: P.kwargs) -> R:
+@dataclass(frozen=True, slots=True)
-    """Run a blocking TTY prompt off the event loop with cancel paused.
+class _CancelHook:
+    """SIGINT target for a poll-based prompt read (abort that read)."""
 
-    Pause/SIGINT rebinding happens on the main thread; only the prompt body
+    cancel_read: Callable[[], None]
-    runs in a worker so the asyncio loop stays free and turn cancel is disabled.
+
+    def cancel(self) -> None:
+        self.cancel_read()
+
+
+@dataclass(slots=True)
+class _WaitingPrompt:
+    """SIGINT target for a prompt read that cannot be interrupted.
+
+    A readline read blocks inside the C library on a worker thread; aborting it
+    would abandon that thread on stdin, where it would eat the next line the
+    human types. Explaining itself once is the best available answer.
     """
-    with pause_task_cancel_for_prompt():
+
-        return await asyncio.to_thread(func, *args, **kwargs)
+    _hinted: bool = False
+
+    def cancel(self) -> None:
+        if self._hinted:
+            return
+        self._hinted = True
+        click.secho(
+            "Ctrl+C ignored while a prompt is waiting — answer it to continue",
+            fg="yellow",
+            err=True,
+        )
+
+
+@contextmanager
+def off_loop_prompt(cancel: Callable[[], None] | None = None) -> Generator[None]:
+    """A human prompt is waiting in a worker thread.
+
+    While inside, SIGINT must not cancel the enclosing turn: poll-based reads
+    (``cancel`` hook) abort just that read, readline-based ones ignore it with a
+    one-line hint (see :class:`_WaitingPrompt`).
+    """
+    target: InterruptTarget = _WaitingPrompt() if cancel is None else _CancelHook(cancel)
+    with sigint_cancels(target):
+        yield

src/plyngent/cli/limits.py

@@ -7,7 +7,7 @@ import sys
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Literal
 
-from plyngent.cli.interrupt import pause_task_cancel_for_prompt
+from plyngent.cli.interrupt import off_loop_prompt, pause_task_cancel_for_prompt
 from plyngent.prompting import (
     ChoiceOption,
     NonInteractiveError,
@@ -521,10 +521,14 @@ async def prompt_workspace_mismatch_async(
 
 
 def install_cli_limit_hooks() -> None:
-    """Register interactive continue hooks and prompt cancel-pause.
+    """Register interactive continue hooks and prompt guards.
 
-    Command denylist policy confirm is set on :class:`~plyngent.tools.context.InstanceState`
+    Off-loop prompts (worker-thread reads) register their own SIGINT target so a
-    when the REPL starts (no process-global hook).
+    Ctrl+C there never cancels the enclosing turn; synchronous prompts read on
+    the main thread and take SIGINT as a KeyboardInterrupt instead. Command
+    denylist policy confirm is set on
+    :class:`~plyngent.tools.context.InstanceState` when the REPL starts (no
+    process-global hook).
     """
-    configure_prompting(pause_factory=pause_task_cancel_for_prompt)
+    configure_prompting(pause_factory=off_loop_prompt)
     PtyManager.set_limit_continue_hook(prompt_continue_limit)

src/plyngent/cli/repl.py

@@ -6,6 +6,7 @@ from typing import TYPE_CHECKING
 import click
 
 from plyngent.cli.input_text import read_repl_entry
+from plyngent.cli.interrupt import pause_task_cancel_for_prompt
 from plyngent.cli.readline_setup import setup_readline
 from plyngent.cli.retry import run_user_text_with_retries
 from plyngent.cli.slash import handle_slash
@@ -25,7 +26,7 @@ def _echo_user(text: str) -> None:
 
 def _echo_interrupted() -> None:
     # A further Ctrl+C during this output must stay benign (same rule as
-    # retry._echo_cancel_lines): after a turn the SIGINT handler is removed, so
+    # retry._echo_cancel_lines): on platforms without asyncio signal handlers
     # the next Ctrl+C raises KeyboardInterrupt at an arbitrary bytecode.
     with contextlib.suppress(KeyboardInterrupt):
         click.echo()
@@ -51,16 +52,22 @@ async def run_repl(state: ReplState) -> None:
     click.echo('Type /help for commands. Multiline: """ … """. Empty line is ignored.')
 
     while True:
+        # The prompt read blocks the main thread inside this coroutine, so hand
+        # SIGINT to it: Ctrl+C raises KeyboardInterrupt here (caught below and
+        # inside read_repl_entry) and re-prompts. Everywhere else the session
+        # SIGINT router owns the signal (turn cancel, prompt cancel).
         try:
+            with pause_task_cancel_for_prompt():
                 entry = read_repl_entry()
         except EOFError:
             click.echo()
             break
+        except KeyboardInterrupt:
+            # Stray interrupt outside the read (e.g. right after the router was
+            # suspended): re-prompt rather than leave the REPL.
+            _echo_interrupted()
+            continue
 
-        # Between turns the turn-task SIGINT handler is removed (run_cancellable),
-        # so a stray Ctrl+C raises KeyboardInterrupt at whatever bytecode is
-        # executing — echoing user text, expire_yolo_once, gaps between
-        # statements. Swallow it and re-prompt instead of exiting the REPL.
         try:
             if entry is None:
                 continue

src/plyngent/cli/retry.py

@@ -2,7 +2,6 @@ from __future__ import annotations
 
 import asyncio
 import contextlib
-import signal
 import time
 from typing import TYPE_CHECKING
 
@@ -10,7 +9,7 @@ import click
 
 from plyngent.agent import AssistantMessageEvent
 from plyngent.cli.display import render_events
-from plyngent.cli.interrupt import allow_task_cancel, set_sigint_reinstall
+from plyngent.cli.interrupt import install_sigint_router, sigint_cancels
 from plyngent.cli.limits import reset_auto_continue_turn
 
 if TYPE_CHECKING:
@@ -47,7 +46,9 @@ async def sleep_cancellable(seconds: float) -> bool:
             remaining = deadline - time.monotonic()
             if remaining <= 0:
                 return True
-            await asyncio.sleep(min(0.5, remaining))
+            sleeper = asyncio.ensure_future(asyncio.sleep(min(0.5, remaining)))
+            with sigint_cancels(sleeper):
+                await sleeper
     except asyncio.CancelledError:
         return False
     except KeyboardInterrupt:
@@ -57,56 +58,32 @@ async def sleep_cancellable(seconds: float) -> bool:
 async def run_cancellable[T](coro: Coroutine[object, object, T]) -> T:
     """Await ``coro`` as a task; Ctrl+C / SIGINT cancels the task.
 
-    Interactive prompts (max-rounds / destructive confirm) temporarily disable
+    Interactive prompts (max-rounds / destructive confirm) register their own
-    task cancel so the user can answer instead of aborting the whole turn.
+    SIGINT target for as long as they wait, so the turn is not cancelled while
+    the human is answering (see :func:`~plyngent.cli.interrupt.off_loop_prompt`).
 
     Raises:
         asyncio.CancelledError: If the task was cancelled (including via SIGINT).
     """
     task: asyncio.Task[T] = asyncio.create_task(coro)
-    loop = asyncio.get_running_loop()
+    # Keep a loop-level SIGINT handler in place for the rest of the session, so
-    installed = False
+    # a Ctrl+C that lands while the loop is parked in its selector cancels this
-    # What to restore once the turn-cancel handler is removed. remove_signal_handler
+    # task instead of being raised in the loop's own frame.
-    # leaves SIG_DFL, which would terminate the process on a stray Ctrl+C between
+    _ = install_sigint_router()
-    # turns; restore the handler from before (the CLI installs a
-    # KeyboardInterrupt-raising one so the REPL can catch it).
-    previous_sigint = signal.getsignal(signal.SIGINT)
-
-    def _on_sigint() -> None:
-        # allow_task_cancel() uses a process-level depth counter (not ContextVar)
-        # so this remains correct even if the handler was installed under a
-        # frozen context (asyncio signal handles capture contextvars).
-        if allow_task_cancel() and not task.done():
-            _ = task.cancel()
-
-    def _reinstall() -> None:
     try:
-            loop.add_signal_handler(signal.SIGINT, _on_sigint)
+        with sigint_cancels(task):
-        except NotImplementedError, RuntimeError, ValueError:
-            return
-
-    try:
-        loop.add_signal_handler(signal.SIGINT, _on_sigint)
-        installed = True
-        set_sigint_reinstall(_reinstall)
-    except NotImplementedError, RuntimeError, ValueError:
-        installed = False
-
             try:
                 return await task
             except KeyboardInterrupt:
+                # Platforms without asyncio signal handlers (Windows proactor)
+                # keep the CLI's KeyboardInterrupt handler, so the interrupt
+                # surfaces here while the task is still awaiting.
                 if not task.done():
                     _ = task.cancel()
                     with contextlib.suppress(asyncio.CancelledError):
                         await task
                 raise asyncio.CancelledError from None
     finally:
-        set_sigint_reinstall(None)
-        if installed:
-            with contextlib.suppress(NotImplementedError, RuntimeError, ValueError):
-                _ = loop.remove_signal_handler(signal.SIGINT)
-            with contextlib.suppress(ValueError):
-                _ = signal.signal(signal.SIGINT, previous_sigint)
         # Only cancel if still running (e.g. KeyboardInterrupt path above).
         # Do not cancel a finished task — that would mask success.
         if not task.done():
@@ -143,9 +120,10 @@ def _echo_turn_usage(agent: ChatAgent) -> None:
 def _echo_cancel_lines(*lines: str) -> None:
     """Print post-cancel lines, treating a further Ctrl+C as benign.
 
-    After a turn-cancel the asyncio SIGINT handler is removed, so a second
+    The session SIGINT router keeps the loop safe between turns, but platforms
-    Ctrl+C during this output would otherwise raise KeyboardInterrupt and
+    without asyncio signal handlers (Windows proactor) deliver a second Ctrl+C
-    exit the REPL instead of returning to the prompt.
+    here as a KeyboardInterrupt, which must not exit the REPL instead of
+    returning to the prompt.
     """
     try:
         for line in lines:

tests/test_cli/test_interrupt.py

@@ -1,124 +1,67 @@
 from __future__ import annotations
 
 import asyncio
+import os
 import signal
+import threading
+import time
 from typing import TYPE_CHECKING
 
 import pytest
 
 from plyngent.cli.interrupt import (
-    allow_task_cancel,
     install_keyboard_interrupt_sigint,
+    install_sigint_router,
+    off_loop_prompt,
     pause_task_cancel_for_prompt,
     raise_keyboard_interrupt_sigint,
-    run_in_prompt_thread,
+    sigint_cancels,
-    set_sigint_reinstall,
+    sigint_routed,
+    uninstall_sigint_router,
 )
-from plyngent.cli.limits import prompt_continue_limit, prompt_continue_limit_async
+from plyngent.cli.limits import prompt_continue_limit
-from plyngent.cli.retry import run_cancellable
+from plyngent.cli.retry import run_cancellable, sleep_cancellable
 from plyngent.prompting import temporary_backend
 from tests.test_prompting import ScriptedBackend
 
 if TYPE_CHECKING:
-    pass
+    from collections.abc import AsyncIterator
+
+    import pytest as pytest_module
 
 
-def test_pause_task_cancel_for_prompt() -> None:
+class FakeTarget:
-    assert allow_task_cancel() is True
+    """Minimal ``InterruptTarget`` recording its cancels."""
-    with pause_task_cancel_for_prompt():
+
-        assert allow_task_cancel() is False
+    def __init__(self, log: list[str], name: str = "cancel") -> None:
-    assert allow_task_cancel() is True
+        self.log = log
+        self.name = name
+
+    def cancel(self) -> None:
+        self.log.append(self.name)
 
 
-def test_nested_pause_depth() -> None:
+@pytest.fixture
-    assert allow_task_cancel() is True
+async def router() -> AsyncIterator[None]:
-    with pause_task_cancel_for_prompt():
+    """Install the session SIGINT router; restore the CLI handler afterwards."""
-        assert allow_task_cancel() is False
+    install_keyboard_interrupt_sigint()
-        with pause_task_cancel_for_prompt():
+    _ = install_sigint_router()
-            assert allow_task_cancel() is False
+    try:
-        assert allow_task_cancel() is False
+        yield
-    assert allow_task_cancel() is True
+    finally:
+        uninstall_sigint_router()
+        signal.signal(signal.SIGINT, signal.default_int_handler)
 
 
-def test_prompt_continue_limit_under_pause(monkeypatch: pytest.MonkeyPatch) -> None:
+async def _deliver_sigint() -> None:
-    backend = ScriptedBackend(["y"])
+    """Send SIGINT to this process and let the loop run its signal callback.
 
-    def _read_line(prompt: str, *, default: str | None = None, completions: object = None) -> str:
+    asyncio's signal wakeup is only noticed while the loop parks in its
-        del prompt, default, completions
+    selector, so a real (short) wait is needed — ``sleep(0)`` never parks and
-        assert allow_task_cancel() is False
+    the callback would not run at all.
-        return "y"
-
-    monkeypatch.setattr(backend, "read_line", _read_line)
-    with temporary_backend(backend):
-        assert prompt_continue_limit("too many rounds") is True
-
-
-async def test_run_in_prompt_thread_pauses_cancel() -> None:
-    """Cancel is paused on the main thread for the whole to_thread call."""
-    assert allow_task_cancel() is True
-
-    def work() -> str:
-        return "ok"
-
-    result = await run_in_prompt_thread(work)
-    assert result == "ok"
-    assert allow_task_cancel() is True
-
-
-async def test_prompt_continue_limit_async(monkeypatch: pytest.MonkeyPatch) -> None:
-    del monkeypatch
-    with temporary_backend(ScriptedBackend(["y"])):
-        assert await prompt_continue_limit_async("too many rounds") is True
-
-
-async def test_sigint_cancels_after_prompt_pause() -> None:
-    """Regression: after a mid-turn prompt, SIGINT must still cancel the turn.
-
-    Previously, reinstalling the asyncio SIGINT handler while
-    allow_task_cancel was still False froze that value into the callback
-    forever (ContextVar snapshot at add_signal_handler).
     """
-    loop = asyncio.get_running_loop()
-    try:
-        loop.add_signal_handler(signal.SIGINT, lambda: None)
-        loop.remove_signal_handler(signal.SIGINT)
-    except NotImplementedError, RuntimeError, ValueError:
-        pytest.skip("asyncio signal handlers not available on this platform")
-
-    started = asyncio.Event()
-    cancelled = asyncio.Event()
-
-    async def hang() -> None:
-        started.set()
-        try:
-            await asyncio.sleep(3600)
-        except asyncio.CancelledError:
-            cancelled.set()
-            raise
-
-    async def turn() -> None:
-        # Simulate soft-confirm pause mid-turn, then keep streaming.
-        with pause_task_cancel_for_prompt():
-            assert allow_task_cancel() is False
-        assert allow_task_cancel() is True
-        await hang()
-
-    task = asyncio.create_task(run_cancellable(turn()))
-    await started.wait()
-    # Give run_cancellable a tick to install the handler after the pause reinstall.
-    await asyncio.sleep(0)
-    assert allow_task_cancel() is True
-    # Deliver SIGINT the same way Ctrl+C does under asyncio.
-    loop.call_soon(lambda: None)  # ensure loop is processing
-    # Invoke the process signal: raise SIGINT to this process.
-    import os
-
     os.kill(os.getpid(), signal.SIGINT)
-    with pytest.raises(asyncio.CancelledError):
+    await asyncio.sleep(0.05)
-        await task
-    assert cancelled.is_set()
-    # Cleanup any leftover reinstall hook from run_cancellable
-    set_sigint_reinstall(None)
 
 
 def test_install_keyboard_interrupt_sigint() -> None:
@@ -148,8 +91,6 @@ def test_asyncio_run_keeps_keyboard_interrupt_sigint() -> None:
     KeyboardInterrupt, which also cancelled ``memory.close()`` on a Ctrl+D exit.
     With a non-default handler installed first, Runner leaves SIGINT alone.
     """
-    import asyncio
-
     install_keyboard_interrupt_sigint()
     try:
 
@@ -162,28 +103,180 @@ def test_asyncio_run_keeps_keyboard_interrupt_sigint() -> None:
         signal.signal(signal.SIGINT, signal.default_int_handler)
 
 
-async def test_run_cancellable_restores_previous_sigint() -> None:
+async def test_sigint_cancels_registered_target(router: None) -> None:
-    """Regression: after a turn SIGINT is restored, not left as SIG_DFL.
+    """SIGINT cancels the target registered around the awaited turn task."""
+    del router
+    cancelled = asyncio.Event()
+
+    async def hang() -> None:
+        try:
+            await asyncio.sleep(3600)
+        except asyncio.CancelledError:
+            cancelled.set()
+            raise
+
+    task = asyncio.create_task(hang())
+    await asyncio.sleep(0)
+    with sigint_cancels(task):
+        os.kill(os.getpid(), signal.SIGINT)
+        # Waiting on the effect (not on a fixed sleep) keeps this deterministic.
+        with pytest.raises(asyncio.CancelledError):
+            await asyncio.wait_for(task, timeout=5)
+    assert cancelled.is_set()
+
+
+async def test_sigint_while_loop_is_idle_does_not_escape(router: None) -> None:
+    """Regression: a stray Ctrl+C while the loop is parked in its selector.
+
+    The interrupt used to be raised in the loop's own frame (not inside any
+    coroutine), escaping ``run_until_complete``: the Runner cancelled the main
+    task and the REPL was left half-shut-down, so the next Ctrl+C killed the
+    process instead of returning to the prompt.
+    """
+    del router
+
+    def _send() -> None:
+        time.sleep(0.05)
+        os.kill(os.getpid(), signal.SIGINT)
+
+    sender = threading.Thread(target=_send)
+    sender.start()
+    try:
+        await asyncio.sleep(0.2)  # parked in the selector when SIGINT lands
+    finally:
+        sender.join()
+    task = asyncio.current_task()
+    assert task is not None
+    assert task.cancelling() == 0
+    assert not task.cancelled()
+
+
+async def test_sigint_cancels_innermost_target_only(router: None) -> None:
+    """Nested targets: the innermost (the active turn or prompt) wins."""
+    del router
+    outer: list[str] = []
+    inner: list[str] = []
+    with sigint_cancels(FakeTarget(outer, "outer")):
+        await _deliver_sigint()
+        assert outer == ["outer"]
+        with sigint_cancels(FakeTarget(inner, "inner")):
+            await _deliver_sigint()
+            assert inner == ["inner"]
+        assert outer == ["outer"]  # the nested signal never reached it
+
+
+async def test_sigint_target_stops_after_context_exit(router: None) -> None:
+    """A popped target no longer sees SIGINT (and an empty stack ignores it)."""
+    del router
+    calls: list[str] = []
+    with sigint_cancels(FakeTarget(calls)):
+        await _deliver_sigint()
+    assert calls == ["cancel"]
+    await _deliver_sigint()
+    assert calls == ["cancel"]
+
+
+async def test_off_loop_prompt_keeps_turn_alive(
+    router: None,
+    capsys: pytest_module.CaptureFixture[str],
+) -> None:
+    """A prompt's SIGINT target shields the turn; afterwards the turn cancels.
+
+    Regression (badcb3c, revisited): the router resolves the innermost target
+    per signal instead of freezing cancel state into the handler, so a mid-turn
+    prompt neither aborts the turn nor freezes turn-cancel.
+    """
+    del router
+    started = asyncio.Event()
+    cancelled = asyncio.Event()
+
+    async def hang() -> None:
+        started.set()
+        try:
+            await asyncio.sleep(3600)
+        except asyncio.CancelledError:
+            cancelled.set()
+            raise
+
+    task = asyncio.create_task(hang())
+    await started.wait()
+    with sigint_cancels(task):
+        with off_loop_prompt():
+            await _deliver_sigint()
+            assert not task.cancelling()
+        err = capsys.readouterr().err
+        assert "Ctrl+C ignored while a prompt is waiting" in err
+        # The prompt returned (outer context still active): SIGINT cancels now.
+        os.kill(os.getpid(), signal.SIGINT)
+        with pytest.raises(asyncio.CancelledError):
+            await asyncio.wait_for(task, timeout=5)
+    assert cancelled.is_set()
+
+
+async def test_off_loop_prompt_cancel_hook_is_preferred(router: None) -> None:
+    """A poll-based prompt read gets cancelled instead of the enclosing turn."""
+    del router
+    hooks: list[str] = []
+    turn_calls: list[str] = []
+    with sigint_cancels(FakeTarget(turn_calls, "turn")), off_loop_prompt(lambda: hooks.append("read")):
+        await _deliver_sigint()
+    assert hooks == ["read"]
+    assert turn_calls == []
+
+
+async def test_pause_task_cancel_for_prompt_swaps_sigint(router: None) -> None:
+    """A main-thread blocking read takes SIGINT as a KeyboardInterrupt."""
+    del router
+    routed = signal.getsignal(signal.SIGINT)
+    assert routed is not raise_keyboard_interrupt_sigint
+    with pause_task_cancel_for_prompt():
+        assert signal.getsignal(signal.SIGINT) is raise_keyboard_interrupt_sigint
+        with pause_task_cancel_for_prompt():
+            assert signal.getsignal(signal.SIGINT) is raise_keyboard_interrupt_sigint
+        # Inner exit must not resume the router while the outer read blocks.
+        assert signal.getsignal(signal.SIGINT) is raise_keyboard_interrupt_sigint
+    assert signal.getsignal(signal.SIGINT) is routed
+
+
+async def test_prompt_continue_limit_under_pause(
+    router: None,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Synchronous prompts read on the main thread, where SIGINT is a KI."""
+    del router
+    backend = ScriptedBackend(["y"])
+
+    def _read_line(prompt: str, *, default: str | None = None, completions: object = None) -> str:
+        del prompt, default, completions
+        assert signal.getsignal(signal.SIGINT) is raise_keyboard_interrupt_sigint
+        return "y"
+
+    monkeypatch.setattr(backend, "read_line", _read_line)
+    with temporary_backend(backend):
+        assert prompt_continue_limit("too many rounds") is True
+
+
+async def test_run_cancellable_keeps_router_installed(router: None) -> None:
+    """Regression: after a turn SIGINT must not be left as SIG_DFL.
 
     ``loop.remove_signal_handler`` leaves SIG_DFL, which would terminate the
-    process on a stray Ctrl+C between turns instead of letting the REPL catch
+    process on a stray Ctrl+C between turns; the session router stays installed.
-    a KeyboardInterrupt.
     """
-    loop = asyncio.get_running_loop()
+    del router
-    try:
-        loop.add_signal_handler(signal.SIGINT, lambda: None)
-        loop.remove_signal_handler(signal.SIGINT)
-    except NotImplementedError, RuntimeError, ValueError:
-        pytest.skip("asyncio signal handlers not available on this platform")
-
-    previous = signal.getsignal(signal.SIGINT)
 
     async def quick() -> None:
         return None
 
-    try:
     await run_cancellable(quick())
-        assert signal.getsignal(signal.SIGINT) is previous
+    assert sigint_routed()
     assert signal.getsignal(signal.SIGINT) is not signal.SIG_DFL
-    finally:
+    assert signal.getsignal(signal.SIGINT) is not signal.default_int_handler
-        signal.signal(signal.SIGINT, signal.default_int_handler)
+
+
+async def test_sleep_cancellable_cancelled_by_sigint(router: None) -> None:
+    """The auto-retry countdown is cancelled by Ctrl+C (was a teardown window)."""
+    del router
+    waiter = asyncio.create_task(sleep_cancellable(30))
+    await asyncio.sleep(0)
+    os.kill(os.getpid(), signal.SIGINT)
+    assert await asyncio.wait_for(waiter, timeout=5) is False