ai: apply changes for #873 (2 review threads)

Addresses:
  - #3635836346 at src/databricks/sql/client.py:1764
  - #3635836348 at src/databricks/sql/client.py:1770

Signed-off-by: peco-engineer-bot[bot] <peco-engineer-bot[bot]@users.noreply.github.com>

Author: peco-engineer-bot[bot]

src/databricks/sql/client.py

@@ -1761,11 +1761,13 @@ def close(self) -> None:
         self.open = False
         if self.active_result_set:
             self._close_and_clear_active_result_set()
-        elif self.active_command_id is not None:
+        elif self.active_command_id is not None and self.connection.open:
             # Async submission whose result was never fetched (no
             # get_execution_result call), so the result-set close path never
             # fired. Issue an explicit close_command to free the server-side
             # statement handle instead of leaking it until session close.
+            # Gate on connection.open (mirroring ResultSet.close) so we don't
+            # attempt a network call on an already-torn-down session.
             try:
                 self.backend.close_command(self.active_command_id)
             except Exception as exc:

tests/unit/test_client.py

@@ -161,6 +161,67 @@ def test_closing_connection_closes_commands(self, mock_thrift_client_class):
                     # Should NOT have called backend.close_command (already closed)
                     mock_backend.close_command.assert_not_called()
 
+    def test_cursor_close_frees_command_when_no_result_set(self):
+        """Closing a cursor with an unfetched async command frees the server handle.
+
+        When active_result_set is None but active_command_id is set (an async
+        submission whose result was never fetched), close() must issue an
+        explicit backend.close_command so the server-side statement handle is
+        not leaked. This branch is backend-agnostic (Thrift/SEA/kernel).
+        """
+        mock_backend = Mock(spec=ThriftDatabricksClient)
+        mock_connection = Mock()
+        cursor = client.Cursor(connection=mock_connection, backend=mock_backend)
+
+        command_id = Mock(spec=CommandId)
+        cursor.active_command_id = command_id
+        cursor.active_result_set = None
+
+        cursor.close()
+
+        mock_backend.close_command.assert_called_once_with(command_id)
+        self.assertIsNone(cursor.active_command_id)
+
+    def test_cursor_close_skips_command_when_connection_closed(self):
+        """Closing a cursor after its session is gone must not call close_command.
+
+        Mirrors ResultSet.close, which gates backend.close_command on
+        connection.open, to avoid a network call on a dead session (and a
+        spurious warning) during shutdown ordering.
+        """
+        mock_backend = Mock(spec=ThriftDatabricksClient)
+        mock_connection = Mock()
+        mock_connection.open = False
+        cursor = client.Cursor(connection=mock_connection, backend=mock_backend)
+
+        cursor.active_command_id = Mock(spec=CommandId)
+        cursor.active_result_set = None
+
+        cursor.close()
+
+        mock_backend.close_command.assert_not_called()
+        self.assertIsNone(cursor.active_command_id)
+
+    def test_cursor_close_does_not_double_close_when_result_set_present(self):
+        """Closing a cursor with an active result set must NOT call close_command.
+
+        The result-set close path is responsible for freeing the handle in that
+        case, so the elif branch must not fire (no double-close).
+        """
+        mock_backend = Mock(spec=ThriftDatabricksClient)
+        mock_connection = Mock()
+        cursor = client.Cursor(connection=mock_connection, backend=mock_backend)
+
+        cursor.active_command_id = Mock(spec=CommandId)
+        result_set = Mock(spec=ResultSet)
+        cursor.active_result_set = result_set
+
+        cursor.close()
+
+        result_set.close.assert_called_once()
+        mock_backend.close_command.assert_not_called()
+        self.assertIsNone(cursor.active_command_id)
+
     @patch("%s.session.ThriftDatabricksClient" % PACKAGE_NAME)
     def test_cant_open_cursor_on_closed_connection(self, mock_client_class):
         connection = databricks.sql.connect(**self.DUMMY_CONNECTION_ARGS)