Repository: aio-libs/aiomonitor (language: python)

Issue:
Announce: The new maintainer
As I have presented [aiomonitor-ng (next generation)](https://github.com/achimnol/aiomonitor-ng), which is a fork of this project, as [a session talk](https://us.pycon.org/2023/schedule/presentation/13/) and [a poster](https://us.pycon.org/2023/events/sprints/#sprint-45) in PyCon US 2023, @webknjaz has promoted me as the maintainer of this project.

aiomonitor-ng adds:
- A completely rewritten terminal UI using [prompt_toolkit](https://python-prompt-toolkit.readthedocs.io/en/master/) and [Click](https://click.palletsprojects.com/en/8.1.x/) including auto-completion of commands and task IDs
- Support for concurrent client connections (#140 is resolved!)
- Task creation chain tracker
- Task termination history
- Task cancellation chain tracker
- Filtering of long task lists using the task name and coro name

I'm going to do the following things during/after the PyCon US 2023 sprints:

- [x] Backport the changes of aiomonitor-ng
- [x] Rename the `master` branch to `main`
- Revamp the CI/CD and fix the broken test suite (let it use `pytest-asyncio`)
- Add a minimal web UI support in addition to the terminal UI (#84)
- Diagnose an issue which causes negative side-effects to the application event loop when the telnet server is broken (more details will follow as a separate issue)
- Pagination support for long task lists (#326)
- Integrate [pystack](https://github.com/bloomberg/pystack) as an additional option to inspect the deeper side of task stacks

Sprinters and contributors are also welcome!
Update pytest-sugar to 0.8.0

There's a new version of [pytest-sugar](https://pypi.python.org/pypi/pytest-sugar) available.
You are currently using **0.7.1**. I have updated it to **0.8.0**



These links might come in handy:  <a href="http://pypi.python.org/pypi/pytest-sugar">PyPI</a> | <a href="https://pyup.io/changelogs/pytest-sugar/">Changelog</a> | <a href="http://pivotfinland.com/pytest-sugar/">Homepage</a> 



### Changelog
> 
>### 0.8.0

>^^^^^^^^^^^^^^^^^^^

>- Release as an universal wheel
>- Pytest3 compatibility
>- Treat setup/teardown failures as errors
>- Fix path issue in --new-summary
>- Disable sugar output when not in terminal, should help with testing other pytest plugins
>- Add double colons when in verbose mode
>- Make --new-summary default, replaced flag with --old-summary







*Got merge conflicts? Close this PR and delete the branch. I'll create a new PR for you.*

Happy merging! 🤖



Discussion / hints:

## [Current coverage](https://codecov.io/gh/aio-libs/aiomonitor/pull/10?src=pr) is 93.33% (diff: 100%)
> Merging [#10](https://codecov.io/gh/aio-libs/aiomonitor/pull/10?src=pr) into [master](https://codecov.io/gh/aio-libs/aiomonitor/branch/master?src=pr) will not change coverage

```diff
@@             master        #10   diff @@
==========================================
  Files             3          3          
  Lines           225        225          
  Methods           0          0          
  Messages          0          0          
  Branches         31         31          
==========================================
  Hits            210        210          
  Misses            8          8          
  Partials          7          7          
```

> Powered by [Codecov](https://codecov.io?src=pr). Last update [5e693ee...d6c493c](https://codecov.io/gh/aio-libs/aiomonitor/compare/5e693eef73645e2a1dbf021be6d812228b287ca5...d6c493c6ffa2805020b6456fa72e099d77810f74?src=pr)

Fix patch:
diff --git a/.editorconfig b/.editorconfig
new file mode 100644
index 00000000..30c32c7a
--- /dev/null
+++ b/.editorconfig
@@ -0,0 +1,28 @@
+root = true
+
+[*]
+end_of_line = lf
+charset = utf-8
+
+[*.sh]
+indent_style = space
+indent_size = 2
+insert_final_newline = true
+trim_trailing_whitespace = true
+
+[*.{py,md}]
+max_line_length = 88
+indent_style = space
+indent_size = 4
+insert_final_newline = true
+
+[*.py]
+trim_trailing_whitespace = true
+
+[*.md]
+trim_trailing_whitespace = false
+
+[*.rst]
+max_line_length = 0
+indent_style = space
+indent_size = 3
diff --git a/.flake8 b/.flake8
new file mode 100644
index 00000000..9581cd15
--- /dev/null
+++ b/.flake8
@@ -0,0 +1,11 @@
+[flake8]
+# ref: http://pep8.readthedocs.io/en/latest/intro.html#error-codes
+ignore = E203,E731,W503,W504,Q000
+# allow exeeding max-line-length because black sometimes fails to fit the code in the limit
+# when there are long string constants composed of paths and URLs
+extend-ignore = E501
+max-line-length = 88
+builtins = _
+exclude = .git,.cache,.idea,.egg,__pycache__,venv,dist,build,docs
+
+# vim: ft=dosini
diff --git a/.github/workflows/default.yml b/.github/workflows/default.yml
new file mode 100644
index 00000000..789b88ce
--- /dev/null
+++ b/.github/workflows/default.yml
@@ -0,0 +1,102 @@
+name: default
+
+on: [push, pull_request]
+
+jobs:
+  lint-flake8:
+    runs-on: ubuntu-latest
+    steps:
+    - uses: actions/checkout@v3
+      with:
+        fetch-depth: 2
+    - name: Set up Python
+      uses: actions/setup-python@v4
+      with:
+        python-version: "3.10"
+        check-latest: true
+        cache: pip
+        cache-dependency-path: |
+          setup.py
+          requirements-dev.txt
+          requirements-doc.txt
+    - name: Install
+      run: |
+        pip install -U pip setuptools wheel
+        pip install -U -r requirements-dev.txt
+    - name: Lint with flake8
+      run: |
+        echo "::add-matcher::.github/workflows/flake8-matcher.json"
+        python -m flake8 aiomonitor/ examples/ tests/ setup.py
+
+  lint-isort:
+    runs-on: ubuntu-latest
+    steps:
+    - uses: actions/checkout@v3
+      with:
+        fetch-depth: 2
+    - name: Set up Python
+      uses: actions/setup-python@v4
+      with:
+        python-version: "3.10"
+        check-latest: true
+        cache: pip
+        cache-dependency-path: |
+          setup.py
+          requirements-dev.txt
+          requirements-doc.txt
+    - name: Install
+      run: |
+        pip install -U pip setuptools wheel
+        pip install -U -r requirements-dev.txt
+    - name: Lint with isort
+      run: |
+        python -m isort --check aiomonitor/ examples/ tests/ setup.py
+
+  lint-black:
+    runs-on: ubuntu-latest
+    steps:
+    - uses: actions/checkout@v3
+      with:
+        fetch-depth: 2
+    - name: Set up Python
+      uses: actions/setup-python@v4
+      with:
+        python-version: "3.10"
+        check-latest: true
+        cache: pip
+        cache-dependency-path: |
+          setup.py
+          requirements-dev.txt
+          requirements-doc.txt
+    - name: Install
+      run: |
+        pip install -U pip setuptools wheel
+        pip install -U -r requirements-dev.txt
+    - name: Lint with black
+      run: |
+        python -m black --check aiomonitor/ examples/ tests/ setup.py
+
+  typecheck-mypy:
+    runs-on: ubuntu-latest
+    steps:
+    - uses: actions/checkout@v3
+      with:
+        fetch-depth: 2
+    - name: Set up Python
+      uses: actions/setup-python@v4
+      with:
+        python-version: "3.10"
+        check-latest: true
+        cache: pip
+        cache-dependency-path: |
+          setup.py
+          requirements-dev.txt
+          requirements-doc.txt
+    - name: Install
+      run: |
+        pip install -U pip setuptools wheel
+        pip install -U -r requirements-dev.txt
+    - name: Typecheck with mypy
+      run: |
+        echo "::add-matcher::.github/workflows/mypy-matcher.json"
+        python -m mypy aiomonitor/ examples/ tests/ setup.py
diff --git a/.github/workflows/flake8-matcher.json b/.github/workflows/flake8-matcher.json
new file mode 100644
index 00000000..676ada9c
--- /dev/null
+++ b/.github/workflows/flake8-matcher.json
@@ -0,0 +1,16 @@
+{
+    "problemMatcher": [
+        {
+            "owner": "flake8",
+            "pattern": [
+                {
+                    "regexp": "^([^:]*):(\\d+):(\\d+): (.*)$",
+                    "file": 1,
+                    "line": 2,
+                    "column": 3,
+                    "message": 4
+                }
+            ]
+        }
+    ]
+}
diff --git a/.github/workflows/mypy-matcher.json b/.github/workflows/mypy-matcher.json
new file mode 100644
index 00000000..ee743aa1
--- /dev/null
+++ b/.github/workflows/mypy-matcher.json
@@ -0,0 +1,16 @@
+{
+    "problemMatcher": [
+        {
+            "owner": "mypy",
+            "pattern": [
+                {
+                    "regexp": "^([^:]*):(\\d+): ([^:]+): (.*)$",
+                    "file": 1,
+                    "line": 2,
+                    "severity": 3,
+                    "message": 4
+                }
+            ]
+        }
+    ]
+}
diff --git a/.gitignore b/.gitignore
index 50021265..09710c0d 100644
--- a/.gitignore
+++ b/.gitignore
@@ -71,3 +71,7 @@ Pipfile.lock
 .tags
 
 .pytest_cache
+.python-version
+
+.DS_Store
+.vim/
diff --git a/.travis.yml b/.travis.yml
deleted file mode 100644
index 3e8a792a..00000000
--- a/.travis.yml
+++ /dev/null
@@ -1,36 +0,0 @@
-dist: xenial
-language: python
-
-
-python:
-  - '3.8'
-  - '3.6'
-
-matrix:
-  include:
-    - python: '3.7'
-      script:
-        - make mypy
-        - make cov
-
-install:
-    - pip install --upgrade setuptools
-    - pip install codecov
-    - pip install -r requirements-dev.txt
-
-script:
-  - make cov
-
-after_success:
-    codecov
-
-deploy:
-  provider: pypi
-  user: aio-libs-bot
-  password:
-    secure: "Wd5klsrMmt61+cNGOz3Gd/eh33knLS3oAaf1Xg31Tdtc/yL/oPVb9MyMFZdMxPJBJWyyrtJVF9G5vwSHTcJStaxHF6e2kIgVJbH4uBl/uFxct2Zu66WMlzfCPmjXhRZBNXnjlxa16y1YH9AmIhoSC32HezVOwqxGWm35Lgxn5fpTMG3rE4dYGQ4eAPVM1/2tM3D0BU3sSRvr0xRfqsbEGZojVKOUN2ywXyPpYqVSD3I7AQm7rz1ldYO/VyFCd4gjPhj3fcKp6QW/DGdB/loKIGq7oeeWJ/5YaUme+jZDNBv6/W2gxyIzdbHWkpQ/ZSR4GC8tDNLRlOZuEwKFGya3JBeYsFbBMaf9J8rqRT8I8Tnu9d4SZ/Yr4MFI4Xv/uug0XI1dUaN/d2lvN8LKt0AVfXDwxdaFNWCfjsdOdlGHsWD7vMbmRQ0p3w7Y0NyiwbodpocSPG648qU7LxWryzA8BxRiQgcBDqCFIIW+MEnGmKP9iKq2e6mCaXj7KuZl3RYdp4/VkuemDbOpd7vZM6g6iQHmnPjFXLBxB7s00oamBT8XlHBMW/KkIHNL2oMV8QCEBdlOvPKSY2PvOpcS9IddAV7R0q7zyzSsbDXA2TsJNs0+jIJzF4dDCyehAjFgW/ISlLwUs6sqjRaD/4vYqafFECFgNtyoHnS/Qs393ZeW02k="
-  distributions: sdist bdist_wheel
-  on:
-    tags: true
-    all_branches: true
-    python: 3.7
diff --git a/CHANGES.txt b/CHANGES.rst
similarity index 57%
rename from CHANGES.txt
rename to CHANGES.rst
index 04058aa8..b809fe38 100644
--- a/CHANGES.txt
+++ b/CHANGES.rst
@@ -1,10 +1,45 @@
 CHANGES
 =======
 
-0.4.6 (undefined)
-------------------
+0.5.0 (undefined)
+-----------------
+
+* Fix a regression in Python 3.10 due to #10 (#11)
+
+* Support Python 3.11 properly by allowing the optional `name` and `context` kwargs passed to `asyncio.create_task()` in the hooked task factory function (#10)
+
+* Update development dependencies
+
+* Selective persistent termination logs (#9)
+
+* Implement cancellation chain tracker (#8)
+
+* Trigger auto-completion only when Tab is pressed
+
+* Support auto-completion of commands and arguments (#7)
+
+* Add missing explicit dependency to Click
+
+* Promote `console_locals` as public attr
+
+* Reimplement console command (#6)
+
+* Migrate to Click-based command line interface (#5)
+
+* Adopt `prompt_toolkit` and support concurrent clients (#4)
+
+* Show the total number of tasks when executing `ps` (#3)
+
+* Apply black, isort, mypy, flake8 and automate CI workflows using GitHub Actions
+
+* Fix the task creation location in the 'ps' command output
+
+* Remove loop=loop from all asynchronous calls to support newer Python versions (#329)
+
+* Added the task creation stack chain display to the 'where' command by setting a custom task factory (#1)
 
-* Remove loop=loop from all asynchronous calls to support newer Python versions #329
+These are the backported changes from [aiomonitor-ng](https://github.com/achimnol/aiomonitor-ng).
+As the version bumps have gone far away in the fork, all those extra releases are squashed into the v0.5.0 release.
 
 
 0.4.5 (2019-11-03)
diff --git a/README.rst b/README.rst
index b0c937cf..7e729eaa 100644
--- a/README.rst
+++ b/README.rst
@@ -1,22 +1,10 @@
-aiomonitor
-==========
-.. image:: https://travis-ci.com/aio-libs/aiomonitor.svg?branch=master
-    :target: https://travis-ci.com/aio-libs/aiomonitor
-.. image:: https://codecov.io/gh/aio-libs/aiomonitor/branch/master/graph/badge.svg
-    :target: https://codecov.io/gh/aio-libs/aiomonitor
-.. image:: https://api.codeclimate.com/v1/badges/d14af4cfb0c4ff52b1ef/maintainability
-   :target: https://codeclimate.com/github/aio-libs/aiomonitor/maintainability
-   :alt: Maintainability
-.. image:: https://img.shields.io/pypi/v/aiomonitor.svg
-    :target: https://pypi.python.org/pypi/aiomonitor
-.. image:: https://readthedocs.org/projects/aiomonitor/badge/?version=latest
-    :target: http://aiomonitor.readthedocs.io/en/latest/?badge=latest
-    :alt: Documentation Status
-.. image:: https://badges.gitter.im/Join%20Chat.svg
-    :target: https://gitter.im/aio-libs/Lobby
-    :alt: Chat on Gitter
-
-**aiomonitor** is a Python 3.5+ module that adds monitor and cli capabilities
+aiomonitor-ng
+=============
+
+**aiomonitor-ng** is a (temporary) fork of **aiomonitor** with support for
+Python 3.10+ and additional usability & debuggability improvements.
+
+**aiomonitor** is a module that adds monitor and cli capabilities
 for asyncio_ applications. Idea and code were borrowed from curio_ project.
 Task monitor that runs concurrently to the asyncio_ loop (or fast drop-in
 replacement uvloop_) in a separate thread as result monitor will work even if
@@ -26,15 +14,13 @@ This library provides a python console using aioconsole_ module. It is possible
 to execute asynchronous commands inside your running application. Extensible
 with you own commands, in the style of the standard library's cmd_ module
 
-+--------------------------------------------------------------------------------------+
-| .. image:: https://raw.githubusercontent.com/aio-libs/aiomonitor/master/docs/tty.gif |
-+--------------------------------------------------------------------------------------+
+.. image:: https://raw.githubusercontent.com/achimnol/aiomonitor-ng/master/docs/screenshot-ps-where-example.png
 
 Installation
 ------------
 Installation process is simple, just::
 
-    $ pip install aiomonitor
+    $ pip install aiomonitor-ng
 
 
 Example
@@ -45,27 +31,26 @@ Monitor has context manager interface:
 
     import aiomonitor
 
-    loop = asyncio.get_event_loop()
-    with aiomonitor.start_monitor():
-        loop.run_forever()
-
-Now from separate terminal it is possible to connect to the application::
+    async def main():
+        loop = asyncio.get_running_loop()
+        run_forever = loop.create_future()
+        with aiomonitor.start_monitor(loop):
+            await run_forever
 
-    $ nc localhost 50101
-    
+    try:
+        asyncio.run(main())
+    except KeyboardInterrupt:
+        pass
 
-To make arrow keys working properly you can use the `rlwrap` trick:
+Now from separate terminal it is possible to connect to the application::
 
-    $ rlwrap nc localhost 50101
-    
+    $ telnet localhost 50101
 
 or the included python client::
 
     $ python -m aiomonitor.cli
     
     
-    
-
 Tutorial
 --------
 
@@ -108,37 +93,37 @@ Let's save this code in file ``simple_srv.py``, so we can run it with the follow
     (Press CTRL+C to quit)
 
 And now one can connect to a running application from a separate terminal, with
-the ``nc`` command, and ``aiomonitor`` will immediately respond with prompt::
+the ``telnet`` command, and ``aiomonitor`` will immediately respond with prompt::
 
-    $ nc localhost 50101
+    $ telnet localhost 50101
     Asyncio Monitor: 1 tasks running
     Type help for commands
     monitor >>>
 
-
-Note in order to make arrow keys and editing work properly just prepend command with `rlwrap`::
-
-    $ rlwrap nc localhost 50101
-
-
 Now you can type commands, for instance, ``help``::
 
     monitor >>> help
+    Usage: help [OPTIONS] COMMAND [ARGS]...
+    
+      To see the usage of each command, run them with "--help" option.
+    
     Commands:
-                 ps               : Show task table
-                 where taskid     : Show stack frames for a task
-                 cancel taskid    : Cancel an indicated task
-                 signal signame   : Send a Unix signal
-                 stacktrace       : Print a stack trace from the event loop thread
-                 console          : Switch to async Python REPL
-                 quit             : Leave the monitor
+      cancel                  Cancel an indicated task
+      console                 Switch to async Python REPL
+      exit (q,quit)           Leave the monitor client session
+      help (?,h)              Show the list of commands
+      ps (p)                  Show task table
+      ps-terminated (pst,pt)  List recently terminated/cancelled tasks
+      signal                  Send a Unix signal
+      stacktrace (st,stack)   Print a stack trace from the event loop thread
+      where (w)               Show stack frames and the task creation chain of a task
+      where-terminated (wt)   Show stack frames and the termination/cancellation chain of a task
 
 ``aiomonitor`` also supports async python console inside a running event loop
 so you can explore the state of your application::
 
     monitor >>> console
-    Python 3.5.2 (default, Oct 11 2016, 05:05:28)
-    [GCC 4.2.1 Compatible Apple LLVM 8.0.0 (clang-800.0.38)] on darwin
+    Python 3.10.7 (main, Sep  9 2022, 12:31:20) [Clang 13.1.6 (clang-1316.0.21.2.5)] on darwin
     Type "help", "copyright", "credits" or "license" for more information.
     ---
     This console is running in an asyncio event loop.
@@ -146,42 +131,48 @@ so you can explore the state of your application::
     Try: await asyncio.sleep(1, result=3)
     ---
     >>> await asyncio.sleep(1, result=3)
+    3
+    >>>
 
-To leave the console type ``exit()``::
+To leave the console type ``exit()`` or press Ctrl+D::
 
     >>> exit()
+
+    ✓ The console session is closed.
     monitor >>>
 
+Extension
+---------
 
-``aiomonitor`` is very easy to extend with your own console commands.
+Additional console variables
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
-.. code:: python
+You may add more variables that can be directly referenced in the ``console`` command.
+Refer `the console-variables example code <https://github.com/achimnol/aiomonitor-ng/tree/master/examples/console-variables.py>`_
 
-   class WebMonitor(aiomonitor.Monitor):
-       def do_hello(self, sin, sout, name=None):
-           """Using the /hello GET interface
-
-           There is one optional argument, "name".  This name argument must be
-           provided with proper URL excape codes, like %20 for spaces.
-           """
-           name = '' if name is None else '/' + name
-           r = requests.get('http://localhost:8090/hello' + name)
-           sout.write(r.text + '\n')
+Custom console commands
+~~~~~~~~~~~~~~~~~~~~~~~
 
+``aiomonitor`` is very easy to extend with your own console commands.
+Refer `the extension example code <https://github.com/achimnol/aiomonitor-ng/tree/master/examples/extension.py>`_
 
 Requirements
 ------------
 
-* Python_ 3.5+
+* Python_ 3.8+ (3.10.7+ recommended)
 * aioconsole_
+* Click_
+* prompt_toolkit_
 * uvloop_ (optional)
 
 
 .. _PEP492: https://www.python.org/dev/peps/pep-0492/
 .. _Python: https://www.python.org
 .. _aioconsole: https://github.com/vxgmichel/aioconsole
-.. _aiohttp: https://github.com/KeepSafe/aiohttp
-.. _asyncio: http://docs.python.org/3.5/library/asyncio.html
+.. _aiohttp: https://github.com/aio-libs/aiohttp
+.. _asyncio: http://docs.python.org/3/library/asyncio.html
+.. _Click: https://click.palletsprojects.com
 .. _curio: https://github.com/dabeaz/curio
+.. _prompt_toolkit: https://python-prompt-toolkit.readthedocs.io
 .. _uvloop: https://github.com/MagicStack/uvloop
 .. _cmd: http://docs.python.org/3/library/cmd.html
diff --git a/aiomonitor/__init__.py b/aiomonitor/__init__.py
index 3064ddef..69c83ff6 100644
--- a/aiomonitor/__init__.py
+++ b/aiomonitor/__init__.py
@@ -20,11 +20,23 @@
         m.close()
 """
 
-from .monitor import (Monitor, start_monitor,
-                      MONITOR_HOST, MONITOR_PORT, CONSOLE_PORT)
 from importlib.metadata import version
 
-
-__all__ = ('Monitor', 'start_monitor', 'MONITOR_HOST', 'MONITOR_PORT',
-           'CONSOLE_PORT')
-__version__ = version('aiomonitor')
+from .monitor import (
+    CONSOLE_PORT,
+    MONITOR_HOST,
+    MONITOR_PORT,
+    Monitor,
+    monitor_cli,
+    start_monitor,
+)
+
+__all__ = (
+    "Monitor",
+    "monitor_cli",
+    "start_monitor",
+    "MONITOR_HOST",
+    "MONITOR_PORT",
+    "CONSOLE_PORT",
+)
+__version__ = version("aiomonitor")
diff --git a/aiomonitor/cli.py b/aiomonitor/cli.py
index 1a06768e..c2582f1e 100644
--- a/aiomonitor/cli.py
+++ b/aiomonitor/cli.py
@@ -1,32 +1,43 @@
 import argparse
-import telnetlib
+import asyncio
 
 from .monitor import MONITOR_HOST, MONITOR_PORT
+from .telnet import TelnetClient
+
+
+async def async_monitor_client(host: str, port: int) -> None:
+    async with TelnetClient(host, port) as client:
+        await client.interact()
 
 
 def monitor_client(host: str, port: int) -> None:
-    tn = telnetlib.Telnet()
-    tn.open(host, port, timeout=1)
     try:
-        tn.interact()
+        asyncio.run(async_monitor_client(host, port))
     except KeyboardInterrupt:
         pass
-    finally:
-        tn.close()
 
 
 def main() -> None:
-    parser = argparse.ArgumentParser('usage: python -m aiomonitor [options]')
-    parser.add_argument('-H', '--host', dest='monitor_host',
-                        default=MONITOR_HOST, type=str,
-                        help='monitor host ip')
-
-    parser.add_argument('-p', '--port', dest='monitor_port',
-                        default=MONITOR_PORT, type=int,
-                        help='monitor port number')
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "-H",
+        "--host",
+        dest="monitor_host",
+        default=MONITOR_HOST,
+        type=str,
+        help="monitor host ip",
+    )
+    parser.add_argument(
+        "-p",
+        "--port",
+        dest="monitor_port",
+        default=MONITOR_PORT,
+        type=int,
+        help="monitor port number",
+    )
     args = parser.parse_args()
     monitor_client(args.monitor_host, args.monitor_port)
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     main()
diff --git a/aiomonitor/console.py b/aiomonitor/console.py
new file mode 100644
index 00000000..359ead51
--- /dev/null
+++ b/aiomonitor/console.py
@@ -0,0 +1,159 @@
+from __future__ import annotations
+
+import asyncio
+from typing import Any, Dict, Optional
+
+import aioconsole
+from prompt_toolkit import PromptSession
+from prompt_toolkit.input.base import Input
+from prompt_toolkit.output.base import Output
+
+
+class ConsoleProxy:
+    def __init__(
+        self,
+        stdin: Input,
+        stdout: Output,
+        host: str,
+        port: int,
+    ) -> None:
+        self._host = host
+        self._port = port
+        self._stdin = stdin
+        self._stdout = stdout
+
+    async def __aenter__(self) -> ConsoleProxy:
+        self._stdout.write_raw("\r\n")
+        self._conn_reader, self._conn_writer = await asyncio.open_connection(
+            self._host, self._port
+        )
+        self._closed = asyncio.Event()
+        self._recv_task = asyncio.create_task(self._handle_received())
+        self._input_task = asyncio.create_task(self._handle_user_input())
+        return self
+
+    async def __aexit__(self, *exc_info) -> Optional[bool]:
+        self._input_task.cancel()
+        await self._input_task
+        self._conn_writer.close()
+        try:
+            await self._conn_writer.wait_closed()
+        except NotImplementedError:
+            pass
+        self._conn_reader.feed_eof()
+        await self._recv_task
+        return None
+
+    async def interact(self) -> None:
+        try:
+            await self._closed.wait()
+        finally:
+            self._closed.set()
+            self._conn_writer.write_eof()
+
+    async def _handle_user_input(self) -> None:
+        prompt_session: PromptSession[str] = PromptSession(
+            input=self._stdin, output=self._stdout
+        )
+        try:
+            while not self._conn_reader.at_eof():
+                try:
+                    user_input = await prompt_session.prompt_async("")
+                    self._conn_writer.write(user_input.encode("utf8"))
+                    self._conn_writer.write(b"\n")
+                    await self._conn_writer.drain()
+                except KeyboardInterrupt:
+                    # Send Ctrl+C to the console server.
+                    self._conn_writer.write(b"\x03")
+                    await self._conn_writer.drain()
+                except EOFError:
+                    return
+        except asyncio.CancelledError:
+            pass
+        finally:
+            self._closed.set()
+
+    async def _handle_received(self) -> None:
+        try:
+            while True:
+                buf = await self._conn_reader.read(1024)
+                if not buf:
+                    return
+                self._stdout.write_raw(buf.decode("utf8"))
+                self._stdout.flush()
+        except asyncio.CancelledError:
+            pass
+        finally:
+            self._closed.set()
+
+
+async def start(
+    host: str,
+    port: int,
+    locals: Optional[Dict[str, Any]],
+    monitor_loop: asyncio.AbstractEventLoop,
+) -> asyncio.AbstractServer:
+    ui_loop = asyncio.get_running_loop()
+    done = asyncio.Event()
+
+    async def _start(done: asyncio.Event) -> asyncio.AbstractServer:
+        def _factory(streams: Any = None) -> aioconsole.AsynchronousConsole:
+            return aioconsole.AsynchronousConsole(
+                locals=locals, streams=streams, loop=monitor_loop
+            )
+
+        try:
+            server = await aioconsole.start_interactive_server(
+                host=host,
+                port=port,
+                factory=_factory,
+                loop=monitor_loop,
+            )
+        finally:
+            ui_loop.call_soon_threadsafe(done.set)
+        return server
+
+    console_future = asyncio.run_coroutine_threadsafe(
+        _start(done),
+        loop=monitor_loop,
+    )
+    try:
+        await done.wait()
+    finally:
+        server = console_future.result()
+    return server
+
+
+async def close(
+    server: asyncio.AbstractServer,
+    monitor_loop: asyncio.AbstractEventLoop,
+) -> None:
+    ui_loop = asyncio.get_running_loop()
+    done = asyncio.Event()
+
+    async def _close(done: asyncio.Event) -> None:
+        try:
+            server.close()
+            await server.wait_closed()
+        except NotImplementedError:
+            pass
+        finally:
+            ui_loop.call_soon_threadsafe(done.set)
+
+    close_future = asyncio.run_coroutine_threadsafe(
+        _close(done),
+        loop=monitor_loop,
+    )
+    try:
+        # NOTE: If the monitor is interrupted before the aioconsole session is closed,
+        #       run_coroutine_threadsafe() seems to get deadlocked.
+        #       In that case, you may need to send SIGINT multiple times to the
+        #       aiomonitor-running process.
+        await done.wait()
+    finally:
+        close_future.result()
+
+
+async def proxy(sin: Input, sout: Output, host: str, port: int) -> None:
+    async with ConsoleProxy(sin, sout, host, port) as proxy:
+        await proxy.interact()
diff --git a/aiomonitor/monitor.py b/aiomonitor/monitor.py
index 489461ce..d930d086 100644
--- a/aiomonitor/monitor.py
+++ b/aiomonitor/monitor.py
@@ -1,116 +1,286 @@
+from __future__ import annotations
+
 import asyncio
-import inspect
+import contextvars
+import functools
 import logging
 import os
+import shlex
 import signal
-import socket
 import sys
 import threading
+import time
 import traceback
-from textwrap import wrap
+import weakref
+from asyncio.coroutines import _format_coroutine  # type: ignore
+from contextvars import ContextVar, copy_context
+from datetime import timedelta
 from types import TracebackType
-from typing import (IO, Dict, Any, Callable, Optional, Tuple, Generator,  # noqa
-                    List, Type, TypeVar, NamedTuple, get_type_hints,  # noqa
-                    cast, Sequence)  # noqa
-from contextlib import suppress
-from concurrent.futures import Future  # noqa
-
+from typing import (
+    Any,
+    Awaitable,
+    Coroutine,
+    Dict,
+    Final,
+    Generator,
+    Iterable,
+    List,
+    Optional,
+    TextIO,
+    Tuple,
+    Type,
+    TypeVar,
+    cast,
+)
+
+import click
+import janus
+from click.parser import split_arg_string
+from click.shell_completion import CompletionItem, _resolve_context, _resolve_incomplete
+from prompt_toolkit import PromptSession
+from prompt_toolkit.application.current import get_app_session
+from prompt_toolkit.completion import CompleteEvent, Completer, Completion
+from prompt_toolkit.contrib.telnet.server import TelnetConnection, TelnetServer
+from prompt_toolkit.document import Document
+from prompt_toolkit.formatted_text import FormattedText
+from prompt_toolkit.shortcuts import print_formatted_text
 from terminaltables import AsciiTable
 
-from .utils import (_format_stack, cancel_task, task_by_id, console_proxy,
-                    init_console_server, close_server, alt_names, all_tasks)
-from .mypy_types import Loop, OptLocals
-
+from . import console
+from .task import TracedTask, persistent_coro
+from .types import CancellationChain, TerminatedTaskInfo
+from .utils import (
+    AliasGroupMixin,
+    _extract_stack_from_exception,
+    _extract_stack_from_frame,
+    _extract_stack_from_task,
+    _filter_stack,
+    _format_filename,
+    _format_task,
+    _format_terminated_task,
+    _format_timedelta,
+    all_tasks,
+    cancel_task,
+    get_default_args,
+    task_by_id,
+)
+
+__all__ = (
+    "Monitor",
+    "monitor_cli",
+    "start_monitor",
+)
 
-__all__ = ('Monitor', 'start_monitor')
 log = logging.getLogger(__name__)
+current_monitor: ContextVar[Monitor] = ContextVar("current_monitor")
+current_stdout: ContextVar[TextIO] = ContextVar("current_stdout")
+command_done: ContextVar[asyncio.Event] = ContextVar("command_done")
 
+MONITOR_HOST: Final = "127.0.0.1"
+MONITOR_PORT: Final = 50101
+CONSOLE_PORT: Final = 50102
 
-MONITOR_HOST = '127.0.0.1'
-MONITOR_PORT = 50101
-CONSOLE_PORT = 50102
-
-
-Server = asyncio.AbstractServer  # noqa
+T = TypeVar("T")
+T_co = TypeVar("T_co", covariant=True)
 
 
-class CommandException(Exception):
+@click.group(cls=AliasGroupMixin, add_help_option=False)
+def monitor_cli():
+    """
+    To see the usage of each command, run them with "--help" option.
+    """
     pass
 
 
-class UnknownCommandException(CommandException):
-    pass
-
-
-class MultipleCommandException(CommandException):
-    def __init__(self, cmds: List['CmdName']) -> None:
-        self.cmds = cmds
-        super().__init__()
-
-
-class ArgumentMappingException(CommandException):
-    pass
+def _get_current_stdout() -> TextIO:
+    stdout = current_stdout.get(None)
+    if stdout is None:
+        return sys.stdout
+    else:
+        return stdout
+
+
+def _get_current_stderr() -> TextIO:
+    stdout = current_stdout.get(None)
+    if stdout is None:
+        return sys.stderr
+    else:
+        return stdout
+
+
+class ClickCompleter(Completer):
+    def __init__(self, root_command: click.BaseCommand) -> None:
+        self._root_command = root_command
+
+    def get_completions(
+        self,
+        document: Document,
+        complete_event: CompleteEvent,
+    ) -> Iterable[Completion]:
+        args = split_arg_string(document.current_line)
+        incomplete = document.get_word_under_cursor()
+        if incomplete and args and args[-1] == incomplete:
+            args.pop()
+        click_ctx = _resolve_context(self._root_command, {}, "", args)
+        cmd_or_param, incomplete = _resolve_incomplete(click_ctx, args, incomplete)
+        completions: List[CompletionItem] = cmd_or_param.shell_complete(
+            click_ctx, incomplete
+        )
+        start_position = document.find_backwards(incomplete) or 0
+        for completion in completions:
+            yield Completion(
+                completion.value,
+                start_position=start_position,
+            )
 
 
-CmdName = NamedTuple('CmdName', [('cmd_name', str), ('method_name', str)])
+def complete_task_id(
+    ctx: click.Context,
+    param: click.Parameter,
+    incomplete: str,
+) -> Iterable[str]:
+    # ctx here is created in the completer and does not have ctx.obj set as monitor.
+    # We take the monitor instance from the global context variable instead.
+    try:
+        self: Monitor = current_monitor.get()
+    except LookupError:
+        return []
+    return [
+        task_id
+        for task_id in map(str, sorted(map(id, all_tasks(loop=self._monitored_loop))))
+        if task_id.startswith(incomplete)
+    ][:10]
+
+
+def complete_trace_id(
+    ctx: click.Context,
+    param: click.Parameter,
+    incomplete: str,
+) -> Iterable[str]:
+    try:
+        self: Monitor = current_monitor.get()
+    except LookupError:
+        return []
+    return [
+        trace_id
+        for trace_id in map(str, sorted(self._terminated_tasks.keys()))
+        if trace_id.startswith(incomplete)
+    ][:10]
+
+
+def complete_signal_names(
+    ctx: click.Context,
+    param: click.Parameter,
+    incomplete: str,
+) -> Iterable[str]:
+    return [sig.name for sig in signal.Signals if sig.name.startswith(incomplete)]
 
 
 class Monitor:
-    _event_loop_thread_id = None  # type: int
-    _cmd_prefix = 'do_'
+    _event_loop_thread_id: Optional[int] = None
+    _cmd_prefix = "do_"
     _empty_result = object()
 
-    prompt = 'monitor >>> '
-    intro = '\nAsyncio Monitor: {tasknum} task{s} running\nType help for available commands\n\n'  # noqa
-    help_template = '{cmd_name} {arg_list}\n    {doc}\n'
-    help_short_template = '    {cmd_name}{cmd_arg_sep}{arg_list}: {doc_firstline}'  # noqa
-
-    _sin = None  # type: IO[str]
-    _sout = None  # type: IO[str]
-
-    def __init__(self,
-                 loop: asyncio.AbstractEventLoop, *,
-                 host: str = MONITOR_HOST,
-                 port: int = MONITOR_PORT,
-                 console_port: int = CONSOLE_PORT,
-                 console_enabled: bool = True,
-                 locals: OptLocals = None) -> None:
-        self._loop = loop or asyncio.get_event_loop()
+    intro = "\nAsyncio Monitor: {tasknum} task{s} running\nType help for available commands\n"  # noqa
+    help_template = "{cmd_name} {arg_list}\n    {doc}\n"
+    help_short_template = (
+        "    {cmd_name}{cmd_arg_sep}{arg_list}: {doc_firstline}"  # noqa
+    )
+
+    console_locals: Dict[str, Any]
+    _console_tasks: weakref.WeakSet[asyncio.Task[Any]]
+
+    _created_traceback_chains: weakref.WeakKeyDictionary[
+        asyncio.Task[Any],
+        weakref.ReferenceType[asyncio.Task[Any]],
+    ]
+    _created_tracebacks: weakref.WeakKeyDictionary[
+        asyncio.Task[Any], List[traceback.FrameSummary]
+    ]
+    _terminated_tasks: Dict[str, TerminatedTaskInfo]
+    _terminated_history: List[str]
+    _termination_info_queue: janus.Queue[TerminatedTaskInfo]
+    _canceller_chain: Dict[str, str]
+    _canceller_stacks: Dict[str, List[traceback.FrameSummary] | None]
+    _cancellation_chain_queue: janus.Queue[CancellationChain]
+
+    def __init__(
+        self,
+        loop: asyncio.AbstractEventLoop,
+        *,
+        host: str = MONITOR_HOST,
+        port: int = MONITOR_PORT,
+        console_port: int = CONSOLE_PORT,
+        console_enabled: bool = True,
+        hook_task_factory: bool = False,
+        max_termination_history: int = 1000,
+        locals: Optional[Dict[str, Any]] = None,
+    ) -> None:
+        self._monitored_loop = loop or asyncio.get_running_loop()
         self._host = host
         self._port = port
         self._console_port = console_port
         self._console_enabled = console_enabled
-        self._locals = locals
+        if locals is None:
+            self.console_locals = {"__name__": "__console__", "__doc__": None}
+        else:
+            self.console_locals = locals
 
-        log.info('Starting aiomonitor at %s:%d', host, port)
+        self.prompt = "monitor >>> "
+
+        log.info("Starting aiomonitor at %s:%d", host, port)
 
-        self._ui_thread = threading.Thread(target=self._server, args=(),
-                                           daemon=True)
-        self._closing = threading.Event()
         self._closed = False
         self._started = False
-        self._console_future = None  # type: Optional[Future[Any]]
+        self._console_tasks = weakref.WeakSet()
+
+        self._hook_task_factory = hook_task_factory
+        self._created_traceback_chains = weakref.WeakKeyDictionary()
+        self._created_tracebacks = weakref.WeakKeyDictionary()
+        self._terminated_tasks = {}
+        self._canceller_chain = {}
+        self._canceller_stacks = {}
+        self._terminated_history = []
+        self._max_termination_history = max_termination_history
 
-        self.lastcmd = None  # type: Optional[str]
+        self._ui_started = threading.Event()
+        self._ui_thread = threading.Thread(target=self._ui_main, args=(), daemon=True)
+
+    @property
+    def host(self) -> str:
+        return self._host
+
+    @property
+    def port(self) -> int:
+        return self._port
 
     def __repr__(self) -> str:
         name = self.__class__.__name__
-        return '<{name}: {host}:{port}>'.format(
-            name=name, host=self._host, port=self._port)
+        return "<{name}: {host}:{port}>".format(
+            name=name, host=self._host, port=self._port
+        )
 
     def start(self) -> None:
         assert not self._closed
         assert not self._started
-
         self._started = True
+        self._original_task_factory = self._monitored_loop.get_task_factory()
+        if self._hook_task_factory:
+            self._monitored_loop.set_task_factory(self._create_task)
         self._event_loop_thread_id = threading.get_ident()
         self._ui_thread.start()
+        self._ui_started.wait()
+
+        # Override the Click's stdout/stderr reference cache functions
+        # to let them use the correct stdout handler.
+        click.utils._default_text_stdout = _get_current_stdout
+        click.utils._default_text_stderr = _get_current_stderr
 
     @property
     def closed(self) -> bool:
         return self._closed
 
-    def __enter__(self) -> 'Monitor':
+    def __enter__(self) -> Monitor:
         if not self._started:
             self.start()
         return self
@@ -118,310 +288,646 @@ def __enter__(self) -> 'Monitor':
     # exc_type should be Optional[Type[BaseException]], but
     # this runs into https://github.com/python/typing/issues/266
     # on Python 3.5.
-    def __exit__(self, exc_type: Any,
-                 exc_value: Optional[BaseException],
-                 traceback: Optional[TracebackType]) -> None:
+    def __exit__(
+        self,
+        exc_type: Any,
+        exc_value: Optional[BaseException],
+        traceback: Optional[TracebackType],
+    ) -> None:
         self.close()
 
     def close(self) -> None:
+        assert self._started, "The monitor must have been started to close it."
         if not self._closed:
-            self._closing.set()
+            self._ui_loop.call_soon_threadsafe(
+                self._ui_forever_future.cancel,
+            )
             self._ui_thread.join()
+            self._monitored_loop.set_task_factory(self._original_task_factory)
             self._closed = True
 
-    def _server(self) -> None:
-        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    async def _coro_wrapper(self, coro: Awaitable[T_co]) -> T_co:
+        myself = asyncio.current_task()
+        assert isinstance(myself, TracedTask)
+        try:
+            return await coro
+        except BaseException as e:
+            myself._termination_stack = _extract_stack_from_exception(e)[:-1]
+            raise
+
+    def _create_task(
+        self,
+        loop: asyncio.AbstractEventLoop,
+        coro: Coroutine[Any, Any, T_co] | Generator[Any, None, T_co],
+        *,
+        name: str | None = None,
+        context: contextvars.Context | None = None,
+    ) -> asyncio.Future[T_co]:
+        assert loop is self._monitored_loop
+        try:
+            parent_task = asyncio.current_task()
+        except RuntimeError:
+            parent_task = None
+        persistent = coro in persistent_coro
+        task = TracedTask(
+            self._coro_wrapper(coro),  # type: ignore
+            termination_info_queue=self._termination_info_queue.sync_q,
+            cancellation_chain_queue=self._cancellation_chain_queue.sync_q,
+            persistent=persistent,
+            loop=self._monitored_loop,
+            name=name,  # since Python 3.8
+            context=context,  # since Python 3.11
+        )
+        task._orig_coro = cast(Coroutine[Any, Any, T_co], coro)
+        self._created_tracebacks[task] = _extract_stack_from_frame(sys._getframe())[
+            :-1
+        ]  # strip this wrapper method
+        if parent_task is not None:
+            self._created_traceback_chains[task] = weakref.ref(parent_task)
+        return task
+
+    def _ui_main(self) -> None:
+        asyncio.run(self._ui_main_async())
+
+    async def _ui_main_async(self) -> None:
+        loop = asyncio.get_running_loop()
+        self._termination_info_queue = janus.Queue()
+        self._cancellation_chain_queue = janus.Queue()
+        self._ui_loop = loop
+        self._ui_forever_future = loop.create_future()
+        self._ui_termination_handler_task = asyncio.create_task(
+            self._ui_handle_termination_updates()
+        )
+        self._ui_cancellation_handler_task = asyncio.create_task(
+            self._ui_handle_cancellation_updates()
+        )
+        telnet_server = TelnetServer(
+            interact=self._interact, host=self._host, port=self._port
+        )
+        telnet_server.start()
+        await asyncio.sleep(0)
+        self._ui_started.set()
         try:
-            sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
-        except AttributeError:
+            await self._ui_forever_future
+        except asyncio.CancelledError:
             pass
+        finally:
+            console_tasks = {*self._console_tasks}
+            for console_task in console_tasks:
+                console_task.cancel()
+            await asyncio.gather(*console_tasks, return_exceptions=True)
+            self._ui_termination_handler_task.cancel()
+            self._ui_cancellation_handler_task.cancel()
+            await self._ui_termination_handler_task
+            await self._ui_cancellation_handler_task
+            await telnet_server.stop()
+
+    async def _ui_handle_termination_updates(self) -> None:
+        while True:
+            try:
+                update: TerminatedTaskInfo = (
+                    await self._termination_info_queue.async_q.get()
+                )
+            except asyncio.CancelledError:
+                return
+            self._terminated_tasks[update.id] = update
+            if not update.persistent:
+                self._terminated_history.append(update.id)
+            # canceller stack is already put in _ui_handle_cancellation_updates()
+            if canceller_stack := self._canceller_stacks.pop(update.id, None):
+                update.canceller_stack = canceller_stack
+            while len(self._terminated_history) > self._max_termination_history:
+                removed_id = self._terminated_history.pop(0)
+                self._terminated_tasks.pop(removed_id, None)
+                self._canceller_chain.pop(removed_id, None)
+                self._canceller_stacks.pop(removed_id, None)
+
+    async def _ui_handle_cancellation_updates(self) -> None:
+        while True:
+            try:
+                update: CancellationChain = (
+                    await self._cancellation_chain_queue.async_q.get()
+                )
+            except asyncio.CancelledError:
+                return
+            self._canceller_stacks[update.target_id] = update.canceller_stack
+            self._canceller_chain[update.target_id] = update.canceller_id
+
+    def print_ok(self, msg: str) -> None:
+        print_formatted_text(
+            FormattedText(
+                [
+                    ("ansibrightgreen", "✓ "),
+                    ("", msg),
+                ]
+            )
+        )
+
+    def print_fail(self, msg: str) -> None:
+        print_formatted_text(
+            FormattedText(
+                [
+                    ("ansibrightred", "✗ "),
+                    ("", msg),
+                ]
+            )
+        )
 
-        # set the timeout to prevent the server loop from
-        # blocking indefinitaly on sock.accept()
-        sock.settimeout(0.5)
-        sock.bind((self._host, self._port))
-        sock.listen(1)
-        with sock:
-            while not self._closing.is_set():
-                try:
-                    client, addr = sock.accept()
-                    with client:
-                        sout = client.makefile('w', encoding='utf-8')
-                        sin = client.makefile('r', encoding='utf-8')
-                        self._interactive_loop(sin, sout)
-                except (socket.timeout, OSError):
-                    continue
-
-    def _interactive_loop(self, sin: IO[str], sout: IO[str]) -> None:
-        """Main interactive loop of the monitor"""
-        self._sin = sin
-        self._sout = sout
-        tasknum = len(all_tasks(loop=self._loop))
-        s = '' if tasknum == 1 else 's'
-        self._sout.write(self.intro.format(tasknum=tasknum, s=s))
+    async def _interact(self, connection: TelnetConnection) -> None:
+        """
+        The interactive loop for each telnet client connection.
+        """
+        await asyncio.sleep(0.3)  # wait until telnet negotiation is done
+        tasknum = len(all_tasks(loop=self._monitored_loop))
+        s = "" if tasknum == 1 else "s"
+        print(self.intro.format(tasknum=tasknum, s=s), file=connection.stdout)
+        current_monitor_token = current_monitor.set(self)
+        current_stdout_token = current_stdout.set(connection.stdout)
+        # NOTE: prompt_toolkit's all internal console output automatically uses
+        #       an internal contextvar to keep the stdout consistent with the
+        #       current telnet connection.
+        prompt_session: PromptSession[str] = PromptSession(
+            completer=ClickCompleter(monitor_cli),
+            complete_while_typing=False,
+        )
+        lastcmd = "noop"
+        style_prompt = "#5fd7ff bold"
         try:
-            while not self._closing.is_set():
-                self._sout.write(self.prompt)
-                self._sout.flush()
+            while True:
                 try:
-                    user_input = sin.readline()
-                    if not user_input:
-                        break
-                    user_input = user_input.strip()
-                except Exception as e:
-                    msg = 'Could not read from user input due to:\n{}\n'
-                    log.exception(msg)
-                    self._sout.write(msg.format(repr(e)))
-                    self._sout.flush()
+                    user_input = (
+                        await prompt_session.prompt_async(
+                            FormattedText(
+                                [
+                                    (style_prompt, self.prompt),
+                                ]
+                            )
+                        )
+                    ).strip()
+                except KeyboardInterrupt:
+                    self.print_fail("To terminate, press Ctrl+D or type 'exit'.")
+                except (EOFError, asyncio.CancelledError):
+                    return
+                except Exception:
+                    self.print_fail(traceback.format_exc())
                 else:
+                    command_done_event = asyncio.Event()
+                    command_done_token = command_done.set(command_done_event)
                     try:
-                        self._command_dispatch(user_input)
-                    except Exception as e:
-                        msg = 'Unexpected Exception during command execution:\n{}\n'  # noqa
-                        log.exception(msg)
-                        self._sout.write(msg.format(repr(e)))
-                        self._sout.flush()
+                        if not user_input and lastcmd is not None:
+                            user_input = lastcmd
+                        args = shlex.split(user_input)
+                        term_size = prompt_session.output.get_size()
+                        ctx = copy_context()
+                        ctx.run(
+                            monitor_cli.main,
+                            args,
+                            prog_name="",
+                            obj=self,
+                            standalone_mode=False,  # type: ignore
+                            max_content_width=term_size.columns,
+                        )
+                        await command_done_event.wait()
+                        if args[0] == "console":
+                            lastcmd = "noop"
+                        else:
+                            lastcmd = user_input
+                    except (click.BadParameter, click.UsageError) as e:
+                        self.print_fail(str(e))
+                    except asyncio.CancelledError:
+                        return
+                    except Exception:
+                        self.print_fail(traceback.format_exc())
+                    finally:
+                        command_done.reset(command_done_token)
         finally:
-            self._sin = None  # type: ignore
-            self._sout = None  # type: ignore
+            current_stdout.reset(current_stdout_token)
+            current_monitor.reset(current_monitor_token)
 
-    def _command_dispatch(self, user_input: str) -> None:
-        if not user_input:
-            return self.emptyline()
 
-        self.lastcmd = user_input
-        comm, *args = user_input.split(' ')
+def auto_command_done(cmdfunc):
+    @functools.wraps(cmdfunc)
+    def _inner(ctx: click.Context, *args, **kwargs):
+        command_done_event = command_done.get()
         try:
-            cmd, args = self.precmd(comm, args)
-            result = cmd(*args)
-        except UnknownCommandException as e:
-            result = self._empty_result
-            caught_ex = e  # type: Optional[Exception]
-            self.default(comm, *args)
-        except MultipleCommandException as e:
-            result = self._empty_result
-            caught_ex = e
-            msg = 'Ambiguous command "{}"'.format(comm)
-            self._sout.write(msg + '\n')
-        except ArgumentMappingException as e:
-            result = self._empty_result
-            caught_ex = e
-            msg = 'An argument to {} could not be converted according to the methods type annotation because of this error:\n{}\n'  # noqa
-            self._sout.write(msg.format(e, repr(e.__cause__)))
-        except TypeError as e:
-            result = self._empty_result
-            caught_ex = e
-            msg = 'Probably incorrect number of arguments to command method:\n{}\n'  # noqa
-            self._sout.write(msg.format(repr(e)))
-        except Exception as e:
-            result = self._empty_result
-            caught_ex = e
-        else:
-            caught_ex = None
+            return cmdfunc(ctx, *args, **kwargs)
         finally:
-            self.postcmd(comm, args, result, caught_ex)
-
-    def _filter_cmds(self, *,
-                     startswith: str = '',
-                     with_alts: bool = True) -> Generator[CmdName, None, None]:
-        cmds = (cmd for cmd in dir(self) if cmd.startswith(self._cmd_prefix))
-        for name in cmds:
-            if name.startswith(self._cmd_prefix + startswith):
-                yield CmdName(name[len(self._cmd_prefix):], name)
-            meth = getattr(self, name)
-            if with_alts and hasattr(meth, 'alt_names'):
-                for altname in meth.alt_names:
-                    if altname.startswith(startswith):
-                        yield CmdName(altname, name)
-
-    def map_args(self, cmd: Callable[..., Any], args: Sequence[str]
-                 ) -> Generator[Any, None, None]:
-        params = inspect.signature(cmd).parameters.values()
-        ia = iter(args)
-        for param in params:
-            if (param.annotation is param.empty or
-                    not callable(param.annotation)):
-
-                def type_(x: Any) -> Any:
-                    return x
-
-            else:
-                type_ = param.annotation
-            try:
-                if str(param).startswith('*'):
-                    for arg in ia:
-                        yield type_(arg)
-                else:
-                    # We iterate over the functions' annotation for its
-                    # parameters and also manually over the given arguments
-                    # (they can have arbitrarily different lengths).
-                    # Here we could be in the situation where a further
-                    # parameter exists, but no argument is given to it.
-                    # Since we might have a method with optional, non-star
-                    # arguments, we must ignore a StopIteration from this call
-                    # to next.
-                    with suppress(StopIteration):
-                        yield type_(next(ia))
-            except Exception as e:
-                raise ArgumentMappingException(cmd.__name__) from e
-        if tuple(ia):
-            msg = 'Too many arguments for command {}()'
-            raise TypeError(msg.format(cmd.__name__))
-
-    def precmd(self, comm: str, args: Sequence[str]
-               ) -> Tuple[Callable[..., Any], List[str]]:
-        cmd = self.getcmd(comm)
-        return cmd, list(self.map_args(cmd, args))
-
-    def postcmd(self,
-                comm: str,
-                args: Sequence[str],
-                result: Any,
-                exception: Optional[Exception] = None) -> None:
-        if (exception is not None
-                and not isinstance(exception, (CommandException, TypeError))):
-            raise exception
-
-    def getcmd(self, comm: str) -> Callable[..., Any]:
-        allcmds = sorted(self._filter_cmds(startswith=comm))
-        if not allcmds:
-            raise UnknownCommandException(comm)
-        if len(allcmds) > 1 and allcmds[0].cmd_name != comm:
-            raise MultipleCommandException(allcmds)
-        return getattr(self, allcmds[0].method_name)  # type: ignore
-
-    def emptyline(self) -> None:
-        if self.lastcmd is not None:
-            self._command_dispatch(self.lastcmd)
-
-    def default(self, comm: str, *args: str) -> None:
-        self._sout.write('No such command: {}\n'.format(comm))
-
-    @alt_names('? h')
-    def do_help(self, *cmd_names: str) -> None:
-        """Show help for command name
-
-        Any number of command names may be given to help, and the long help
-        text for all of them will be shown.
-        """
-        def _h(cmd: str, template: str) -> None:
-            try:
-                func = getattr(self, cmd)
-            except AttributeError:
-                self._sout.write('No such command: {}\n'.format(cmd))
-            else:
-                doc = func.__doc__ if func.__doc__ else ''
-                doc_firstline = doc.split('\n', maxsplit=1)[0]
-                arg_list = ' '.join(
-                    p for p in inspect.signature(func).parameters)
-                self._sout.write(
-                    template.format(
-                        cmd_name=cmd[len(self._cmd_prefix):],
-                        arg_list=arg_list,
-                        cmd_arg_sep=' ' if arg_list else '',
-                        doc=doc,
-                        doc_firstline=doc_firstline
-                    ) + '\n'
-                )
+            command_done_event.set()
 
-        if not cmd_names:
-            cmds = sorted(
-                c.method_name for c in self._filter_cmds(with_alts=False)
-            )
-            self._sout.write('Available Commands are:\n\n')
-            for cmd in cmds:
-                _h(cmd, self.help_short_template)
-        else:
-            for cmd in cmd_names:
-                _h(self._cmd_prefix + cmd, self.help_template)
-
-    @alt_names('p')
-    def do_ps(self) -> None:
-        """Show task table"""
-        headers = ('Task ID', 'State', 'Task')
-        table_data = [headers]
-        for task in sorted(all_tasks(loop=self._loop), key=id):
-            taskid = str(id(task))
-            if task:
-                t = '\n'.join(wrap(str(task), 80))
-                table_data.append((taskid, task._state, t))
-        table = AsciiTable(table_data)
-        self._sout.write(table.table)
-        self._sout.write('\n')
-        self._sout.flush()
-
-    @alt_names('w')
-    def do_where(self, taskid: int) -> None:
-        """Show stack frames for a task"""
-        task = task_by_id(taskid, self._loop)
-        if task:
-            self._sout.write(_format_stack(task))
-            self._sout.write('\n')
-        else:
-            self._sout.write('No task %d\n' % taskid)
+    return _inner
 
-    def do_signal(self, signame: str) -> None:
-        """Send a Unix signal"""
-        if hasattr(signal, signame):
-            os.kill(os.getpid(), getattr(signal, signame))
-        else:
-            self._sout.write('Unknown signal %s\n' % signame)
-
-    @alt_names('st')
-    def do_stacktrace(self) -> None:
-        """Print a stack trace from the event loop thread"""
-        frame = sys._current_frames()[self._event_loop_thread_id]
-        traceback.print_stack(frame, file=self._sout)
-
-    def do_cancel(self, taskid: int) -> None:
-        """Cancel an indicated task"""
-        task = task_by_id(taskid, self._loop)
-        if task:
-            fut = asyncio.run_coroutine_threadsafe(
-                cancel_task(task), loop=self._loop)
-            fut.result(timeout=3)
-            self._sout.write('Cancel task %d\n' % taskid)
-        else:
-            self._sout.write('No task %d\n' % taskid)
-
-    @alt_names('quit q')
-    def do_exit(self) -> None:
-        """Leave the monitor"""
-        self._sout.write('Leaving monitor. Hit Ctrl-C to exit\n')
-        self._sout.flush()
-
-    def do_console(self) -> None:
-        """Switch to async Python REPL"""
-        if not self._console_enabled:
-            self._sout.write('Python console disabled for this sessiong\n')
-            self._sout.flush()
-            return
 
-        h, p = self._host, self._console_port
-        log.info('Starting console at %s:%d', h, p)
-        fut = init_console_server(
-            self._host, self._console_port, self._locals, self._loop)
-        server = fut.result(timeout=3)
+def auto_async_command_done(cmdfunc):
+    @functools.wraps(cmdfunc)
+    async def _inner(ctx: click.Context, *args, **kwargs):
+        command_done_event = command_done.get()
         try:
-            console_proxy(
-                self._sin, self._sout, self._host, self._console_port)
+            return await cmdfunc(ctx, *args, **kwargs)
         finally:
-            coro = close_server(server)
-            close_fut = asyncio.run_coroutine_threadsafe(coro, loop=self._loop)
-            close_fut.result(timeout=15)
+            command_done_event.set()
 
+    return _inner
 
-def start_monitor(loop: Loop, *,
-                  monitor: Type[Monitor] = Monitor,
-                  host: str = MONITOR_HOST,
-                  port: int = MONITOR_PORT,
-                  console_port: int = CONSOLE_PORT,
-                  console_enabled: bool = True,
-                  locals: OptLocals = None) -> Monitor:
 
-    m = monitor(loop, host=host, port=port, console_port=console_port,
-                console_enabled=console_enabled, locals=locals)
-    m.start()
+def custom_help_option(cmdfunc):
+    """
+    A custom help option to ensure setting `command_done_event`.
+    """
+
+    @auto_command_done
+    def show_help(ctx: click.Context, param: click.Parameter, value: bool) -> None:
+        if not value:
+            return
+        click.echo(ctx.get_help(), color=ctx.color)
+        ctx.exit()
+
+    return click.option(
+        "--help",
+        is_flag=True,
+        expose_value=False,
+        is_eager=True,
+        callback=show_help,
+        help="Show the help message",
+    )(cmdfunc)
+
+
+@monitor_cli.command(name="noop", hidden=True)
+@auto_command_done
+def do_noop(ctx: click.Context) -> None:
+    pass
 
+
+@monitor_cli.command(name="help", aliases=["?", "h"])
+@custom_help_option
+@auto_command_done
+def do_help(ctx: click.Context) -> None:
+    """Show the list of commands"""
+    click.echo(monitor_cli.get_help(ctx))
+
+
+@monitor_cli.command(name="signal")
+@click.argument("signame", type=str, shell_complete=complete_signal_names)
+@custom_help_option
+@auto_command_done
+def do_signal(ctx: click.Context, signame: str) -> None:
+    """Send a Unix signal"""
+    self: Monitor = ctx.obj
+    if hasattr(signal, signame):
+        os.kill(os.getpid(), getattr(signal, signame))
+        self.print_ok(f"Sent signal to {signame} PID {os.getpid()}")
+    else:
+        self.print_fail(f"Unknown signal {signame}")
+
+
+@monitor_cli.command(name="stacktrace", aliases=["st", "stack"])
+@custom_help_option
+@auto_command_done
+def do_stacktrace(ctx: click.Context) -> None:
+    """Print a stack trace from the event loop thread"""
+    self: Monitor = ctx.obj
+    stdout = _get_current_stdout()
+    tid = self._event_loop_thread_id
+    assert tid is not None
+    frame = sys._current_frames()[tid]
+    traceback.print_stack(frame, file=stdout)
+
+
+@monitor_cli.command(name="cancel")
+@click.argument("taskid", shell_complete=complete_task_id)
+@custom_help_option
+@auto_command_done
+def do_cancel(ctx: click.Context, taskid: str) -> None:
+    """Cancel an indicated task"""
+    self: Monitor = ctx.obj
+    task_id = int(taskid)
+    task = task_by_id(task_id, self._monitored_loop)
+    if task:
+        fut = asyncio.run_coroutine_threadsafe(
+            cancel_task(task), loop=self._monitored_loop
+        )
+        fut.result(timeout=3)
+        self.print_ok(f"Cancelled task {task_id}")
+    else:
+        self.print_fail(f"No task {task_id}")
+
+
+@monitor_cli.command(name="exit", aliases=["q", "quit"])
+@custom_help_option
+@auto_command_done
+def do_exit(ctx: click.Context) -> None:
+    """Leave the monitor client session"""
+    raise asyncio.CancelledError("exit by user")
+
+
+@monitor_cli.command(name="console")
+@custom_help_option
+def do_console(ctx: click.Context) -> None:
+    """Switch to async Python REPL"""
+    self: Monitor = ctx.obj
+    if not self._console_enabled:
+        self.print_fail("Python console is disabled for this session!")
+        return
+
+    @auto_async_command_done
+    async def _console(ctx: click.Context) -> None:
+        h, p = self._host, self._console_port
+        log.info("Starting aioconsole at %s:%d", h, p)
+        app_session = get_app_session()
+        server = await console.start(
+            self._host,
+            self._console_port,
+            self.console_locals,
+            self._monitored_loop,
+        )
+        try:
+            await console.proxy(
+                app_session.input,
+                app_session.output,
+                self._host,
+                self._console_port,
+            )
+        except asyncio.CancelledError:
+            return
+        finally:
+            await console.close(server, self._monitored_loop)
+            log.info("Terminated aioconsole at %s:%d", h, p)
+            self.print_ok("The console session is closed.")
+
+    # Since we are already inside the UI's event loop,
+    # spawn the async command function as a new task and let it
+    # set `command_done_event` internally.
+    task = asyncio.create_task(_console(ctx))
+    self._console_tasks.add(task)
+
+
+@monitor_cli.command(name="ps", aliases=["p"])
+@click.option("-f", "--filter", "filter_", help="filter by coroutine or task name")
+@click.option("-p", "--persistent", is_flag=True, help="show only persistent tasks")
+@custom_help_option
+@auto_command_done
+def do_ps(
+    ctx: click.Context,
+    filter_: str,
+    persistent: bool,
+) -> None:
+    """Show task table"""
+    headers = (
+        "Task ID",
+        "State",
+        "Name",
+        "Coroutine",
+        "Created Location",
+        "Since",
+    )
+    self: Monitor = ctx.obj
+    stdout = _get_current_stdout()
+    table_data: List[Tuple[str, str, str, str, str, str]] = [headers]
+    all_running_tasks = all_tasks(loop=self._monitored_loop)
+
+    for task in sorted(all_running_tasks, key=id):
+        taskid = str(id(task))
+        if isinstance(task, TracedTask):
+            coro_repr = _format_coroutine(task._orig_coro).partition(" ")[0]
+            if persistent and task._orig_coro not in persistent_coro:
+                continue
+        else:
+            coro_repr = _format_coroutine(task.get_coro()).partition(" ")[0]
+            if persistent:
+                # untracked tasks should be skipped when showing persistent ones only
+                continue
+        if filter_ and (filter_ not in coro_repr and filter_ not in task.get_name()):
+            continue
+        creation_stack = self._created_tracebacks.get(task)
+        # Some values are masked as "-" when they are unavailable
+        # if it's the root task/coro or if the task factory is not applied.
+        if not creation_stack:
+            created_location = "-"
+        else:
+            creation_stack = _filter_stack(creation_stack)
+            fn = _format_filename(creation_stack[-1].filename)
+            lineno = creation_stack[-1].lineno
+            created_location = f"{fn}:{lineno}"
+        if isinstance(task, TracedTask):
+            running_since = _format_timedelta(
+                timedelta(
+                    seconds=(time.perf_counter() - task._started_at),
+                )
+            )
+        else:
+            running_since = "-"
+        table_data.append(
+            (
+                taskid,
+                task._state,
+                task.get_name(),
+                coro_repr,
+                created_location,
+                running_since,
+            )
+        )
+    table = AsciiTable(table_data)
+    table.inner_row_border = False
+    table.inner_column_border = False
+    if filter_ or persistent:
+        stdout.write(
+            f"{len(all_running_tasks)} tasks running "
+            f"(showing {len(table_data) - 1} tasks)\n"
+        )
+    else:
+        stdout.write(f"{len(all_running_tasks)} tasks running\n")
+    stdout.write(table.table)
+    stdout.write("\n")
+    stdout.flush()
+
+
+@monitor_cli.command(name="ps-terminated", aliases=["pt", "pst"])
+@click.option("-f", "--filter", "filter_", help="filter by coroutine or task name")
+@click.option("-p", "--persistent", is_flag=True, help="show only persistent tasks")
+@custom_help_option
+@auto_command_done
+def do_ps_terminated(
+    ctx: click.Context,
+    filter_: str,
+    persistent: bool,
+) -> None:
+    """List recently terminated/cancelled tasks"""
+    headers = (
+        "Trace ID",
+        "Name",
+        "Coro",
+        "Since Started",
+        "Since Terminated",
+    )
+    self: Monitor = ctx.obj
+    stdout = _get_current_stdout()
+    table_data: List[Tuple[str, str, str, str, str]] = [headers]
+    terminated_tasks = self._terminated_tasks.values()
+    for item in sorted(
+        terminated_tasks,
+        key=lambda info: info.terminated_at,
+        reverse=True,
+    ):
+        if persistent and not item.persistent:
+            continue
+        if filter_ and (filter_ not in item.coro and filter_ not in item.name):
+            continue
+        run_since = _format_timedelta(
+            timedelta(seconds=time.perf_counter() - item.started_at)
+        )
+        cancelled_since = _format_timedelta(
+            timedelta(seconds=time.perf_counter() - item.terminated_at)
+        )
+        table_data.append(
+            (
+                str(item.id),
+                item.name,
+                item.coro,
+                run_since,
+                cancelled_since,
+            )
+        )
+    table = AsciiTable(table_data)
+    table.inner_row_border = False
+    table.inner_column_border = False
+    if filter_ or persistent:
+        stdout.write(
+            f"{len(terminated_tasks)} tasks terminated "
+            f"(showing {len(table_data) - 1} tasks)\n"
+        )
+    else:
+        stdout.write(
+            f"{len(terminated_tasks)} tasks terminated (old ones may be stripped)\n"
+        )
+    stdout.write(table.table)
+    stdout.write("\n")
+    stdout.flush()
+
+
+@monitor_cli.command(name="where", aliases=["w"])
+@click.argument("taskid", shell_complete=complete_task_id)
+@custom_help_option
+@auto_command_done
+def do_where(ctx: click.Context, taskid: str) -> None:
+    """Show stack frames and the task creation chain of a task"""
+    self: Monitor = ctx.obj
+    stdout = _get_current_stdout()
+    depth = 0
+    task_id = int(taskid)
+    task = task_by_id(task_id, self._monitored_loop)
+    if task is None:
+        self.print_fail(f"No task {task_id}")
+        return
+    task_chain: List[asyncio.Task[Any]] = []
+    while task is not None:
+        task_chain.append(task)
+        task_ref = self._created_traceback_chains.get(task)
+        task = task_ref() if task_ref is not None else None
+    prev_task = None
+    for task in reversed(task_chain):
+        if depth == 0:
+            stdout.write(
+                "Stack of the root task or coroutine scheduled in the event loop (most recent call last):\n\n"
+            )
+        elif depth > 0:
+            assert prev_task is not None
+            stdout.write(
+                "Stack of %s when creating the next task (most recent call last):\n\n"
+                % _format_task(prev_task)
+            )
+        stack = self._created_tracebacks.get(task)
+        if stack is None:
+            stdout.write(
+                "  No stack available (maybe it is a native code, a synchronous callback function, or the event loop itself)\n"
+            )
+        else:
+            stack = _filter_stack(stack)
+            stdout.write("".join(traceback.format_list(stack)))
+        prev_task = task
+        depth += 1
+        stdout.write("\n")
+    task = task_chain[0]
+    stdout.write("Stack of %s (most recent call last):\n\n" % _format_task(task))
+    stack = _extract_stack_from_task(task)
+    if not stack:
+        stdout.write("  No stack available for %s" % _format_task(task))
+    else:
+        stdout.write("".join(traceback.format_list(stack)))
+    stdout.write("\n")
+
+
+@monitor_cli.command(name="where-terminated", aliases=["wt"])
+@click.argument("trace_id", shell_complete=complete_trace_id)
+@custom_help_option
+@auto_command_done
+def do_where_terminated(ctx: click.Context, trace_id: str) -> None:
+    """Show stack frames and the termination/cancellation chain of a task"""
+    self: Monitor = ctx.obj
+    stdout = _get_current_stdout()
+    depth = 0
+    tinfo_chain: List[TerminatedTaskInfo] = []
+    while trace_id is not None:
+        tinfo_chain.append(self._terminated_tasks[trace_id])
+        trace_id = self._canceller_chain.get(trace_id)  # type: ignore
+    prev_tinfo = None
+    for tinfo in reversed(tinfo_chain):
+        if depth == 0:
+            stdout.write(
+                "Stack of the root task or coroutine scheduled in the event loop (most recent call last):\n\n"
+            )
+        elif depth > 0:
+            assert prev_tinfo is not None
+            stdout.write(
+                "Stack of %s when creating the next task (most recent call last):\n\n"
+                % _format_terminated_task(prev_tinfo)
+            )
+        stack = tinfo.canceller_stack
+        if stack is None:
+            stdout.write(
+                "  No stack available (maybe it is a self-raised cancellation or exception)\n"
+            )
+        else:
+            stack = _filter_stack(stack)
+            stdout.write("".join(traceback.format_list(stack)))
+        prev_tinfo = tinfo
+        depth += 1
+        stdout.write("\n")
+    tinfo = tinfo_chain[0]
+    stdout.write(
+        "Stack of %s (most recent call last):\n\n" % _format_terminated_task(tinfo)
+    )
+    stack = tinfo.termination_stack
+    if not stack:
+        stdout.write(
+            "  No stack available for %s (the task has run to completion)"
+            % _format_terminated_task(tinfo)
+        )
+    else:
+        stdout.write("".join(traceback.format_list(stack)))
+    stdout.write("\n")
+
+
+def start_monitor(
+    loop: asyncio.AbstractEventLoop,
+    *,
+    monitor_cls: Type[Monitor] = Monitor,
+    host: str = MONITOR_HOST,
+    port: int = MONITOR_PORT,
+    console_port: int = CONSOLE_PORT,
+    console_enabled: bool = True,
+    hook_task_factory: bool = False,
+    max_termination_history: Optional[int] = None,
+    locals: Optional[Dict[str, Any]] = None,
+) -> Monitor:
+    m = monitor_cls(
+        loop,
+        host=host,
+        port=port,
+        console_port=console_port,
+        console_enabled=console_enabled,
+        hook_task_factory=hook_task_factory,
+        max_termination_history=max_termination_history
+        or get_default_args(monitor_cls.__init__)["max_termination_history"],
+        locals=locals,
+    )
+    m.start()
     return m
diff --git a/aiomonitor/mypy_types.py b/aiomonitor/mypy_types.py
deleted file mode 100644
index df6c3047..00000000
--- a/aiomonitor/mypy_types.py
+++ /dev/null
@@ -1,6 +0,0 @@
-import asyncio
-from typing import Dict, Any, Optional
-
-
-Loop = asyncio.AbstractEventLoop
-OptLocals = Optional[Dict[str, Any]]
diff --git a/aiomonitor/task.py b/aiomonitor/task.py
new file mode 100644
index 00000000..96887806
--- /dev/null
+++ b/aiomonitor/task.py
@@ -0,0 +1,113 @@
+import asyncio
+import base64
+import functools
+import struct
+import sys
+import time
+import traceback
+import weakref
+from asyncio.coroutines import _format_coroutine  # type: ignore
+from typing import Any, Callable, Coroutine, List, Optional, TypeVar
+
+import janus
+from typing_extensions import ParamSpec
+
+from .types import CancellationChain, TerminatedTaskInfo
+from .utils import _extract_stack_from_frame
+
+__all__ = (
+    "TracedTask",
+    "preserve_termination_log",
+    "persistent_coro",
+)
+
+persistent_coro: weakref.WeakSet[Coroutine] = weakref.WeakSet()
+
+T = TypeVar("T")
+P = ParamSpec("P")
+
+
+class TracedTask(asyncio.Task):
+    _orig_coro: Coroutine[Any, Any, Any]
+    _termination_stack: Optional[List[traceback.FrameSummary]]
+
+    def __init__(
+        self,
+        *args,
+        termination_info_queue: janus._SyncQueueProxy[TerminatedTaskInfo],
+        cancellation_chain_queue: janus._SyncQueueProxy[CancellationChain],
+        persistent: bool = False,
+        **kwargs,
+    ) -> None:
+        if sys.version_info < (3, 11):
+            kwargs.pop("context")
+        super().__init__(*args, **kwargs)
+        self._termination_info_queue = termination_info_queue
+        self._cancellation_chain_queue = cancellation_chain_queue
+        self._started_at = time.perf_counter()
+        self._termination_stack = None
+        self.add_done_callback(self._trace_termination)
+        self._persistent = persistent
+
+    def get_trace_id(self) -> str:
+        h = hash(
+            (
+                id(self),
+                self.get_name(),
+            )
+        )
+        b = struct.pack("P", h)
+        return base64.b32encode(b).rstrip(b"=").decode()
+
+    def _trace_termination(self, _: asyncio.Task[Any]) -> None:
+        self_id = self.get_trace_id()
+        exc_repr = (
+            repr(self.exception())
+            if not self.cancelled() and self.exception()
+            else None
+        )
+        task_info = TerminatedTaskInfo(
+            self_id,
+            name=self.get_name(),
+            coro=_format_coroutine(self._orig_coro).partition(" ")[0],
+            cancelled=self.cancelled(),
+            exc_repr=exc_repr,
+            started_at=self._started_at,
+            terminated_at=time.perf_counter(),
+            termination_stack=self._termination_stack,
+            canceller_stack=None,
+            persistent=self._persistent,
+        )
+        self._termination_info_queue.put_nowait(task_info)
+
+    def cancel(self, msg: Optional[str] = None) -> bool:
+        try:
+            canceller_task = asyncio.current_task()
+        except RuntimeError:
+            canceller_task = None
+        if canceller_task is not None and isinstance(canceller_task, TracedTask):
+            canceller_stack = _extract_stack_from_frame(sys._getframe())[:-1]
+            cancellation_chain = CancellationChain(
+                self.get_trace_id(),
+                canceller_task.get_trace_id(),
+                canceller_stack,
+            )
+            self._cancellation_chain_queue.put_nowait(cancellation_chain)
+        return super().cancel(msg)
+
+
+def preserve_termination_log(
+    corofunc: Callable[P, Coroutine[Any, None, T]]
+) -> Callable[P, Coroutine[Any, None, T]]:
+    """
+    Guard the given coroutine function from being stripped out due to the max history
+    limit when created as TracedTask.
+    """
+
+    @functools.wraps(corofunc)
+    def inner(*args: P.args, **kwargs: P.kwargs) -> Coroutine[Any, None, T]:
+        coro = corofunc(*args, **kwargs)
+        persistent_coro.add(coro)
+        return coro
+
+    return inner
diff --git a/aiomonitor/telnet.py b/aiomonitor/telnet.py
new file mode 100644
index 00000000..d375d1d2
--- /dev/null
+++ b/aiomonitor/telnet.py
@@ -0,0 +1,228 @@
+from __future__ import annotations
+
+import asyncio
+import collections
+import fcntl
+import os
+import struct
+import sys
+import telnetlib
+import termios
+from typing import Dict, Optional, TextIO, Tuple
+
+ModeDef = collections.namedtuple(
+    "ModeDef", ["iflag", "oflag", "cflag", "lflag", "ispeed", "ospeed", "cc"]
+)
+
+
+class TelnetClient:
+    """
+    A minimal telnet client-side protocol implementation to support `prompt_toolkit`.
+    Some details like terminal mode handling is taken from telnetlib3.
+    (https://github.com/jquast/telnetlib3)
+    """
+
+    def __init__(
+        self,
+        host: str,
+        port: int,
+        stdin: Optional[TextIO] = None,
+        stdout: Optional[TextIO] = None,
+    ) -> None:
+        self._host = host
+        self._port = port
+        self._term = os.environ.get("TERM", "unknown")
+        self._stdin = stdin or sys.stdin
+        self._stdout = stdout or sys.stdout
+        self._isatty = os.path.sameopenfile(self._stdin.fileno(), self._stdout.fileno())
+        self._remote_options: Dict[bytes, bool] = collections.defaultdict(lambda: False)
+
+    def get_mode(self) -> Optional[ModeDef]:
+        if self._isatty:
+            return ModeDef(*termios.tcgetattr(self._stdin.fileno()))
+        return None
+
+    def set_mode(self, mode: ModeDef) -> None:
+        termios.tcsetattr(sys.stdin.fileno(), termios.TCSAFLUSH, list(mode))
+
+    def restore_mode(self) -> None:
+        if self._isatty and self._saved_mode is not None:
+            termios.tcsetattr(
+                self._stdin.fileno(), termios.TCSAFLUSH, list(self._saved_mode)
+            )
+
+    def determine_mode(self, mode: ModeDef) -> ModeDef:
+        # Reference: https://github.com/jquast/telnetlib3/blob/b90616f/telnetlib3/client_shell.py#L76
+        if not self._remote_options[telnetlib.ECHO]:
+            return mode
+        iflag = mode.iflag & ~(
+            termios.BRKINT
+            | termios.ICRNL  # Do not send INTR signal on break
+            | termios.INPCK  # Do not map CR to NL on input
+            | termios.ISTRIP  # Disable input parity checking
+            | termios.IXON  # Do not strip input characters to 7 bits
+        )
+        cflag = mode.cflag & ~(termios.CSIZE | termios.PARENB)
+        cflag = cflag | termios.CS8
+        lflag = mode.lflag & ~(
+            termios.ICANON | termios.IEXTEN | termios.ISIG | termios.ECHO
+        )
+        oflag = mode.oflag & ~(termios.OPOST | termios.ONLCR)
+        cc = list(mode.cc)
+        cc[termios.VMIN] = 1
+        cc[termios.VTIME] = 0
+        return ModeDef(
+            iflag=iflag,
+            oflag=oflag,
+            cflag=cflag,
+            lflag=lflag,
+            ispeed=mode.ispeed,
+            ospeed=mode.ospeed,
+            cc=cc,
+        )
+
+    async def _create_stdio_streams(
+        self,
+    ) -> Tuple[asyncio.StreamReader, asyncio.StreamWriter]:
+        loop = asyncio.get_running_loop()
+        stdin_reader = asyncio.StreamReader()
+        stdin_protocol = asyncio.StreamReaderProtocol(stdin_reader)
+        await loop.connect_read_pipe(lambda: stdin_protocol, self._stdin)
+        write_target = self._stdin if self._isatty else self._stdout
+        writer_transport, writer_protocol = await loop.connect_write_pipe(
+            asyncio.streams.FlowControlMixin,
+            write_target,
+        )
+        stdout_writer = asyncio.StreamWriter(
+            writer_transport, writer_protocol, stdin_reader, loop
+        )
+        return stdin_reader, stdout_writer
+
+    async def __aenter__(self) -> TelnetClient:
+        self._conn_reader, self._conn_writer = await asyncio.open_connection(
+            self._host, self._port
+        )
+        self._closed = asyncio.Event()
+        self._stdin_reader, self._stdout_writer = await self._create_stdio_streams()
+        self._recv_task = asyncio.create_task(self._handle_received())
+        self._input_task = asyncio.create_task(self._handle_user_input())
+        await asyncio.sleep(0.3)  # wait for negotiation to complete
+        self._saved_mode = self.get_mode()
+        if self._isatty:
+            assert self._saved_mode is not None
+            self.set_mode(self.determine_mode(self._saved_mode))
+        return self
+
+    async def __aexit__(self, *exc_info) -> Optional[bool]:
+        try:
+            self._input_task.cancel()
+            await self._input_task
+            self._conn_writer.close()
+            try:
+                await self._conn_writer.wait_closed()
+            except NotImplementedError:
+                pass
+            self._conn_reader.feed_eof()
+            await self._recv_task
+            self._stdout_writer.close()
+            try:
+                await self._stdout_writer.wait_closed()
+            except NotImplementedError:
+                pass
+        finally:
+            self.restore_mode()
+        return None
+
+    async def interact(self) -> None:
+        try:
+            await self._closed.wait()
+        finally:
+            self._closed.set()
+            self._conn_writer.write_eof()
+
+    async def _handle_user_input(self) -> None:
+        try:
+            while True:
+                buf = await self._stdin_reader.read(128)
+                if not buf:
+                    return
+                self._conn_writer.write(buf)
+                await self._conn_writer.drain()
+        except asyncio.CancelledError:
+            pass
+
+    async def _handle_nego(self, command: bytes, option: bytes) -> None:
+        if command == telnetlib.DO and option == telnetlib.TTYPE:
+            self._conn_writer.write(telnetlib.IAC + telnetlib.WILL + telnetlib.TTYPE)
+            self._conn_writer.write(telnetlib.IAC + telnetlib.SB + telnetlib.TTYPE)
+            self._conn_writer.write(telnetlib.BINARY + self._term.encode("ascii"))
+            self._conn_writer.write(telnetlib.IAC + telnetlib.SE)
+            await self._conn_writer.drain()
+        elif command == telnetlib.DO and option == telnetlib.NAWS:
+            self._conn_writer.write(telnetlib.IAC + telnetlib.WILL + telnetlib.NAWS)
+            self._conn_writer.write(telnetlib.IAC + telnetlib.SB + telnetlib.NAWS)
+            fmt = "HHHH"
+            buf = b"\x00" * struct.calcsize(fmt)
+            buf = fcntl.ioctl(self._stdin.fileno(), termios.TIOCGWINSZ, buf)
+            rows, cols, _, _ = struct.unpack(fmt, buf)
+            self._conn_writer.write(struct.pack(">HH", cols, rows))
+            self._conn_writer.write(telnetlib.IAC + telnetlib.SE)
+            await self._conn_writer.drain()
+        elif command == telnetlib.WILL:
+            self._remote_options[option] = True
+            if option in (telnetlib.ECHO, telnetlib.BINARY, telnetlib.SGA):
+                self._conn_writer.write(telnetlib.IAC + telnetlib.DO + option)
+                await self._conn_writer.drain()
+        elif command == telnetlib.WONT:
+            self._remote_options[option] = False
+
+    async def _handle_sb(self, option: bytes, chunk: bytes) -> None:
+        pass
+
+    async def _handle_received(self):
+        buf = b""
+        try:
+            while not self._conn_reader.at_eof():
+                buf += await self._conn_reader.read(128)
+                while buf:
+                    cmd_begin = buf.find(telnetlib.IAC)
+                    if cmd_begin == -1:
+                        self._stdout_writer.write(buf)
+                        await self._stdout_writer.drain()
+                        buf = b""
+                    else:
+                        if cmd_begin >= len(buf) - 2:
+                            buf += await self._conn_reader.readexactly(
+                                3 - (len(buf) - cmd_begin),
+                            )
+                        command = buf[cmd_begin + 1 : cmd_begin + 2]
+                        option = buf[cmd_begin + 2 : cmd_begin + 3]
+                        buf_before, buf_after = buf[:cmd_begin], buf[cmd_begin + 3 :]
+                        self._stdout_writer.write(buf_before)
+                        await self._stdout_writer.drain()
+                        if command in (
+                            telnetlib.WILL,
+                            telnetlib.WONT,
+                            telnetlib.DO,
+                            telnetlib.DONT,
+                        ):
+                            await self._handle_nego(command, option)
+                        elif command == telnetlib.SB:
+                            subnego_end = buf_after.find(telnetlib.IAC + telnetlib.SE)
+                            if subnego_end == -1:
+                                subnego_chunk = (
+                                    buf_after
+                                    + (
+                                        await self._conn_reader.readuntil(
+                                            telnetlib.IAC + telnetlib.SE
+                                        )
+                                    )[:-2]
+                                )
+                                buf_after = b""
+                            else:
+                                subnego_chunk = buf_after[:subnego_end]
+                                buf_after = buf_after[subnego_end + 2 :]
+                            await self._handle_sb(option, subnego_chunk)
+                        buf = buf_after
+        finally:
+            self._closed.set()
diff --git a/aiomonitor/types.py b/aiomonitor/types.py
new file mode 100644
index 00000000..d73f15cd
--- /dev/null
+++ b/aiomonitor/types.py
@@ -0,0 +1,25 @@
+import traceback
+from typing import List, Optional
+
+from attrs import define
+
+
+@define
+class TerminatedTaskInfo:
+    id: str
+    name: str
+    coro: str
+    started_at: float
+    terminated_at: float
+    cancelled: bool
+    termination_stack: Optional[List[traceback.FrameSummary]]
+    canceller_stack: Optional[List[traceback.FrameSummary]] = None
+    exc_repr: Optional[str] = None
+    persistent: bool = False
+
+
+@define
+class CancellationChain:
+    target_id: str
+    canceller_id: str
+    canceller_stack: Optional[List[traceback.FrameSummary]] = None
diff --git a/aiomonitor/utils.py b/aiomonitor/utils.py
index 868a0535..31445f95 100644
--- a/aiomonitor/utils.py
+++ b/aiomonitor/utils.py
@@ -1,38 +1,131 @@
+from __future__ import annotations
+
 import asyncio
 import contextlib
+import inspect
 import linecache
-import selectors
 import sys
-import telnetlib
 import traceback
-from concurrent.futures import Future  # noqa
-from typing import Callable, IO, Any, Optional, List, Set  # noqa
+from asyncio.coroutines import _format_coroutine  # type: ignore
+from datetime import timedelta
+from pathlib import Path
+from types import FrameType
+from typing import Any, List, Optional, Set
 
-import aioconsole
+import click
 
-from .mypy_types import Loop, OptLocals
+from .types import TerminatedTaskInfo
 
 
-Server = asyncio.AbstractServer  # noqa
+def _format_task(task: asyncio.Task[Any]) -> str:
+    """
+    A simpler version of task's repr()
+    """
+    coro = _format_coroutine(task.get_coro()).partition(" ")[0]
+    return f"<Task name={task.get_name()} coro={coro}>"
 
 
-def _get_stack(task: 'asyncio.Task[Any]') -> List[Any]:
-    frames = []  # type: List[Any]
+def _format_terminated_task(tinfo: TerminatedTaskInfo) -> str:
+    """
+    A simpler version of task's repr()
+    """
+    status = []
+    if tinfo.cancelled:
+        status.append("cancelled")
+    if tinfo.exc_repr and not tinfo.cancelled:
+        status.append(f"exc={tinfo.exc_repr}")
+    return f"<TerminatedTask name={tinfo.name} coro={tinfo.coro} {' '.join(status)}>"
+
+
+def _format_filename(filename: str) -> str:
+    """
+    Simplifies the site-pkg directory path of the given source filename.
+    """
+    stdlib = (
+        f"{sys.prefix}/lib/python{sys.version_info.major}.{sys.version_info.minor}/"
+    )
+    site_pkg = f"{sys.prefix}/lib/python{sys.version_info.major}.{sys.version_info.minor}/site-packages/"
+    home = f"{Path.home()}/"
+    cwd = f"{Path.cwd()}/"
+    if filename.startswith(site_pkg):
+        return "<sitepkg>/" + filename[len(site_pkg) :]
+    if filename.startswith(stdlib):
+        return "<stdlib>/" + filename[len(stdlib) :]
+    if filename.startswith(cwd):
+        return "<cwd>/" + filename[len(cwd) :]
+    if filename.startswith(home):
+        return "<home>/" + filename[len(home) :]
+    return filename
+
+
+def _format_timedelta(td: timedelta) -> str:
+    seconds = int(td.total_seconds())
+    periods = [
+        ("y", 60 * 60 * 24 * 365),
+        ("m", 60 * 60 * 24 * 30),
+        ("d", 60 * 60 * 24),
+        ("h", 60 * 60),
+        (":", 60),
+        ("", 1),
+    ]
+    parts = []
+    for period_name, period_seconds in periods:
+        period_value, seconds = divmod(seconds, period_seconds)
+        if period_name in (":", ""):
+            parts.append(f"{period_value:02d}{period_name}")
+        else:
+            if period_value == 0:
+                continue
+            parts.append(f"{period_value}{period_name}")
+    parts.append(f"{td.microseconds / 1e6:.03f}"[1:])
+    return "".join(parts)
+
+
+def _filter_stack(
+    stack: List[traceback.FrameSummary],
+) -> List[traceback.FrameSummary]:
+    """
+    Filters out commonly repeated frames of the asyncio internals from the given stack.
+    """
+    # strip the task factory frame in the vanilla event loop
+    if (
+        stack[-1].filename.endswith("asyncio/base_events.py")
+        and stack[-1].name == "create_task"
+    ):
+        stack = stack[:-1]
+    # strip the loop.create_task frame
+    if (
+        stack[-1].filename.endswith("asyncio/tasks.py")
+        and stack[-1].name == "create_task"
+    ):
+        stack = stack[:-1]
+    _cut_idx = 0
+    for _cut_idx, f in reversed(list(enumerate(stack))):
+        # uvloop
+        if f.filename.endswith("asyncio/runners.py") and f.name == "run":
+            break
+        # vanilla
+        if f.filename.endswith("asyncio/events.py") and f.name == "_run":
+            break
+    return stack[_cut_idx + 1 :]
+
+
+def _extract_stack_from_task(
+    task: asyncio.Task[Any],
+) -> List[traceback.FrameSummary]:
+    """
+    Extracts the stack as a list of FrameSummary objects from an asyncio task.
+    """
+    frames: List[Any] = []
     coro = task._coro  # type: ignore
     while coro:
-        f = getattr(coro, 'cr_frame', getattr(coro, 'gi_frame', None))
-
+        f = getattr(coro, "cr_frame", getattr(coro, "gi_frame", None))
         if f is not None:
             frames.append(f)
-
-        coro = getattr(coro, 'cr_await', getattr(coro, 'gi_yieldfrom', None))
-    return frames
-
-
-def _format_stack(task: 'asyncio.Task[Any]') -> str:
+        coro = getattr(coro, "cr_await", getattr(coro, "gi_yieldfrom", None))
     extracted_list = []
-    checked = set()  # type: Set[str]
-    for f in _get_stack(task):
+    checked: Set[str] = set()
+    for f in frames:
         lineno = f.f_lineno
         co = f.f_code
         filename = co.co_filename
@@ -41,94 +134,118 @@ def _format_stack(task: 'asyncio.Task[Any]') -> str:
             checked.add(filename)
             linecache.checkcache(filename)
         line = linecache.getline(filename, lineno, f.f_globals)
-        extracted_list.append((filename, lineno, name, line))
-    if not extracted_list:
-        resp = 'No stack for %r' % task
-    else:
-        resp = 'Stack for %r (most recent call last):\n' % task
-        resp += ''.join(traceback.format_list(extracted_list))  # type: ignore
-    return resp
-
-
-def task_by_id(taskid: int, loop: Loop) -> 'Optional[asyncio.Task[Any]]':
+        extracted_list.append(traceback.FrameSummary(filename, lineno, name, line=line))
+    return extracted_list
+
+
+def _extract_stack_from_frame(frame: FrameType) -> List[traceback.FrameSummary]:
+    stack = traceback.StackSummary.extract(traceback.walk_stack(frame))
+    stack.reverse()
+    return stack
+
+
+def _extract_stack_from_exception(e: BaseException) -> List[traceback.FrameSummary]:
+    stack = traceback.StackSummary.extract(traceback.walk_tb(e.__traceback__))
+    stack.reverse()
+    return stack
+
+
+def get_default_args(func):
+    signature = inspect.signature(func)
+    return {
+        k: v.default
+        for k, v in signature.parameters.items()
+        if v.default is not inspect.Parameter.empty
+    }
+
+
+class AliasGroupMixin(click.Group):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self._commands = {}
+        self._aliases = {}
+
+    def command(self, *args, **kwargs):
+        aliases = kwargs.pop("aliases", [])
+        decorator = super().command(*args, **kwargs)
+
+        def _decorator(f):
+            cmd = decorator(click.pass_context(f))
+            if aliases:
+                self._commands[cmd.name] = aliases
+                for alias in aliases:
+                    self._aliases[alias] = cmd.name
+            return cmd
+
+        return _decorator
+
+    def group(self, *args, **kwargs):
+        aliases = kwargs.pop("aliases", [])
+        # keep the same class type
+        kwargs["cls"] = type(self)
+        decorator = super().group(*args, **kwargs)
+        if not aliases:
+            return decorator
+
+        def _decorator(f):
+            cmd = decorator(f)
+            if aliases:
+                self._commands[cmd.name] = aliases
+                for alias in aliases:
+                    self._aliases[alias] = cmd.name
+            return cmd
+
+        return _decorator
+
+    def get_command(self, ctx, cmd_name):
+        if cmd_name in self._aliases:
+            cmd_name = self._aliases[cmd_name]
+        command = super().get_command(ctx, cmd_name)
+        if command:
+            return command
+
+    def format_commands(self, ctx, formatter):
+        commands = []
+        for subcommand in self.list_commands(ctx):
+            cmd = self.get_command(ctx, subcommand)
+            # What is this, the tool lied about a command. Ignore it
+            if cmd is None:
+                continue
+            if cmd.hidden:
+                continue
+            if subcommand in self._commands:
+                aliases = ",".join(sorted(self._commands[subcommand]))
+                subcommand = "{0} ({1})".format(subcommand, aliases)
+            commands.append((subcommand, cmd))
+
+        # allow for 3 times the default spacing
+        if len(commands):
+            limit = formatter.width - 6 - max(len(cmd[0]) for cmd in commands)
+            rows = []
+            for subcommand, cmd in commands:
+                help = cmd.get_short_help_str(limit)
+                rows.append((subcommand, help))
+            if rows:
+                with formatter.section("Commands"):
+                    formatter.write_dl(rows)
+
+
+def task_by_id(
+    taskid: int, loop: asyncio.AbstractEventLoop
+) -> Optional[asyncio.Task[Any]]:
     tasks = all_tasks(loop=loop)
     return next(filter(lambda t: id(t) == taskid, tasks), None)
 
 
-async def cancel_task(task: 'asyncio.Task[Any]') -> None:
+async def cancel_task(task: asyncio.Task[Any]) -> None:
     with contextlib.suppress(asyncio.CancelledError):
         task.cancel()
         await task
 
 
-def init_console_server(host: str,
-                        port: int,
-                        locals: OptLocals,
-                        loop: Loop) -> 'Future[Server]':
-    def _factory(streams: Any = None) -> aioconsole.AsynchronousConsole:
-        return aioconsole.AsynchronousConsole(
-            locals=locals, streams=streams)
-
-    coro = aioconsole.start_interactive_server(
-        host=host, port=port, factory=_factory)
-    console_future = asyncio.run_coroutine_threadsafe(coro, loop=loop)
-    return console_future
-
-
-async def close_server(server: Server) -> None:
-    server.close()
-    await server.wait_closed()
-
-
-_TelnetSelector = getattr(
-    selectors, 'PollSelector',
-    selectors.SelectSelector)  # Type: selectors.BaseSelector
-
-
-def console_proxy(sin: IO[str], sout: IO[str], host: str, port: int) -> None:
-    tn = telnetlib.Telnet()
-    with contextlib.closing(tn):
-        tn.open(host, port, timeout=10)
-        with _TelnetSelector() as selector:
-            selector.register(tn, selectors.EVENT_READ)
-            selector.register(sin, selectors.EVENT_READ)
-
-            while True:
-                for key, _ in selector.select():
-                    if key.fileobj is tn:
-                        try:
-                            data = tn.read_eager()
-                        except EOFError:
-                            print('*Connection closed by remote host*')
-                            return
-
-                        if data:
-                            sout.write(data.decode('utf-8'))
-                            sout.flush()
-                    else:
-                        resp = sin.readline()
-                        if not resp:
-                            return
-                        tn.write(resp.encode('utf-8'))
-
-
-def alt_names(names: str) -> Callable[..., Any]:
-    """Add alternative names to you custom commands.
-
-    `names` is a single string with a space separated list of aliases for the
-    decorated command.
-    """
-    names_split = names.split()
-
-    def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
-        func.alt_names = names_split  # type: ignore
-        return func
-    return decorator
-
-
-def all_tasks(loop: Loop) -> 'Set[asyncio.Task[Any]]':
+def all_tasks(loop: asyncio.AbstractEventLoop) -> Set[asyncio.Task[Any]]:
     if sys.version_info >= (3, 7):
-        tasks = asyncio.all_tasks(loop=loop)  # type: Set[asyncio.Task[Any]]
+        tasks = asyncio.all_tasks(loop=loop)
     else:
         tasks = asyncio.Task.all_tasks(loop=loop)
     return tasks
diff --git a/docs/conf.py b/docs/conf.py
index 96fef152..5da0da57 100644
--- a/docs/conf.py
+++ b/docs/conf.py
@@ -32,46 +32,48 @@
 from importlib.metadata import version
 
 _docs_path = pathlib.Path(__file__).parent
-_version_path = _docs_path / '../aiomonitor/__init__.py'
+_version_path = _docs_path / "../aiomonitor/__init__.py"
 
 
 try:
-    _version_info = version('aiomonitor')
+    _version_info = version("aiomonitor")
 except IndexError:
-    raise RuntimeError('Unable to determine version.')
+    raise RuntimeError("Unable to determine version.")
 
 # Add any Sphinx extension module names here, as strings. They can be
 # extensions coming with Sphinx (named 'sphinx.ext.*') or your custom
 # ones.
-extensions = ['sphinx.ext.autodoc',
-    'sphinx.ext.intersphinx',
-    'sphinx.ext.todo',
-    'sphinx.ext.coverage',
-    'sphinx.ext.viewcode']
+extensions = [
+    "sphinx.ext.autodoc",
+    "sphinx.ext.intersphinx",
+    "sphinx.ext.todo",
+    "sphinx.ext.coverage",
+    "sphinx.ext.viewcode",
+]
 
 # Add any paths that contain templates here, relative to this directory.
-templates_path = ['_templates']
+templates_path = ["_templates"]
 
 # The suffix(es) of source filenames.
 # You can specify multiple suffix as a list of string:
 #
 # source_suffix = ['.rst', '.md']
-source_suffix = '.rst'
+source_suffix = ".rst"
 
 # The master toctree document.
-master_doc = 'index'
+master_doc = "index"
 
 # General information about the project.
-project = 'aiomonitor'
-copyright = '2016, Nikolay Novik'
-author = 'Nikolay Novik'
+project = "aiomonitor"
+copyright = "2016, Nikolay Novik"
+author = "Nikolay Novik"
 
 # The version info for the project you're documenting, acts as replacement for
 # |version| and |release|, also used in various other places throughout the
 # built documents.
 #
 # The short X.Y version.
-version = '.'.join(_version_info.split('.')[:2])
+version = ".".join(_version_info.split(".")[:2])
 # The full version, including alpha/beta/rc tags.
 release = _version_info
 # The language for content autogenerated by Sphinx. Refer to documentation
@@ -84,10 +86,10 @@
 # List of patterns, relative to source directory, that match files and
 # directories to ignore when looking for source files.
 # This patterns also effect to html_static_path and html_extra_path
-exclude_patterns = ['_build', 'Thumbs.db', '.DS_Store']
+exclude_patterns = ["_build", "Thumbs.db", ".DS_Store"]
 
 # The name of the Pygments (syntax highlighting) style to use.
-pygments_style = 'sphinx'
+pygments_style = "sphinx"
 
 # If true, `todo` and `todoList` produce output, else they produce nothing.
 todo_include_todos = True
@@ -98,32 +100,33 @@
 # The theme to use for HTML and HTML Help pages.  See the documentation for
 # a list of builtin themes.
 #
-html_theme = 'alabaster'
+html_theme = "alabaster"
 
 # Theme options are theme-specific and customize the look and feel of a theme
 # further.  For a list of options available for each theme, see the
 # documentation.
 #
-aiomonitor_desc = ('module that adds monitor and cli capabilities'
-                   'for asyncio application')
+aiomonitor_desc = (
+    "module that adds monitor and cli capabilities" "for asyncio application"
+)
 html_theme_options = {
-    'description': aiomonitor_desc,
-    'github_user': 'aio-libs',
-    'github_repo': 'aiomonitor',
-    'github_button': True,
-    'github_type': 'star',
-    'github_banner': True,
+    "description": aiomonitor_desc,
+    "github_user": "aio-libs",
+    "github_repo": "aiomonitor",
+    "github_button": True,
+    "github_type": "star",
+    "github_banner": True,
 }
 # Add any paths that contain custom static files (such as style sheets) here,
 # relative to this directory. They are copied after the builtin static files,
 # so a file named "default.css" will overwrite the builtin "default.css".
-html_static_path = ['_static']
+html_static_path = ["_static"]
 
 
 # -- Options for HTMLHelp output ------------------------------------------
 
 # Output file base name for HTML help builder.
-htmlhelp_basename = 'aiomonitordoc'
+htmlhelp_basename = "aiomonitordoc"
 
 
 # -- Options for LaTeX output ---------------------------------------------
@@ -132,15 +135,12 @@
     # The paper size ('letterpaper' or 'a4paper').
     #
     # 'papersize': 'letterpaper',
-
     # The font size ('10pt', '11pt' or '12pt').
     #
     # 'pointsize': '10pt',
-
     # Additional stuff for the LaTeX preamble.
     #
     # 'preamble': '',
-
     # Latex figure (float) alignment
     #
     # 'figure_align': 'htbp',
@@ -150,8 +150,13 @@
 # (source start file, target name, title,
 #  author, documentclass [howto, manual, or own class]).
 latex_documents = [
-    (master_doc, 'aiomonitor.tex', 'aiomonitor Documentation',
-     'Nikolay Novik', 'manual'),
+    (
+        master_doc,
+        "aiomonitor.tex",
+        "aiomonitor Documentation",
+        "Nikolay Novik",
+        "manual",
+    ),
 ]
 
 
@@ -159,10 +164,7 @@
 
 # One entry per manual page. List of tuples
 # (source start file, name, description, authors, manual section).
-man_pages = [
-    (master_doc, 'aiomonitor', 'aiomonitor Documentation',
-     [author], 1)
-]
+man_pages = [(master_doc, "aiomonitor", "aiomonitor Documentation", [author], 1)]
 
 
 # -- Options for Texinfo output -------------------------------------------
@@ -171,13 +173,17 @@
 # (source start file, target name, title, author,
 #  dir menu entry, description, category)
 texinfo_documents = [
-    (master_doc, 'aiomonitor', 'aiomonitor Documentation',
-     author, 'aiomonitor', aiomonitor_desc,
-     'Miscellaneous'),
+    (
+        master_doc,
+        "aiomonitor",
+        "aiomonitor Documentation",
+        author,
+        "aiomonitor",
+        aiomonitor_desc,
+        "Miscellaneous",
+    ),
 ]
 
 
-
-
 # Example configuration for intersphinx: refer to the Python standard library.
-intersphinx_mapping = {'https://docs.python.org/': None}
+intersphinx_mapping = {"https://docs.python.org/": None}
diff --git a/docs/screenshot-ps-where-example.png b/docs/screenshot-ps-where-example.png
new file mode 100644
index 00000000..01dbe50f
Binary files /dev/null and b/docs/screenshot-ps-where-example.png differ
diff --git a/examples/cancellation.py b/examples/cancellation.py
new file mode 100644
index 00000000..0598c87a
--- /dev/null
+++ b/examples/cancellation.py
@@ -0,0 +1,109 @@
+import asyncio
+
+import aiomonitor
+from aiomonitor.task import preserve_termination_log
+
+
+@preserve_termination_log
+async def should_have_run_until_exit():
+    print("should_have_run_until_exit: begin")
+    await asyncio.sleep(5)
+    print("should_have_run_until_exit: cancel")
+    raise asyncio.CancelledError("cancelled-suspiciously")
+
+
+async def chain3():
+    await asyncio.sleep(10)
+
+
+async def chain2():
+    t = asyncio.create_task(chain3())
+    try:
+        await asyncio.sleep(10)
+    finally:
+        t.cancel()
+        await t
+
+
+async def chain1():
+    t = asyncio.create_task(chain2())
+    try:
+        await asyncio.sleep(10)
+    finally:
+        t.cancel()
+        await t
+
+
+async def chain_main():
+    try:
+        while True:
+            t = asyncio.create_task(chain1())
+            await asyncio.sleep(0.5)
+            t.cancel()
+    finally:
+        print("terminating chain_main")
+
+
+async def self_cancel_main():
+    async def do_self_cancel(tick):
+        await asyncio.sleep(tick)
+        raise asyncio.CancelledError("self-cancelled")
+
+    try:
+        while True:
+            await asyncio.sleep(0.21)
+            asyncio.create_task(do_self_cancel(0.2))
+    finally:
+        print("terminating self_cancel_main")
+
+
+async def unhandled_exc_main():
+    async def do_unhandled(tick):
+        await asyncio.sleep(tick)
+        return 1 / 0
+
+    try:
+        while True:
+            try:
+                await asyncio.create_task(do_unhandled(0.2))
+            except ZeroDivisionError:
+                continue
+    finally:
+        print("terminating unhandled_exc_main")
+
+
+async def main():
+    loop = asyncio.get_running_loop()
+    with aiomonitor.start_monitor(
+        loop,
+        hook_task_factory=True,
+        max_termination_history=10,
+    ):
+        asyncio.create_task(should_have_run_until_exit())
+        chain_main_task = asyncio.create_task(chain_main())
+        self_cancel_main_task = asyncio.create_task(self_cancel_main())
+        unhandled_exc_main_task = asyncio.create_task(unhandled_exc_main())
+        try:
+            print("running some test workloads for 10 seconds ...")
+            await asyncio.sleep(10)
+        finally:
+            print("stopping ...")
+            chain_main_task.cancel()
+            self_cancel_main_task.cancel()
+            unhandled_exc_main_task.cancel()
+            await asyncio.gather(
+                chain_main_task,
+                self_cancel_main_task,
+                unhandled_exc_main_task,
+                return_exceptions=True,
+            )
+        print("waiting inspection ...")
+        while True:
+            await asyncio.sleep(10)
+
+
+if __name__ == "__main__":
+    try:
+        asyncio.run(main())
+    except KeyboardInterrupt:
+        print("terminated")
diff --git a/examples/console-variables.py b/examples/console-variables.py
new file mode 100644
index 00000000..3adb374e
--- /dev/null
+++ b/examples/console-variables.py
@@ -0,0 +1,28 @@
+import asyncio
+
+import aiomonitor
+
+
+async def main():
+    loop = asyncio.get_running_loop()
+    data = {
+        "a": 123,
+        "b": 456,
+    }
+    with aiomonitor.start_monitor(loop=loop) as monitor:
+        monitor.console_locals["data"] = data
+        print("Now you can connect with:")
+        print(f"  python -m aiomonitor.cli -H {monitor.host} -p {monitor.port}")
+        print("or")
+        print(f"  telnet {monitor.host} {monitor.port}  # Linux only")
+        print("\nTry 'console' command and inspect 'data' variable.")
+        # run forever until interrupted
+        while True:
+            await asyncio.sleep(60)
+
+
+if __name__ == "__main__":
+    try:
+        asyncio.run(main())
+    except KeyboardInterrupt:
+        pass
diff --git a/examples/extension.py b/examples/extension.py
new file mode 100644
index 00000000..72e4670e
--- /dev/null
+++ b/examples/extension.py
@@ -0,0 +1,79 @@
+import asyncio
+
+import click
+
+import aiomonitor
+from aiomonitor.monitor import auto_async_command_done, auto_command_done
+
+
+@aiomonitor.monitor_cli.command(name="hello")
+@auto_command_done
+def do_hello(ctx: click.Context) -> None:
+    """
+    Command extension example
+    """
+    # To send outputs to the telnet client,
+    # you should use either:
+
+    click.echo("Hello, world!")
+
+    # or:
+
+    from aiomonitor.monitor import current_stdout
+
+    stdout = current_stdout.get()
+    print("Hello, world, again!", file=stdout)
+
+    # or:
+
+    from prompt_toolkit import print_formatted_text
+    from prompt_toolkit.formatted_text import FormattedText
+
+    print_formatted_text(
+        FormattedText(
+            [
+                ("ansibrightblue", "Hello, "),
+                ("ansibrightyellow", "world, "),
+                ("ansibrightmagenta", "with color!"),
+            ]
+        )
+    )
+
+    # or:
+
+    monitor: aiomonitor.Monitor = ctx.obj
+    monitor.print_ok("Hello, world, success!")
+    monitor.print_fail("Hello, world, failure!")
+
+
+@aiomonitor.monitor_cli.command(name="hello-async")
+def do_async_hello(ctx: click.Context) -> None:
+    """
+    Command extension example (async)
+    """
+
+    @auto_async_command_done
+    async def _do_async_hello(ctx: click.Context):
+        await asyncio.sleep(1)
+        click.echo("Hello, world, after sleep!")
+
+    asyncio.create_task(_do_async_hello(ctx))
+
+
+async def main():
+    loop = asyncio.get_running_loop()
+    with aiomonitor.start_monitor(loop=loop) as monitor:
+        print("Now you can connect with:")
+        print(f"  python -m aiomonitor.cli -H {monitor.host} -p {monitor.port}")
+        print("or")
+        print(f"  telnet {monitor.host} {monitor.port}  # Linux only")
+        # run forever until interrupted
+        while True:
+            await asyncio.sleep(60)
+
+
+if __name__ == "__main__":
+    try:
+        asyncio.run(main())
+    except KeyboardInterrupt:
+        pass
diff --git a/examples/simple_aiohttp_srv.py b/examples/simple_aiohttp_srv.py
index 1412c99d..177353c1 100644
--- a/examples/simple_aiohttp_srv.py
+++ b/examples/simple_aiohttp_srv.py
@@ -1,22 +1,41 @@
 import asyncio
+import logging
 
-import aiomonitor
+import uvloop
 from aiohttp import web
 
+import aiomonitor
+
 
-async def simple(request):
-    print('Start sleeping')
+async def inner2() -> None:
     await asyncio.sleep(100)
-    return web.Response(text='Simple answer')
 
 
-async def init(loop):
+async def inner1() -> None:
+    t = asyncio.create_task(inner2())
+    await t
+
+
+async def simple(request: web.Request) -> web.Response:
+    print("Start sleeping")
+    t = asyncio.create_task(inner1())
+    await t
+    return web.Response(text="Simple answer")
+
+
+async def main() -> None:
+    loop = asyncio.get_running_loop()
     app = web.Application()
-    app.router.add_get('/simple', simple)
-    return app
+    app.router.add_get("/simple", simple)
+    with aiomonitor.start_monitor(loop, hook_task_factory=True):
+        await web._run_app(app, port=8090, host="localhost")
 
-loop = asyncio.get_event_loop()
-app = loop.run_until_complete(init(loop))
 
-with aiomonitor.start_monitor(loop=loop):
-    web.run_app(app, port=8090, host='localhost')
+if __name__ == "__main__":
+    logging.basicConfig()
+    logging.getLogger("aiomonitor").setLevel(logging.DEBUG)
+    uvloop.install()
+    try:
+        asyncio.run(main())
+    except KeyboardInterrupt:
+        pass
diff --git a/examples/simple_loop.py b/examples/simple_loop.py
index 764eae72..079c6dfb 100644
--- a/examples/simple_loop.py
+++ b/examples/simple_loop.py
@@ -1,7 +1,20 @@
 import asyncio
+
 import aiomonitor
 
-loop = asyncio.get_event_loop()
-with aiomonitor.start_monitor():
-    print('Now you can connect with: nc localhost 50101')
-    loop.run_forever()
+
+async def main() -> None:
+    loop = asyncio.get_running_loop()
+    with aiomonitor.start_monitor(loop=loop) as monitor:
+        print("Now you can connect with:")
+        print(f"  python -m aiomonitor.cli -H {monitor.host} -p {monitor.port}")
+        print("or")
+        print(f"  telnet {monitor.host} {monitor.port}  # Linux only")
+        loop.run_forever()
+
+
+if __name__ == "__main__":
+    try:
+        asyncio.run(main())
+    except KeyboardInterrupt:
+        pass
diff --git a/examples/taskgroup.py b/examples/taskgroup.py
new file mode 100644
index 00000000..73c4f368
--- /dev/null
+++ b/examples/taskgroup.py
@@ -0,0 +1,31 @@
+import asyncio
+
+import aiotools  # FIXME: replace with asyncio.TaskGroup in Python 3.11
+
+import aiomonitor
+
+
+async def inner2():
+    await asyncio.sleep(100)
+
+
+async def inner1():
+    t = asyncio.create_task(inner2())
+    await t
+
+
+async def main():
+    loop = asyncio.get_running_loop()
+    with aiomonitor.start_monitor(loop, hook_task_factory=True):
+        print("Starting...")
+        async with aiotools.TaskGroup() as tg:
+            for _ in range(10):
+                tg.create_task(inner1())
+        print("Done")
+
+
+if __name__ == "__main__":
+    try:
+        asyncio.run(main())
+    except KeyboardInterrupt:
+        pass
diff --git a/examples/web_srv.py b/examples/web_srv.py
index 64ece2e3..723c664d 100644
--- a/examples/web_srv.py
+++ b/examples/web_srv.py
@@ -1,41 +1,44 @@
 import asyncio
 
-import aiomonitor
 import uvloop
-
 from aiohttp import web
 
+import aiomonitor
+
 
-async def simple(request):
+async def simple(request: web.Request) -> web.Response:
     await asyncio.sleep(10)
     await asyncio.sleep(10)
-    return web.Response(text='Simple answer')
+    return web.Response(text="Simple answer")
 
 
-async def hello(request):
+async def hello(request: web.Request) -> web.StreamResponse:
     resp = web.StreamResponse()
-    name = request.match_info.get('name', 'Anonymous')
-    answer = ('Hello, ' + name).encode('utf8')
+    name = request.match_info.get("name", "Anonymous")
+    answer = ("Hello, " + name).encode("utf8")
     resp.content_length = len(answer)
-    resp.content_type = 'text/plain'
+    resp.content_type = "text/plain"
     await resp.prepare(request)
     await asyncio.sleep(100)
-    resp.write(answer)
+    await resp.write(answer)
     await resp.write_eof()
     return resp
 
 
-async def init(loop):
-    app = web.Application(loop=loop)
-    app.router.add_get('/simple', simple)
-    app.router.add_get('/hello/{name}', hello)
-    app.router.add_get('/hello', hello)
-    return app
+async def main() -> None:
+    loop = asyncio.get_running_loop()
+    app = web.Application()
+    app.router.add_get("/simple", simple)
+    app.router.add_get("/hello/{name}", hello)
+    app.router.add_get("/hello", hello)
+    host, port = "localhost", 8090
+    with aiomonitor.start_monitor(loop):
+        web.run_app(app, port=port, host=host)
 
-host, port = 'localhost', 8090
-loop = uvloop.new_event_loop()
-asyncio.set_event_loop(loop)
-app = loop.run_until_complete(init(loop))
 
-with aiomonitor.start_monitor(loop, locals=locals()):
-    web.run_app(app, port=port, host=host)
+if __name__ == "__main__":
+    uvloop.install()
+    try:
+        asyncio.run(main())
+    except KeyboardInterrupt:
+        pass
diff --git a/examples/web_srv_custom_monitor.py b/examples/web_srv_custom_monitor.py
index 65f1b989..0462561d 100644
--- a/examples/web_srv_custom_monitor.py
+++ b/examples/web_srv_custom_monitor.py
@@ -1,23 +1,26 @@
 import asyncio
+from typing import Optional
 
-import aiomonitor
+import click
+import requests
 import uvloop
-
 from aiohttp import web
-import requests
+
+import aiomonitor
+from aiomonitor.monitor import auto_command_done, custom_help_option, monitor_cli
 
 
-async def simple(request):
+async def simple(request: web.Request) -> web.Response:
     await asyncio.sleep(10)
-    return web.Response(text='Simple answer')
+    return web.Response(text="Simple answer")
 
 
-async def hello(request):
+async def hello(request: web.Request) -> web.StreamResponse:
     resp = web.StreamResponse()
-    name = request.match_info.get('name', 'Anonymous')
-    answer = ('Hello, ' + name).encode('utf8')
+    name = request.match_info.get("name", "Anonymous")
+    answer = ("Hello, " + name).encode("utf8")
     resp.content_length = len(answer)
-    resp.content_type = 'text/plain'
+    resp.content_type = "text/plain"
     await resp.prepare(request)
     await asyncio.sleep(10)
     await resp.write(answer)
@@ -25,30 +28,35 @@ async def hello(request):
     return resp
 
 
-async def init(loop):
-    app = web.Application()
-    app.router.add_get('/simple', simple)
-    app.router.add_get('/hello/{name}', hello)
-    app.router.add_get('/hello', hello)
-    return app
-
-host, port = 'localhost', 8090
-loop = uvloop.new_event_loop()
-asyncio.set_event_loop(loop)
-app = loop.run_until_complete(init(loop))
-
+@monitor_cli.command(name="hello")
+@click.argument("name", optional=True)
+@custom_help_option
+@auto_command_done
+def do_hello(ctx: click.Context, name: Optional[str] = None) -> None:
+    """Using the /hello GET interface
 
-class WebMonitor(aiomonitor.Monitor):
-    def do_hello(self, sin, sout, name=None):
-        """Using the /hello GET interface
+    There is one optional argument, "name".  This name argument must be
+    provided with proper URL excape codes, like %20 for spaces.
+    """
+    name = "" if name is None else "/" + name
+    r = requests.get("http://localhost:8090/hello" + name)
+    click.echo(r.text + "\n")
 
-        There is one optional argument, "name".  This name argument must be
-        provided with proper URL excape codes, like %20 for spaces.
-        """
-        name = '' if name is None else '/' + name
-        r = requests.get('http://localhost:8090/hello' + name)
-        sout.write(r.text + '\n')
 
-
-with aiomonitor.start_monitor(loop, monitor=WebMonitor, locals=locals()):
-    web.run_app(app, port=port, host=host)
+async def main() -> None:
+    loop = asyncio.get_running_loop()
+    app = web.Application()
+    app.router.add_get("/simple", simple)
+    app.router.add_get("/hello/{name}", hello)
+    app.router.add_get("/hello", hello)
+    host, port = "localhost", 8090
+    with aiomonitor.start_monitor(loop, locals=locals()):
+        web.run_app(app, port=port, host=host)
+
+
+if __name__ == "__main__":
+    uvloop.install()
+    try:
+        asyncio.run(main())
+    except KeyboardInterrupt:
+        pass
diff --git a/examples/x.py b/examples/x.py
new file mode 100644
index 00000000..682fc6a7
--- /dev/null
+++ b/examples/x.py
@@ -0,0 +1,15 @@
+import asyncio
+
+
+async def do():
+    await asyncio.sleep(100)
+
+
+async def main():
+    t = asyncio.create_task(do())
+    await asyncio.sleep(1)
+    t.cancel()
+    await t
+
+
+asyncio.run(main())
diff --git a/pyproject.toml b/pyproject.toml
new file mode 100644
index 00000000..87193692
--- /dev/null
+++ b/pyproject.toml
@@ -0,0 +1,9 @@
+[tool.black]
+line-length = 88
+
+[tool.isort]
+profile = "black"
+line_length = 88
+
+[tool.mypy]
+ignore_missing_imports = true
diff --git a/requirements-dev.txt b/requirements-dev.txt
index 94644d78..7f697981 100644
--- a/requirements-dev.txt
+++ b/requirements-dev.txt
@@ -1,15 +1,19 @@
 -e .
 aioconsole==0.6.0
 aiohttp==3.8.4
+aiotools==1.5.9
 attrs==22.2.0
+black==22.10.0
 docutils==0.19
 flake8-bugbear==23.2.13
 flake8-quotes==3.3.2
 flake8==6.0.0
 ipdb==0.13.11
+isort==5.10.1
 mypy==1.0.1
 pytest-cov==4.0.0
 pytest-sugar==0.9.6
-pytest==7.2.1
+pytest==7.1.2
 terminaltables==3.1.10
 uvloop==0.17.0
+types-requests
diff --git a/setup.py b/setup.py
index d3ec7469..847d9eab 100644
--- a/setup.py
+++ b/setup.py
@@ -1,5 +1,7 @@
 import os
-from setuptools import setup, find_packages
+from typing import Dict
+
+from setuptools import find_packages, setup
 
 
 def read(f):
@@ -7,40 +9,47 @@ def read(f):
 
 
 install_requires = [
-    'terminaltables',
-    'aioconsole',
+    "attrs>=20",
+    "click>=8",
+    "janus>=1.0",
+    "terminaltables",
+    "typing-extensions>=4.1",
+    "prompt_toolkit>=3.0",
+    "aioconsole",
 ]
-extras_require = {}
+extras_require: Dict[str, str] = {}
 
 
 classifiers = [
-    'License :: OSI Approved :: MIT License',
-    'Intended Audience :: Developers',
-    'Programming Language :: Python :: 3',
-    'Programming Language :: Python :: 3.8',
-    'Programming Language :: Python :: 3.9',
-    'Programming Language :: Python :: 3.10',
-    'Programming Language :: Python :: 3.11',
-    'Operating System :: POSIX',
-    'Development Status :: 3 - Alpha',
-    'Framework :: AsyncIO',
+    "License :: OSI Approved :: MIT License",
+    "Intended Audience :: Developers",
+    "Programming Language :: Python :: 3",
+    "Programming Language :: Python :: 3.8",
+    "Programming Language :: Python :: 3.9",
+    "Programming Language :: Python :: 3.10",
+    "Programming Language :: Python :: 3.11",
+    "Operating System :: POSIX",
+    "Development Status :: 3 - Alpha",
+    "Framework :: AsyncIO",
 ]
 
 
 setup(
-    name='aiomonitor',
+    name="aiomonitor",
     use_scm_version=True,
-    setup_requires=['setuptools_scm'],
-    description=('aiomonitor adds monitor and python REPL '
-                 'capabilities for asyncio application'),
-    long_description='\n\n'.join((read('README.rst'), read('CHANGES.txt'))),
+    setup_requires=["setuptools_scm"],
+    description=(
+        "aiomonitor-ng adds monitor and python REPL "
+        "capabilities for asyncio application"
+    ),
+    long_description="\n\n".join((read("README.rst"), read("CHANGES.txt"))),
     classifiers=classifiers,
-    platforms=['POSIX'],
-    author='Nikolay Novik',
-    author_email='nickolainovik@gmail.com',
-    url='https://github.com/aio-libs/aiomonitor',
-    download_url='https://pypi.python.org/pypi/aiomonitor',
-    license='Apache 2',
+    platforms=["POSIX"],
+    author="Nikolay Novik",
+    author_email="nickolainovik@gmail.com",
+    url="https://github.com/aio-libs/aiomonitor",
+    download_url="https://pypi.python.org/pypi/aiomonitor",
+    license="Apache 2",
     packages=find_packages(),
     install_requires=install_requires,
     extras_require=extras_require,


Test patch:
diff --git a/tests/conftest.py b/tests/conftest.py
index 2a053773..5370cc9c 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -6,26 +6,27 @@
 import uvloop
 
 
-@pytest.fixture(scope='session')
+@pytest.fixture(scope="session")
 def unused_port():
     def f():
         with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
-            s.bind(('127.0.0.1', 0))
+            s.bind(("127.0.0.1", 0))
             return s.getsockname()[1]
+
     return f
 
 
 def pytest_generate_tests(metafunc):
-    if 'loop_type' in metafunc.fixturenames:
-        loop_type = ['asyncio', 'uvloop']
-        metafunc.parametrize('loop_type', loop_type)
+    if "loop_type" in metafunc.fixturenames:
+        loop_type = ["asyncio", "uvloop"]
+        metafunc.parametrize("loop_type", loop_type)
 
 
 @pytest.yield_fixture
 def loop(request, loop_type):
     old_loop = asyncio.get_event_loop()
     asyncio.set_event_loop(None)
-    if loop_type == 'uvloop':
+    if loop_type == "uvloop":
         loop = uvloop.new_event_loop()
     else:
         loop = asyncio.new_event_loop()
diff --git a/tests/test_monitor.py b/tests/test_monitor.py
index ca9944de..93b85213 100644
--- a/tests/test_monitor.py
+++ b/tests/test_monitor.py
@@ -1,9 +1,12 @@
 import asyncio
-import pytest
+import contextvars
+import sys
 import telnetlib
 import threading
 import time
 
+import pytest
+
 from aiomonitor import Monitor, start_monitor
 from aiomonitor.monitor import MONITOR_HOST, MONITOR_PORT
 from aiomonitor.utils import all_tasks
@@ -11,8 +14,9 @@
 
 def monitor_common(loop, monitor_cls):
     def make_baz():
-        return 'baz'
-    locals_ = {'foo': 'bar', 'make_baz': make_baz}
+        return "baz"
+
+    locals_ = {"foo": "bar", "make_baz": make_baz}
     mon = monitor_cls(loop, locals=locals_)
     ev = threading.Event()
 
@@ -32,10 +36,9 @@ def f(mon, loop, ev):
 
 @pytest.yield_fixture
 def monitor(request, loop):
-
     class MonitorSubclass(Monitor):
         def do_something(self, arg):
-            self._sout.write('doing something with ' + arg)
+            self._sout.write("doing something with " + arg)
             self._sout.flush()
 
     yield from monitor_common(loop, MonitorSubclass)
@@ -49,16 +52,16 @@ def tn_client(monitor):
             tn.open(MONITOR_HOST, MONITOR_PORT, timeout=5)
             break
         except OSError as e:
-            print('Retrying after error: {}'.format(str(e)))
+            print("Retrying after error: {}".format(str(e)))
         time.sleep(1)
     else:
-        pytest.fail('Can not connect to the telnet server')
-    tn.read_until(b'monitor >>>', 10)
+        pytest.fail("Can not connect to the telnet server")
+    tn.read_until(b"monitor >>>", 10)
     yield tn
     tn.close()
 
 
-@pytest.fixture(params=[True, False], ids=['console:True', 'console:False'])
+@pytest.fixture(params=[True, False], ids=["console:True", "console:False"])
 def console_enabled(request):
     return request.param
 
@@ -91,13 +94,14 @@ def test_ctor(loop, unused_port, console_enabled):
     async def f():
         with Monitor(loop, console_enabled=console_enabled):
             await asyncio.sleep(0.01)
+
     loop.run_until_complete(f())
 
 
-def execute(tn, command, pattern=b'>>>'):
-    tn.write(command.encode('utf-8'))
+def execute(tn, command, pattern=b">>>"):
+    tn.write(command.encode("utf-8"))
     data = tn.read_until(pattern, 100)
-    return data.decode('utf-8')
+    return data.decode("utf-8")
 
 
 def get_task_ids(loop):
@@ -106,38 +110,84 @@ def get_task_ids(loop):
 
 def test_basic_monitor(monitor, tn_client, loop):
     tn = tn_client
-    resp = execute(tn, 'help\n')
-    assert 'Commands' in resp
+    resp = execute(tn, "help\n")
+    assert "Commands" in resp
+
+    resp = execute(tn, "xxx\n")
+    assert "No such command" in resp
+
+    resp = execute(tn, "ps\n")
+    assert "Task" in resp
+
+    resp = execute(tn, "ps 123\n")
+    assert "TypeError" in resp
+
+    resp = execute(tn, "signal name\n")
+    assert "Unknown signal name" in resp
+
+    resp = execute(tn, "stacktrace\n")
+    assert "loop.run_forever()" in resp
+
+    resp = execute(tn, "w 123\n")
+    assert "No task 123" in resp
+
+    resp = execute(tn, "where 123\n")
+    assert "No task 123" in resp
+
+    resp = execute(tn, "c 123\n")
+    assert "Ambiguous command" in resp
+
+    resp = execute(tn, "cancel 123\n")
+    assert "No task 123" in resp
+
+    resp = execute(tn, "ca 123\n")
+    assert "No task 123" in resp
+
 
-    resp = execute(tn, 'xxx\n')
-    assert 'No such command' in resp
+myvar = contextvars.ContextVar("myvar", default=42)
 
-    resp = execute(tn, 'ps\n')
-    assert 'Task' in resp
 
-    resp = execute(tn, 'ps 123\n')
-    assert 'TypeError' in resp
+def test_monitor_task_factory():
+    async def do():
+        await asyncio.sleep(0)
+        myself = asyncio.current_task()
+        assert myself is not None
+        assert myself.get_name() == "mytask"
 
-    resp = execute(tn, 'signal name\n')
-    assert 'Unknown signal name' in resp
+    async def main():
+        loop = asyncio.get_running_loop()
+        with Monitor(loop, console_enabled=False, hook_task_factory=True):
+            t = asyncio.create_task(do(), name="mytask")
+            await t
 
-    resp = execute(tn, 'stacktrace\n')
-    assert 'loop.run_forever()' in resp
+    asyncio.run(main())
 
-    resp = execute(tn, 'w 123\n')
-    assert 'No task 123' in resp
 
-    resp = execute(tn, 'where 123\n')
-    assert 'No task 123' in resp
+@pytest.mark.skipif(
+    sys.version_info < (3, 11),
+    reason="The context argument of asyncio.create_task() is added in Python 3.11",
+)
+def test_monitor_task_factory_with_context():
+    ctx = contextvars.Context()
+    # This context is bound at the outermost scope,
+    # and inside it the initial value of myvar is kept intact.
 
-    resp = execute(tn, 'c 123\n')
-    assert 'Ambiguous command' in resp
+    async def do():
+        await asyncio.sleep(0)
+        assert myvar.get() == 42  # we are referring the outer context
+        myself = asyncio.current_task()
+        assert myself is not None
+        assert myself.get_name() == "mytask"
 
-    resp = execute(tn, 'cancel 123\n')
-    assert 'No task 123' in resp
+    async def main():
+        myvar.set(99)  # override in the current task's context
+        loop = asyncio.get_running_loop()
+        with Monitor(loop, console_enabled=False, hook_task_factory=True):
+            t = asyncio.create_task(do(), name="mytask", context=ctx)
+            await t
+        assert myvar.get() == 99
 
-    resp = execute(tn, 'ca 123\n')
-    assert 'No task 123' in resp
+    asyncio.run(main())
 
 
 def test_cancel_where_tasks(monitor, tn_client, loop):
@@ -153,31 +203,31 @@ async def sleeper(loop):
     task_ids = get_task_ids(loop)
     assert len(task_ids) > 0
     for t_id in task_ids:
-        resp = execute(tn, 'where {}\n'.format(t_id))
-        assert 'Task' in resp
-        resp = execute(tn, 'cancel {}\n'.format(t_id))
-        assert 'Cancel task' in resp
+        resp = execute(tn, "where {}\n".format(t_id))
+        assert "Task" in resp
+        resp = execute(tn, "cancel {}\n".format(t_id))
+        assert "Cancel task" in resp
     fut.cancel()
 
 
 def test_monitor_with_console(monitor, tn_client):
     tn = tn_client
-    resp = execute(tn, 'console\n')
-    assert 'This console is running in an asyncio event loop' in resp
-    execute(tn, 'await asyncio.sleep(0)\n')
+    resp = execute(tn, "console\n")
+    assert "This console is running in an asyncio event loop" in resp
+    execute(tn, "await asyncio.sleep(0)\n")
 
-    resp = execute(tn, 'foo\n')
+    resp = execute(tn, "foo\n")
     assert " 'bar'\n>>>" == resp
-    resp = execute(tn, 'make_baz()\n')
+    resp = execute(tn, "make_baz()\n")
     assert " 'baz'\n>>>" == resp
 
-    execute(tn, 'exit()\n')
+    execute(tn, "exit()\n")
 
-    resp = execute(tn, 'help\n')
-    assert 'Commands' in resp
+    resp = execute(tn, "help\n")
+    assert "Commands" in resp
 
 
 def test_custom_monitor_class(monitor, tn_client):
     tn = tn_client
-    resp = execute(tn, 'something someargument\n')
-    assert 'doing something with someargument' in resp
+    resp = execute(tn, "something someargument\n")
+    assert "doing something with someargument" in resp