Handle KeyboardInterrupt gracefully in ConcurrencyGroup

libs/concurrency_group/imbue/concurrency_group/concurrency_group.py

@@ -135,8 +135,14 @@ def __exit__(self, exc_type: type | None, exc_value: BaseException | None, trace
         try:
             with self._lock:
                 self._state = ConcurrencyGroupState.EXITING
+            if isinstance(exc_value, KeyboardInterrupt):
+                self.shutdown()
             self._exit(exc_value)
         except BaseException as exit_exception:
+            if isinstance(exc_value, KeyboardInterrupt):
+                # When the user presses Ctrl+C, suppress cleanup errors (strand timeouts, process
+                # failures, etc.) and let the original KeyboardInterrupt propagate cleanly.
+                return
             self._exit_exception = exit_exception
             raise
         finally:

libs/concurrency_group/imbue/concurrency_group/concurrency_group_test.py

@@ -484,3 +484,48 @@ def test_new_resources_cannot_be_created_when_ancestor_has_failed_strands() -> N
                     with pytest.raises(ConcurrencyExceptionGroup) as exception_info_thread:
                         cg_inner.start_new_thread(target=lambda: 1)
     assert exception_info_thread.value.only_exception_is_instance_of(AncestorConcurrentFailure)
+
+
+# KeyboardInterrupt handling tests
+
+
+def test_keyboard_interrupt_sets_shutdown_event_and_propagates_cleanly() -> None:
+    """When KeyboardInterrupt is the exit exception, the shutdown event is set and
+    the interrupt propagates without being wrapped in ConcurrencyExceptionGroup."""
+    with pytest.raises(KeyboardInterrupt):
+        with ConcurrencyGroup(name="outer") as cg:
+            assert not cg.is_shutting_down()
+            raise KeyboardInterrupt()
+    assert cg.is_shutting_down()
+
+
+def test_keyboard_interrupt_suppresses_strand_timeout_errors() -> None:
+    """Strand timeout errors during cleanup are suppressed when KeyboardInterrupt is active,
+    so users don't see scary ConcurrencyExceptionGroup tracebacks on Ctrl+C."""
+    blocked = Event()
+    with pytest.raises(KeyboardInterrupt):
+        with ConcurrencyGroup(name="outer", shutdown_timeout_seconds=TINY_SLEEP) as cg:
+            cg.start_new_thread(target=blocked.wait)
+            raise KeyboardInterrupt()
+    # The KeyboardInterrupt propagated cleanly -- no ConcurrencyExceptionGroup.
+    assert cg.is_shutting_down()
+
+
+@pytest.mark.filterwarnings("ignore::pytest.PytestUnhandledThreadExceptionWarning")
+def test_keyboard_interrupt_suppresses_strand_failure_errors() -> None:
+    """Strand failure errors during cleanup are suppressed when KeyboardInterrupt is active."""
+    with pytest.raises(KeyboardInterrupt):
+        with ConcurrencyGroup(name="outer") as cg:
+            cg.start_new_thread(target=_raise_intentional_error)
+            raise KeyboardInterrupt()
+    assert cg.is_shutting_down()
+
+
+def test_keyboard_interrupt_propagates_to_nested_groups() -> None:
+    """Shutdown event propagates to child concurrency groups on KeyboardInterrupt."""
+    with pytest.raises(KeyboardInterrupt):
+        with ConcurrencyGroup(name="outer") as cg_outer:
+            with cg_outer.make_concurrency_group(name="inner") as cg_inner:
+                raise KeyboardInterrupt()
+    assert cg_outer.is_shutting_down()
+    assert cg_inner.is_shutting_down()

libs/mng/imbue/mng/cli/common_opts.py

@@ -143,9 +143,7 @@ def setup_command_context(
     # Create a top-level ConcurrencyGroup for process management
     cg = ConcurrencyGroup(name=f"mng-{command_name}")
     cg.__enter__()
-    # We explicitly pass None to __exit__ so that Click exceptions (e.g. UsageError) don't get
-    # wrapped in ConcurrencyExceptionGroup, which would break Click's error handling.
-    ctx.call_on_close(lambda: cg.__exit__(None, None, None))
+    ctx.call_on_close(lambda: _close_concurrency_group(cg))
 
     # Load config (is_interactive will be resolved below)
     context_dir = Path(initial_opts.project_context_path) if initial_opts.project_context_path else None
@@ -496,6 +494,20 @@ def _apply_plugin_option_overrides(
     )
 
 
+def _close_concurrency_group(cg: ConcurrencyGroup) -> None:
+    """Close the top-level ConcurrencyGroup during Click context teardown.
+
+    When a KeyboardInterrupt is active (user pressed Ctrl+C), pass it to __exit__ so
+    it can set the shutdown event and suppress cleanup errors. Otherwise pass None so
+    Click exceptions (e.g. UsageError) don't get wrapped in ConcurrencyExceptionGroup.
+    """
+    exc_info = sys.exc_info()
+    if isinstance(exc_info[1], KeyboardInterrupt):
+        cg.__exit__(*exc_info)
+    else:
+        cg.__exit__(None, None, None)
+
+
 def _run_single_script(script: str, cg: ConcurrencyGroup) -> tuple[str, int, str, str]:
     """Run a single script and return (script, exit_code, stdout, stderr)."""
     try: