diff --git a/cli.py b/cli.py
index 6475db00e..bb4286b34 100644
--- a/cli.py
+++ b/cli.py
@@ -7163,13 +7163,13 @@ class HermesCLI:
             if self.agent and getattr(self.agent, '_honcho', None):
                 try:
                     self.agent._honcho.shutdown()
-                except Exception:
+                except (Exception, KeyboardInterrupt):
                     pass
             # Close session in SQLite
             if hasattr(self, '_session_db') and self._session_db and self.agent:
                 try:
                     self._session_db.end_session(self.agent.session_id, "cli_close")
-                except Exception as e:
+                except (Exception, KeyboardInterrupt) as e:
                     logger.debug("Could not close session in DB: %s", e)
             _run_cleanup()
             self._print_exit_summary()
diff --git a/cron/scheduler.py b/cron/scheduler.py
index 133b4c1b8..e6313cd7b 100644
--- a/cron/scheduler.py
+++ b/cron/scheduler.py
@@ -474,11 +474,11 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]:
         if _session_db:
             try:
                 _session_db.end_session(_cron_session_id, "cron_complete")
-            except Exception as e:
+            except (Exception, KeyboardInterrupt) as e:
                 logger.debug("Job '%s': failed to end session: %s", job_id, e)
             try:
                 _session_db.close()
-            except Exception as e:
+            except (Exception, KeyboardInterrupt) as e:
                 logger.debug("Job '%s': failed to close SQLite session store: %s", job_id, e)
 
 
diff --git a/run_agent.py b/run_agent.py
index f2f023959..49fe64adf 100644
--- a/run_agent.py
+++ b/run_agent.py
@@ -2274,7 +2274,7 @@ class AIAgent:
                 return
             try:
                 manager.flush_all()
-            except Exception as exc:
+            except (Exception, KeyboardInterrupt) as exc:
                 logger.debug("Honcho flush on exit failed (non-fatal): %s", exc)
 
         atexit.register(_flush_honcho_on_exit)
diff --git a/tests/test_exit_cleanup_interrupt.py b/tests/test_exit_cleanup_interrupt.py
new file mode 100644
index 000000000..e20ce5c7b
--- /dev/null
+++ b/tests/test_exit_cleanup_interrupt.py
@@ -0,0 +1,105 @@
+"""Tests for KeyboardInterrupt handling in exit cleanup paths.
+
+``except Exception`` does not catch ``KeyboardInterrupt`` (which inherits
+from ``BaseException``).  A second Ctrl+C during exit cleanup must not
+abort remaining cleanup steps.  These tests exercise the actual production
+code paths — not a copy of the try/except pattern.
+"""
+
+import atexit
+import weakref
+from unittest.mock import MagicMock, patch, call
+
+import pytest
+
+
+class TestHonchoAtexitFlush:
+    """run_agent.py — _register_honcho_exit_hook atexit handler."""
+
+    def test_keyboard_interrupt_during_flush_does_not_propagate(self):
+        """The atexit handler must swallow KeyboardInterrupt from flush_all()."""
+        mock_manager = MagicMock()
+        mock_manager.flush_all.side_effect = KeyboardInterrupt
+
+        # Capture functions passed to atexit.register
+        registered_fns = []
+        original_register = atexit.register
+
+        def capturing_register(fn, *args, **kwargs):
+            registered_fns.append(fn)
+            # Don't actually register — we don't want side effects
+
+        with patch("atexit.register", side_effect=capturing_register):
+            from run_agent import AIAgent
+            agent = object.__new__(AIAgent)
+            agent._honcho = mock_manager
+            agent._honcho_exit_hook_registered = False
+            agent._register_honcho_exit_hook()
+
+        # Our handler is the last one registered
+        assert len(registered_fns) >= 1, "atexit handler was not registered"
+        flush_handler = registered_fns[-1]
+
+        # Invoke the registered handler — must not raise
+        flush_handler()
+        mock_manager.flush_all.assert_called_once()
+
+
+class TestCronJobCleanup:
+    """cron/scheduler.py — end_session + close in the finally block."""
+
+    def test_keyboard_interrupt_in_end_session_does_not_skip_close(self):
+        """If end_session raises KeyboardInterrupt, close() must still run."""
+        mock_db = MagicMock()
+        mock_db.end_session.side_effect = KeyboardInterrupt
+
+        from cron import scheduler
+
+        job = {
+            "id": "test-job-1",
+            "name": "test cleanup",
+            "prompt": "hello",
+            "schedule": "0 9 * * *",
+            "model": "test/model",
+        }
+
+        with patch("hermes_state.SessionDB", return_value=mock_db), \
+             patch.object(scheduler, "_build_job_prompt", return_value="hello"), \
+             patch.object(scheduler, "_resolve_origin", return_value=None), \
+             patch.object(scheduler, "_resolve_delivery_target", return_value=None), \
+             patch("dotenv.load_dotenv", return_value=None), \
+             patch("run_agent.AIAgent") as MockAgent:
+            # Make the agent raise immediately so we hit the finally block
+            MockAgent.return_value.run_conversation.side_effect = RuntimeError("boom")
+            scheduler.run_job(job)
+
+        mock_db.end_session.assert_called_once()
+        mock_db.close.assert_called_once()
+
+    def test_keyboard_interrupt_in_close_does_not_propagate(self):
+        """If close() raises KeyboardInterrupt, it must not escape run_job."""
+        mock_db = MagicMock()
+        mock_db.close.side_effect = KeyboardInterrupt
+
+        from cron import scheduler
+
+        job = {
+            "id": "test-job-2",
+            "name": "test close interrupt",
+            "prompt": "hello",
+            "schedule": "0 9 * * *",
+            "model": "test/model",
+        }
+
+        with patch("hermes_state.SessionDB", return_value=mock_db), \
+             patch.object(scheduler, "_build_job_prompt", return_value="hello"), \
+             patch.object(scheduler, "_resolve_origin", return_value=None), \
+             patch.object(scheduler, "_resolve_delivery_target", return_value=None), \
+             patch("dotenv.load_dotenv", return_value=None), \
+             patch("run_agent.AIAgent") as MockAgent:
+            MockAgent.return_value.run_conversation.side_effect = RuntimeError("boom")
+            # Must not raise
+            scheduler.run_job(job)
+
+        mock_db.end_session.assert_called_once()
+        mock_db.close.assert_called_once()