fix(cli): persist interrupted session transcript on SIGHUP/SIGTERM/window-close

cli.py

@@ -11524,6 +11524,36 @@ def _clear_terminal_on_exit(self):
         except Exception:
             pass
 
+    def _persist_active_session_before_close(self):
+        """Best-effort SQLite/JSON flush before the CLI marks a session closed.
+
+        ``run_conversation()`` normally persists at turn boundaries, but a
+        terminal close/SIGHUP/SIGTERM can unwind the prompt_toolkit app while
+        the agent thread still holds the current turn only in memory.  Flush the
+        agent's live ``_session_messages`` before ``end_session()`` so resume,
+        session_search, and state.db do not lose the interrupted turn.
+        """
+        agent = getattr(self, "agent", None)
+        if not agent or not hasattr(agent, "_persist_session"):
+            return
+
+        messages = getattr(agent, "_session_messages", None)
+        if not isinstance(messages, list):
+            messages = getattr(self, "conversation_history", None)
+        if not isinstance(messages, list) or not messages:
+            return
+
+        conversation_history = getattr(self, "conversation_history", None)
+        if not isinstance(conversation_history, list):
+            conversation_history = messages
+
+        try:
+            agent._persist_session(messages, conversation_history)
+            if getattr(agent, "session_id", None):
+                self.session_id = agent.session_id
+        except (Exception, KeyboardInterrupt) as e:
+            logger.debug("Could not persist active CLI session before close: %s", e)
+
     def _print_exit_summary(self):
         """Print session resume info on exit, similar to Claude Code."""
         # Clear the screen + scrollback before printing the summary so the
@@ -14220,6 +14250,12 @@ def new_event_loop(self):
             set_sudo_password_callback(None)
             set_approval_callback(None)
             set_secret_capture_callback(None)
+            # Flush any in-memory turn transcript before marking the session
+            # closed.  On SIGHUP/SIGTERM/window close the agent thread may not
+            # reach its normal run_conversation() persistence path before the
+            # daemon thread is reaped.
+            self._persist_active_session_before_close()
+
             # Close session in SQLite
             if hasattr(self, '_session_db') and self._session_db and self.agent:
                 try:

tests/cli/test_cli_shutdown_memory_messages.py

@@ -109,3 +109,61 @@ def test_cleanup_provider_exception_is_swallowed(mock_invoke_hook):
         cli_mod._cleanup_done = False
 
     agent.shutdown_memory_provider.assert_called_once()
+
+
+def test_cli_close_persists_agent_session_messages_before_end_session():
+    """CLI shutdown flushes live agent messages before closing the session."""
+    import cli as cli_mod
+
+    transcript = [
+        {"role": "user", "content": "long task"},
+        {"role": "assistant", "content": "partial answer"},
+    ]
+    conversation_history = [{"role": "user", "content": "long task"}]
+
+    cli = object.__new__(cli_mod.HermesCLI)
+    cli.conversation_history = conversation_history
+    cli.session_id = "old-session"
+    agent = MagicMock()
+    agent.session_id = "live-session"
+    agent._session_messages = transcript
+    cli.agent = agent
+
+    cli._persist_active_session_before_close()
+
+    agent._persist_session.assert_called_once_with(transcript, conversation_history)
+    assert cli.session_id == "live-session"
+
+
+def test_cli_close_persist_falls_back_to_conversation_history():
+    """Bare MagicMock agents do not provide a real _session_messages list."""
+    import cli as cli_mod
+
+    conversation_history = [{"role": "user", "content": "saved from cli"}]
+    cli = object.__new__(cli_mod.HermesCLI)
+    cli.conversation_history = conversation_history
+    cli.session_id = "session-id"
+    agent = MagicMock()
+    agent.session_id = "session-id"
+    cli.agent = agent
+
+    cli._persist_active_session_before_close()
+
+    agent._persist_session.assert_called_once_with(conversation_history, conversation_history)
+
+
+def test_cli_close_persist_skips_empty_transcripts():
+    """Do not create empty session writes for idle CLI startup/shutdown."""
+    import cli as cli_mod
+
+    cli = object.__new__(cli_mod.HermesCLI)
+    cli.conversation_history = []
+    cli.session_id = "session-id"
+    agent = MagicMock()
+    agent.session_id = "session-id"
+    agent._session_messages = []
+    cli.agent = agent
+
+    cli._persist_active_session_before_close()
+
+    agent._persist_session.assert_not_called()