version bump, interrupt exception handling/suppress

Author: evalstate

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "fast-agent-mcp"
-version = "0.6.6"
+version = "0.6.7"
 description = "Define, Prompt and Test MCP enabled Agents and Workflows"
 readme = "README.md"
 license = { file = "LICENSE" }

src/fast_agent/cli/__main__.py

@@ -125,7 +125,10 @@ def main():
             # Auto-route to go command
             sys.argv.insert(insert_pos, "go")
 
-    app()
+    try:
+        app()
+    except KeyboardInterrupt:
+        raise SystemExit(130) from None
 
 
 if __name__ == "__main__":

src/fast_agent/cli/runtime/runner.py

@@ -4,6 +4,7 @@
 
 import asyncio
 import sys
+from contextlib import suppress
 from typing import TYPE_CHECKING
 
 from fast_agent.cli.asyncio_utils import set_asyncio_exception_handler
@@ -63,21 +64,27 @@ def run_request(request: AgentRunRequest) -> None:
         write_interactive_trace("cli.runner.system_exit", code=exc.code)
         exit_code = exc.code if isinstance(exc.code, int) else None
     finally:
-        try:
+        with suppress(BaseException):
             if not main_task.done():
                 loop.run_until_complete(asyncio.gather(main_task, return_exceptions=True))
 
+        tasks = set()
+        with suppress(BaseException):
             tasks = {task for task in asyncio.all_tasks(loop) if task is not main_task}
-            write_interactive_trace("cli.runner.finally", task_count=len(tasks))
-            for task in tasks:
+        write_interactive_trace("cli.runner.finally", task_count=len(tasks))
+
+        for task in tasks:
+            with suppress(BaseException):
                 task.cancel()
 
-            if sys.version_info >= (3, 7):
+        if sys.version_info >= (3, 7) and tasks:
+            with suppress(BaseException):
                 loop.run_until_complete(asyncio.gather(*tasks, return_exceptions=True))
+
+        with suppress(BaseException):
             loop.run_until_complete(loop.shutdown_asyncgens())
+        with suppress(BaseException):
             loop.close()
-        except Exception:
-            pass
 
     if exit_code not in (None, 0):
         raise SystemExit(exit_code)

src/fast_agent/ui/prompt/input_runtime.py

@@ -14,6 +14,7 @@
 from rich import print as rich_print
 
 from fast_agent.ui.command_payloads import CommandPayload, InterruptCommand
+from fast_agent.ui.prompt.keybindings import PromptInputInterrupt
 from fast_agent.ui.prompt_marks import emit_prompt_mark
 from fast_agent.utils.async_utils import suppress_known_runtime_warnings
 
@@ -156,7 +157,7 @@ def _track_accept(buffer_obj) -> bool:
             rich_print(f"[dim]{prompt_prefix} {stripped.splitlines()[0]}[/dim]")
 
         return parse_special_input(result)
-    except KeyboardInterrupt:
+    except (KeyboardInterrupt, PromptInputInterrupt):
         if prompt_mark_started:
             emit_prompt_mark("B")
         return InterruptCommand()

src/fast_agent/ui/prompt/keybindings.py

@@ -42,6 +42,10 @@ class AgentKeyBindings(KeyBindings):
     current_agent_name: str | None = None
 
 
