asyncio.gather behaves inconsistently when handling KeyboardInterruption/SystemExit

Bug report

Here is a minimal example:

import asyncio

e = KeyboardInterrupt  # or SystemExit


async def main_task():
    await asyncio.gather(
        sub_task(),
    )


async def sub_task():
    raise e


if __name__ == '__main__':
    try:
        asyncio.run(main_task())
    except e:
        print(f'Handle {e}')

This code handles the Interrupt normally as I expected.

Handle <class 'KeyboardInterrupt'>

Process finished with exit code 0

But when I add the asyncio.sleep(0) (can be replaced by other task, not important) into main_task's asyncio.gather

import asyncio

e = KeyboardInterrupt  # or SystemExit


async def main_task():
    await asyncio.gather(
        sub_task(),
        asyncio.sleep(0)
    )


async def sub_task():
    raise e


if __name__ == '__main__':
    try:
        asyncio.run(main_task())
    except e:
        print(f'Handle {e}')

There is an unexpected traceback print out which is really confusing 💦, this traceback indicates that there is
another KeyboardInterrupt raised.

Traceback (most recent call last):
  File "/Users/huanghuiling/PycharmProjects/Lighting-bilibili-download/tests/log_test.py", line 45, in <module>
    asyncio.run(main_task())
  File "/opt/homebrew/Caskroom/miniforge/base/envs/test9/lib/python3.9/asyncio/runners.py", line 47, in run
    _cancel_all_tasks(loop)
  File "/opt/homebrew/Caskroom/miniforge/base/envs/test9/lib/python3.9/asyncio/runners.py", line 63, in _cancel_all_tasks
    loop.run_until_complete(
  File "/opt/homebrew/Caskroom/miniforge/base/envs/test9/lib/python3.9/asyncio/base_events.py", line 629, in run_until_complete
    self.run_forever()
  File "/opt/homebrew/Caskroom/miniforge/base/envs/test9/lib/python3.9/asyncio/base_events.py", line 596, in run_forever
    self._run_once()
  File "/opt/homebrew/Caskroom/miniforge/base/envs/test9/lib/python3.9/asyncio/base_events.py", line 1890, in _run_once
    handle._run()
  File "/opt/homebrew/Caskroom/miniforge/base/envs/test9/lib/python3.9/asyncio/events.py", line 80, in _run
    self._context.run(self._callback, *self._args)
  File "/Users/huanghuiling/PycharmProjects/Lighting-bilibili-download/tests/log_test.py", line 7, in main_task
    await asyncio.gather(
  File "/opt/homebrew/Caskroom/miniforge/base/envs/test9/lib/python3.9/asyncio/runners.py", line 44, in run
    return loop.run_until_complete(main)
  File "/opt/homebrew/Caskroom/miniforge/base/envs/test9/lib/python3.9/asyncio/base_events.py", line 629, in run_until_complete
    self.run_forever()
  File "/opt/homebrew/Caskroom/miniforge/base/envs/test9/lib/python3.9/asyncio/base_events.py", line 596, in run_forever
    self._run_once()
  File "/opt/homebrew/Caskroom/miniforge/base/envs/test9/lib/python3.9/asyncio/base_events.py", line 1890, in _run_once
    handle._run()
  File "/opt/homebrew/Caskroom/miniforge/base/envs/test9/lib/python3.9/asyncio/events.py", line 80, in _run
    self._context.run(self._callback, *self._args)
  File "/Users/huanghuiling/PycharmProjects/Lighting-bilibili-download/tests/log_test.py", line 14, in sub_task
    raise e
KeyboardInterrupt

Process finished with exit code 0

So I go deeply into the asyncio.run, and write code (a simplified asyncio.run) below to figure out what happen.

import asyncio

e = KeyboardInterrupt  # or SystemExit


async def main_task():
    await asyncio.gather(
        sub_task(),
        asyncio.sleep(0)
    )


async def sub_task():
    raise e


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    try:
        loop.run_until_complete(main_task())
    except e:
        print(f'Expected {e}')
    finally:
        try:
            tasks = asyncio.all_tasks(loop)
            for t in tasks:
                t.cancel()
            # ⬇️ this line will raise another KeyboardInterrupt which is unexpected ⬇️ 
            loop.run_until_complete(asyncio.gather(*tasks, return_exceptions=True))
        except e:
            print(f'Unexpected {e} !!!!')

This line will raise another KeyboardInterrupt which is unexpected.

loop.run_until_complete(asyncio.gather(*tasks, return_exceptions=True))

Expected <class 'KeyboardInterrupt'>
Unexpected <class 'KeyboardInterrupt'> !!!!

Process finished with exit code 0

Note that this line is used to cancel all tasks during gracefully shutdown (also in asyncio.run), and when I change
e to other error like IndexError(any BaseException), this code works fine without unexpected another exception.
I believe this is related to the asyncio treats SystemExit and KeyboardInterrupt in different way. For example
in events.py

def _run(self):
    try:
        self._context.run(self._callback, *self._args)
    except (SystemExit, KeyboardInterrupt):
        raise
    except BaseException as exc:
        cb = format_helpers._format_callback_source(
            self._callback, self._args)
        msg = f'Exception in callback {cb}'
        context = {
            'message': msg,
            'exception': exc,
            'handle': self,
        }
        if self._source_traceback:
            context['source_traceback'] = self._source_traceback
        self._loop.call_exception_handler(context)
    self = None  # Needed to break cycles when an exception occurs.

My question is:

1. Why asyncio.gather behaves inconsistently.
2. Is there any reason to treat KeyboardInterrupt differently, since the simplest way
   to solve this bug is to handle it same as BaseException.

I think user would like to handle all error consistently during the running of a task whether
it's KeyboardInterrupt or BaseException.

Even asyncio treats them in different way (incase really necessary)

asyncio.gather(sub_task())

and

asyncio.gather(sub_task(), asyncio.sleep(0))

should behave consistently, so I think this is a bug in asyncio.

Your environment

• CPython versions tested on: 3.8, 3.9, 3.10
• Operating system and architecture: both macOS and windows

Riproduci innanzitutto entrambi gli esempi di asyncio.gather e la sequenza di shutdown semplificata. Leggi asyncio.gather, asyncio/events.py e asyncio/runners.py per tracciare come KeyboardInterrupt e SystemExit si propagano durante la cancellazione dei task; il lavoro è completato quando il comportamento previsto è definito, implementato in modo coerente e coperto da test di regressione.