`create_pipe_input` breaking change in versions `>=3.0.29` results in `OSError: Bad file descriptor`

Problem Description:

I attempted to upgrade prompt-toolkit dependency in rasa to 3.0.30 and that resulted in this unit test test_record_messages() to fail with this stacktrace:

self = <_UnixSelectorEventLoop running=False closed=False debug=True>, fd = 14, callback = <function _attached_input.<locals>.callback_wrapper at 0x14a105670>, args = ()
handle = <Handle _attached_input.<locals>.callback_wrapper() at /Users/ancalita/rasa-projects/oss-0908-3.2.x/lib/python3.9/site...olkit/input/vt100.py:165 created at /Users/ancalita/.pyenv/versions/3.9.9/lib/python3.9/asyncio/selector_events.py:259>

    def _add_reader(self, fd, callback, *args):
        self._check_closed()
        handle = events.Handle(callback, args, self, None)
        try:
>           key = self._selector.get_key(fd)

../../.pyenv/versions/3.9.9/lib/python3.9/asyncio/selector_events.py:261: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <selectors.KqueueSelector object at 0x149eedd60>, fileobj = 14

    def get_key(self, fileobj):
        """Return the key associated to a registered file object.
    
        Returns:
        SelectorKey for this file object
        """
        mapping = self.get_map()
        if mapping is None:
            raise RuntimeError('Selector is closed')
        try:
            return mapping[fileobj]
        except KeyError:
>           raise KeyError("{!r} is not registered".format(fileobj)) from None
E           KeyError: '14 is not registered'

../../.pyenv/versions/3.9.9/lib/python3.9/selectors.py:193: KeyError

During handling of the above exception, another exception occurred:

monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x149eed3d0>, capsys = <_pytest.capture.CaptureFixture object at 0x149eed940>

    async def test_record_messages(monkeypatch: MonkeyPatch, capsys: CaptureFixture):
        input_output = [
            {
                "in": "Give me a question!",
                "out": [
                    {
                        "buttons": [
                            {
                                "title": "button 1 title",
                                "payload": "button 1 payload",
                                "details": "button 1 details",
                            }
                        ],
                        "text": "This is a button 1",
                    },
                    {
                        "buttons": [
                            {
                                "title": "button 2 title",
                                "payload": "button 2 payload",
                                "details": "button 2 details",
                            }
                        ],
                        "text": "This is a button 2",
                    },
                    {
                        "buttons": [
                            {
                                "title": "button 3 title",
                                "payload": "button 3 payload",
                                "details": "button 3 details",
                            }
                        ],
                        "text": "This is a button 3",
                    },
                ],
            },
            {"in": ENTER, "out": [{"text": "You've pressed the button"}]},
            {"in": "Dummy message", "out": [{"text": "Dummy response"}]},
        ]
    
        inp = mock_stdin([m["in"] for m in input_output])
    
        server_url = "http://example.com"
        endpoint = f"{server_url}/webhooks/rest/webhook"
    
        with aioresponses() as mocked:
    
            for output in [m["out"] for m in input_output]:
                if output:
                    mocked.post(url=endpoint, payload=output)
    
>           num_of_messages = await record_messages(
                "123",
                server_url=server_url,
                max_message_limit=len(input_output),
                use_response_stream=False,
            )

tests/core/channels/test_cmdline.py:94: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
rasa/core/channels/console.py:208: in record_messages
    text = await _get_user_input(previous_response)
rasa/core/channels/console.py:134: in _get_user_input
    response = await question.ask_async()
../oss-0908-3.2.x/lib/python3.9/site-packages/questionary/question.py:46: in ask_async
    return await self.unsafe_ask_async(patch_stdout)
../oss-0908-3.2.x/lib/python3.9/site-packages/questionary/question.py:132: in unsafe_ask_async
    return await r
../oss-0908-3.2.x/lib/python3.9/site-packages/prompt_toolkit/application/application.py:856: in run_async
    return await _run_async2()
../oss-0908-3.2.x/lib/python3.9/site-packages/prompt_toolkit/application/application.py:826: in _run_async2
    result = await _run_async()
../oss-0908-3.2.x/lib/python3.9/site-packages/prompt_toolkit/application/application.py:741: in _run_async
    with self.input.raw_mode(), self.input.attach(
../../.pyenv/versions/3.9.9/lib/python3.9/contextlib.py:119: in __enter__
    return next(self.gen)
../oss-0908-3.2.x/lib/python3.9/site-packages/prompt_toolkit/input/vt100.py:176: in _attached_input
    loop.add_reader(fd, callback_wrapper)
../../.pyenv/versions/3.9.9/lib/python3.9/asyncio/selector_events.py:336: in add_reader
    self._add_reader(fd, callback, *args)
../../.pyenv/versions/3.9.9/lib/python3.9/asyncio/selector_events.py:263: in _add_reader
    self._selector.register(fd, selectors.EVENT_READ,
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <selectors.KqueueSelector object at 0x149eedd60>, fileobj = 14, events = 1
data = (<Handle _attached_input.<locals>.callback_wrapper() at /Users/ancalita/rasa-projects/oss-0908-3.2.x/lib/python3.9/sit...nput/vt100.py:165 created at /Users/ancalita/.pyenv/versions/3.9.9/lib/python3.9/asyncio/selector_events.py:259>, None)

    def register(self, fileobj, events, data=None):
        key = super().register(fileobj, events, data)
        try:
            if events & EVENT_READ:
                kev = select.kevent(key.fd, select.KQ_FILTER_READ,
                                    select.KQ_EV_ADD)
>               self._selector.control([kev], 0, 0)
E               OSError: [Errno 9] Bad file descriptor

This error might have to do with changes introduced in this PR by @jonathanslenders
In the end I had to pin prompt-toolkit to version 3.0.28 to make sure CI pipeline checks pass.

Start with the reported failure in tests/core/channels/test_cmdline.py and the input path through rasa/core/channels/console.py. Compare prompt_toolkit/input/vt100.py before and after PR 1586, focusing on create_pipe_input and file-descriptor attachment. Reproduce the failure with prompt-toolkit versions 3.0.29 and later; done means the test no longer raises OSError: Bad file descriptor without pinning to 3.0.28.