+class PromptInputInterrupt(Exception):
+    """Internal prompt-toolkit interrupt used instead of raw KeyboardInterrupt."""
+
+
 def _cycle_completion(buffer: Buffer, *, backwards: bool) -> bool:
     """Cycle through current completion menu items.
 
@@ -243,7 +247,7 @@ def _(event) -> None:
     @kb.add("c-c")
     def _(event) -> None:
         """Ctrl+C: interrupt prompt input (handled by caller policy)."""
-        event.app.exit(exception=KeyboardInterrupt())
+        event.app.exit(exception=PromptInputInterrupt())
 
     @kb.add("c-d")
     def _(event) -> None:

tests/unit/fast_agent/commands/test_cli_main_routing.py

@@ -6,6 +6,7 @@
 import pytest
 from click.utils import strip_ansi
 
+from fast_agent.cli import __main__ as cli_main
 from fast_agent.cli.__main__ import _first_positional_argument
 
 
@@ -78,3 +79,12 @@ def test_demo_subcommand_still_detected_after_env_option_value() -> None:
     assert result.returncode == 0, result.stderr
     assert "demo [OPTIONS] COMMAND" in output
     assert "Demo commands for UI features." in output
+
+
+def test_main_converts_keyboard_interrupt_to_clean_exit(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(cli_main, "app", lambda: (_ for _ in ()).throw(KeyboardInterrupt()))
+
+    with pytest.raises(SystemExit) as exc_info:
+        cli_main.main()
+
+    assert exc_info.value.code == 130

tests/unit/fast_agent/commands/test_runtime_result_export.py

@@ -12,6 +12,7 @@
 from mcp.types import CallToolRequestParams, TextContent
 
 from fast_agent.agents.agent_types import AgentConfig
+from fast_agent.cli.runtime import runner
 from fast_agent.cli.runtime.agent_setup import (
     _apply_shell_cwd_policy_preflight,
     _build_fan_out_result_paths,
@@ -130,6 +131,69 @@ def test_should_convert_keyboard_interrupt_to_task_cancel_only_for_interactive_r
     assert _should_convert_keyboard_interrupt_to_task_cancel(one_shot_request) is False
 
 
+def test_run_request_closes_loop_when_cleanup_is_interrupted(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    class _FakeTask:
+        def __init__(self) -> None:
+            self.cancelled = False
+
+        def done(self) -> bool:
+            return False
+
+        def cancel(self) -> None:
+            self.cancelled = True
+
+    class _FakeLoop:
+        def __init__(self) -> None:
+            self.task = _FakeTask()
+            self.main_gather = object()
+            self.shutdown_marker = object()
+            self.closed = False
+
+        def is_running(self) -> bool:
+            return False
+
+        def create_task(self, coro: Any) -> _FakeTask:
+            coro.close()
+            return self.task
+
+        def run_until_complete(self, awaitable: object) -> None:
+            if awaitable is self.task:
+                raise KeyboardInterrupt()
+            if awaitable is self.main_gather:
+                raise KeyboardInterrupt()
+
+        def shutdown_asyncgens(self) -> object:
+            return self.shutdown_marker
+
+        def close(self) -> None:
+            self.closed = True
+
+    fake_loop = _FakeLoop()
+
+    async def _fake_run_agent_request(_request: AgentRunRequest) -> None:
+        return None
+
+    def _fake_gather(*aws: object, return_exceptions: bool = False) -> object:
+        del return_exceptions
+        if aws == (fake_loop.task,):
+            return fake_loop.main_gather
+        return object()
+
+    monkeypatch.setattr(runner, "configure_uvloop", lambda: (False, False))
+    monkeypatch.setattr(runner, "ensure_event_loop", lambda: fake_loop)
+    monkeypatch.setattr(runner, "set_asyncio_exception_handler", lambda _loop: None)
+    monkeypatch.setattr(runner, "run_agent_request", _fake_run_agent_request)
+    monkeypatch.setattr(asyncio, "all_tasks", lambda _loop: set())
+    monkeypatch.setattr(asyncio, "gather", _fake_gather)
+
+    with pytest.raises(KeyboardInterrupt):
+        runner.run_request(_make_request(result_file=None, message="hello"))
+
+    assert fake_loop.closed is True
+
+
 def test_sanitize_result_suffix() -> None:
     assert _sanitize_result_suffix("openai/gpt-4o") == "openai_gpt-4o"
     assert _sanitize_result_suffix("  model name  ") == "model_name"

tests/unit/fast_agent/ui/test_prompt_input_runtime.py

@@ -1,8 +1,14 @@
 from __future__ import annotations
 
-from typing import TextIO, cast
+from typing import TYPE_CHECKING, TextIO, cast
+
+import pytest
 
 from fast_agent.ui.prompt import input_runtime
+from fast_agent.ui.prompt.keybindings import PromptInputInterrupt
+
+if TYPE_CHECKING:
+    from prompt_toolkit import PromptSession
 
 
 class _FakeStream:
@@ -57,3 +63,28 @@ def test_format_prompt_prefix_omits_default_agent_name() -> None:
     assert input_runtime._format_prompt_prefix("dev", default_agent_name="dev") == "❯"
     assert input_runtime._format_prompt_prefix("default") == "default ❯"
     assert input_runtime._format_prompt_prefix("dev") == "dev ❯"
+
+
+@pytest.mark.asyncio
+async def test_run_prompt_once_converts_prompt_input_interrupt_to_interrupt_command() -> None:
+    class _Buffer:
+        def __init__(self) -> None:
+            self.accept_handler = None
+
+    class _Session:
+        def __init__(self) -> None:
+            self.default_buffer = _Buffer()
+
+        async def prompt_async(self, *_args, **_kwargs):
+            raise PromptInputInterrupt()
+
+    result = await input_runtime.run_prompt_once(
+        session=cast("PromptSession", _Session()),
+        agent_name="agent",
+        default_agent_name="agent",
+        default_buffer="",
+        resolve_prompt_text=lambda: "❯ ",
+        parse_special_input=lambda value: value,
+    )
+
+    assert type(result).__name__ == "InterruptCommand"

tests/unit/fast_agent/ui/test_prompt_keybindings.py

@@ -8,6 +8,7 @@
 from prompt_toolkit.keys import Keys
 
 from fast_agent.ui.prompt.keybindings import (
+    PromptInputInterrupt,
     _accept_completion,
     _cycle_completion,
     create_keybindings,
@@ -137,3 +138,20 @@ def invalidate(self) -> None:
         binding = _binding_for(kb, key)
         binding.handler(SimpleNamespace(current_buffer=Buffer(), app=_App()))
         assert label in events
+
+
+def test_ctrl_c_binding_exits_with_prompt_input_interrupt() -> None:
+    class _App:
+        def __init__(self) -> None:
+            self.exception: BaseException | None = None
+
+        def exit(self, *, exception: BaseException | None = None) -> None:
+            self.exception = exception
+
+    app = _App()
+    kb = create_keybindings()
+    binding = _binding_for(kb, Keys.ControlC)
+
+    binding.handler(SimpleNamespace(current_buffer=Buffer(), app=app))
+
+    assert isinstance(app.exception, PromptInputInterrupt)