Skip to content
Back to top

# Source code for xonsh.shells.readline_shell

    """The readline based xonsh shell.

    Portions of this code related to initializing the readline library
    are included from the IPython project.  The IPython project is:

    * Copyright (c) 2008-2014, IPython Development Team
    * Copyright (c) 2001-2007, Fernando Perez <fernando.perez@colorado.edu>
    * Copyright (c) 2001, Janko Hauser <jhauser@zscout.de>
    * Copyright (c) 2001, Nathaniel Gray <n8gray@caltech.edu>

    """

    import cmd
    import collections
    import importlib
    import os
    import select
    import shutil
    import sys
    import threading
    import typing as tp

    import xonsh.completers.tools as xct
    from xonsh.ansi_colors import (
        ansi_color_style,
        ansi_color_style_names,
        ansi_partial_color_format,
    )
    from xonsh.built_ins import XSH
    from xonsh.events import events
    from xonsh.lib.lazyasd import LazyObject, lazyobject
    from xonsh.lib.lazyimps import pyghooks, pygments, winutils
    from xonsh.lib.string import commonprefix
    from xonsh.platform import (
        ON_CYGWIN,
        ON_DARWIN,
        ON_MSYS,
        ON_POSIX,
        ON_WINDOWS,
        os_environ,
    )
    from xonsh.prompt.base import multiline_prompt
    from xonsh.shells.base_shell import BaseShell
    from xonsh.tools import (
        carriage_return,
        columnize,
        ends_with_colon_token,
        print_exception,
        to_bool,
    )

    if tp.TYPE_CHECKING:
        from types import ModuleType

    readline: "ModuleType|None" = None
    RL_COMPLETION_SUPPRESS_APPEND = RL_LIB = RL_STATE = None  # type: tp.Any
    RL_COMPLETION_QUERY_ITEMS: "tp.Any" = None
    RL_CAN_RESIZE = False
    RL_VARIABLE_VALUE: "tp.Callable[..., tp.Any]|None" = None
    _RL_STATE_DONE = 0x1000000
    _RL_STATE_ISEARCH = 0x0000080

    _RL_PREV_COMPLETION_CASE_SENSITIVE = "to-be-set"

    def _parse_dsr_cursor_column(resp: bytes) -> "int | None":
        """Return the column number from a DSR cursor-position reply.

        A DSR reply has the form ``ESC [ <row> ; <col> R``.  Returns ``None``
        when the reply is incomplete or malformed.
        """
        end = resp.find(b"R")
        if end < 0:
            return None
        semi = resp.rfind(b";", 0, end)
        if semi < 0:
            return None
        try:
            return int(resp[semi + 1 : end])
        except ValueError:
            return None

    def _ensure_newline():
        """Print a newline if the cursor is not at column 1.

        Uses the DSR (Device Status Report) escape sequence to query the
        terminal for the current cursor position.  If the cursor is past
        column 1, a previous command left a partial line (output without a
        trailing newline) and we need to move to a fresh line so that the
        prompt does not overwrite it.
        """
        import termios  # noqa: E401
        import tty

        fd = sys.stdin.fileno()
        if not os.isatty(fd):
            return
        # When running inside an SSH session, do not issue the DSR query.
        # The terminal's reply travels through stdin and is observed by the
        # local ssh client, which uses byte-stream state to detect tilde
        # escape sequences; a DSR reply arriving between the user's Enter
        # and a following `~` silently breaks ssh's `~.` etc. (issue #5686).
        if os.environ.get("SSH_TTY") or os.environ.get("SSH_CONNECTION"):
            return
        old = termios.tcgetattr(fd)
        try:
            tty.setcbreak(fd)
            # Ask the terminal: "where is the cursor?"
            sys.stdout.write("\033[6n")
            sys.stdout.flush()
            # Read the reply straight from the file descriptor.  Going through
            # sys.stdin would route bytes via a TextIOWrapper buffer: a single
            # read(1) can pull several bytes into that buffer, then select()
            # on the fd reports "not ready" and the tail of the reply leaks
            # into the next input() call as phantom user input (issue #6344).
            resp = b""
            timeout = 0.5
            while b"R" not in resp:
                ready, _, _ = select.select([fd], [], [], timeout)
                if not ready:
                    break
                try:
                    chunk = os.read(fd, 32)
                except OSError:
                    break
                if not chunk:
                    break
                resp += chunk
                timeout = 0.05  # the rest of a reply arrives back-to-back
            col = _parse_dsr_cursor_column(resp)
            if col is not None and col > 1:
                sys.stdout.write("\n")
                sys.stdout.flush()
        except (OSError, termios.error):
            pass
        finally:
            termios.tcsetattr(fd, termios.TCSADRAIN, old)

    def setup_readline():
        """Sets up the readline module and completion suppression, if available."""
        global \
            RL_COMPLETION_SUPPRESS_APPEND, \
            RL_LIB, \
            RL_CAN_RESIZE, \
            RL_STATE, \
            readline, \
            RL_COMPLETION_QUERY_ITEMS
        if RL_COMPLETION_SUPPRESS_APPEND is not None:
            return
        for _rlmod_name in ("gnureadline", "readline"):
            try:
                readline = importlib.import_module(_rlmod_name)
                sys.modules["readline"] = readline
            except ImportError:
                pass
            else:
                break

        if readline is None:
            print(
                """Skipping setup. Because no `readline` implementation available.
                Please install a backend (`readline`, `prompt-toolkit`, etc) to use
                `xonsh` interactively.
                See https://github.com/xonsh/xonsh/issues/1170"""
            )
            return

        import ctypes.util

        uses_libedit = readline.__doc__ and "libedit" in readline.__doc__
        readline.set_completer_delims(" \t\n")
        # Cygwin seems to hang indefinitely when querying the readline lib
        if (
            (not ON_CYGWIN)
            and (not ON_MSYS)
            and (readline.__spec__.has_location and (not readline.__file__.endswith(".py")))
        ):
            RL_LIB = lib = ctypes.cdll.LoadLibrary(readline.__file__)
            try:
                RL_COMPLETION_SUPPRESS_APPEND = ctypes.c_int.in_dll(
                    lib, "rl_completion_suppress_append"
                )
            except ValueError:
                # not all versions of readline have this symbol, ie Macs sometimes
                RL_COMPLETION_SUPPRESS_APPEND = None
            try:
                RL_COMPLETION_QUERY_ITEMS = ctypes.c_int.in_dll(
                    lib, "rl_completion_query_items"
                )
            except ValueError:
                # not all versions of readline have this symbol, ie Macs sometimes
                RL_COMPLETION_QUERY_ITEMS = None
            try:
                RL_STATE = ctypes.c_int.in_dll(lib, "rl_readline_state")
            except Exception:
                pass
            RL_CAN_RESIZE = hasattr(lib, "rl_reset_screen_size")
        env = XSH.env
        # reads in history
        readline.set_history_length(-1)
        ReadlineHistoryAdder()
        # sets up IPython-like history matching with up and down
        readline.parse_and_bind('"\\e[B": history-search-forward')
        readline.parse_and_bind('"\\e[A": history-search-backward')
        # Setup Shift-Tab to indent
        readline.parse_and_bind('"\\e[Z": "{}"'.format(env.get("INDENT")))

        # handle tab completion differences found in libedit readline compatibility
        # as discussed at http://stackoverflow.com/a/7116997
        if uses_libedit:
            readline.parse_and_bind("bind ^I rl_complete")
            print(
                "\n".join(
                    [
                        "",
                        "*" * 78,
                        "libedit detected - readline will not be well behaved, including but not limited to:",
                        "   * crashes on tab completion",
                        "   * incorrect history navigation",
                        "   * corrupting long-lines",
                        "   * failure to wrap or indent lines properly",
                        "",
                        "It is highly recommended that you install gnureadline, which is installable with:",
                        "     xpip install gnureadline",
                        "*" * 78,
                    ]
                ),
                file=sys.stderr,
            )
        else:
            readline.parse_and_bind("tab: complete")
        # try to load custom user settings
        inputrc_name = os_environ.get("INPUTRC")
        if inputrc_name is None:
            if uses_libedit:
                inputrc_name = ".editrc"
            else:
                inputrc_name = ".inputrc"
            inputrc_name = os.path.join(os.path.expanduser("~"), inputrc_name)
        if (not ON_WINDOWS) and (not os.path.isfile(inputrc_name)):
            inputrc_name = "/etc/inputrc"
        if ON_WINDOWS:
            winutils.enable_virtual_terminal_processing()
            if nbuf := XSH.env.get("XONSH_WIN_CONSOLE_HISTORY_BUFFERS"):
                try:
                    winutils.set_console_history_info(nbuf=nbuf)
                except OSError:
                    pass  # stdin is not a real console (redirect/ConPTY)
        if os.path.isfile(inputrc_name):
            try:
                readline.read_init_file(inputrc_name)
            except Exception:
                # this fails with libedit
                print_exception("xonsh: could not load readline default init file.")

        # Protection against paste jacking (issue #1154)
        # This must be set after the init file is loaded since read_init_file()
        # automatically disables bracketed paste
        # (https://github.com/python/cpython/pull/24108)
        readline.parse_and_bind("set enable-bracketed-paste on")

        # properly reset input typed before the first prompt
        readline.set_startup_hook(carriage_return)

    def teardown_readline():
        """Tears down up the readline module, if available."""
        try:
            import readline
        except (ImportError, TypeError):
            return

    def _rebind_case_sensitive_completions():
        # Tell readline to ignore case when presenting completions.
        # Without this, readline may reject candidates before xonsh's
        # own completer sees them.  Xonsh handles case preference via
        # sort-order tiers in Completer.complete_from_context.
        # See Github issue #1342 for background.
        global _RL_PREV_COMPLETION_CASE_SENSITIVE
        if _RL_PREV_COMPLETION_CASE_SENSITIVE is False:
            return
        readline.parse_and_bind("set completion-ignore-case on")
        _RL_PREV_COMPLETION_CASE_SENSITIVE = False

    def fix_readline_state_after_ctrl_c():
        """
        Fix to allow Ctrl-C to exit reverse-i-search.

        Based on code from:
            http://bugs.python.org/file39467/raw_input__workaround_demo.py
        """
        if ON_WINDOWS:
            # hack to make pyreadline mimic the desired behavior
            try:
                _q = readline.rl.mode.process_keyevent_queue
                if len(_q) > 1:
                    _q.pop()
            except Exception:
                pass
        if RL_STATE is None:
            return
        if RL_STATE.value & _RL_STATE_ISEARCH:
            RL_STATE.value &= ~_RL_STATE_ISEARCH
        if not RL_STATE.value & _RL_STATE_DONE:
            RL_STATE.value |= _RL_STATE_DONE

    def rl_completion_suppress_append(val=1):
        """Sets the rl_completion_suppress_append variable, if possible.
        A value of 1 (default) means to suppress, a value of 0 means to enable.
        """
        if RL_COMPLETION_SUPPRESS_APPEND is None:
            return
        RL_COMPLETION_SUPPRESS_APPEND.value = val

    def rl_completion_query_items(val=None):
        """Sets the rl_completion_query_items variable, if possible.
        A None value will set this to $COMPLETION_QUERY_LIMIT, otherwise any integer
        is accepted.
        """
        if RL_COMPLETION_QUERY_ITEMS is None:
            return
        if val is None:
            val = XSH.env.get("COMPLETION_QUERY_LIMIT")
        RL_COMPLETION_QUERY_ITEMS.value = val

    def rl_variable_dumper(readable=True):
        """Dumps the currently set readline variables. If readable is True, then this
        output may be used in an inputrc file.
        """
        RL_LIB.rl_variable_dumper(int(readable))

    def rl_variable_value(variable):
        """Returns the currently set value for a readline configuration variable."""
        global RL_VARIABLE_VALUE
        if RL_VARIABLE_VALUE is None:
            import ctypes

            RL_VARIABLE_VALUE = RL_LIB.rl_variable_value
            RL_VARIABLE_VALUE.restype = ctypes.c_char_p
        env = XSH.env
        enc, errors = env.get("XONSH_ENCODING"), env.get("XONSH_ENCODING_ERRORS")
        if isinstance(variable, str):
            variable = variable.encode(encoding=enc, errors=errors)
        rtn = RL_VARIABLE_VALUE(variable)
        return rtn.decode(encoding=enc, errors=errors)

    @lazyobject
    def rl_on_new_line():
        """Grabs one of a few possible redisplay functions in readline."""
        names = ["rl_on_new_line", "rl_forced_update_display", "rl_redisplay"]
        for name in names:
            func = getattr(RL_LIB, name, None)
            if func is not None:
                break
        else:

            def print_for_newline():
                print()

            func = print_for_newline
        return func

    def _insert_text_func(s, readline):
        """Creates a function to insert text via readline."""

        def inserter():
            readline.insert_text(s)
            readline.redisplay()

        return inserter

    def _render_completions(completions, prefix, prefix_len):
        """Render the completions according to the required prefix_len.

        Readline will replace the current prefix with the chosen rendered completion.
        """
        chopped = prefix[:-prefix_len] if prefix_len else prefix

        rendered_completions = []
        for comp in completions:
            if isinstance(comp, xct.RichCompletion) and comp.prefix_len is not None:
                if comp.prefix_len:
                    comp = prefix[: -comp.prefix_len] + comp
                else:
                    comp = prefix + comp
            elif chopped:
                comp = chopped + comp
            rendered_completions.append(comp)

        return rendered_completions

    def _rendered_keeps_prefix(rendered, prefix):
        """Return True if a rendered completion is compatible with readline's
        word-replacement (i.e. won't shrink readline's Greatest Common Prefix
        below what the user already typed).

        A rendered completion is compatible when:

        * it begins with the typed ``prefix`` (the common case), OR
        * a quoted path completer prepended a single leading ``'`` or ``"`` so
          the rendered form looks like ``'file with spaces' `` for a typed
          ``file``. The completion is still valid — readline will replace
          ``file`` with ``'file with `` (the LCP) on the first tab — but the
          simple ``startswith`` test would otherwise reject it.

        True substring matches (e.g. ``_json`` for prefix ``jso``) are still
        filtered out, which is what motivated the original check (#6209).
        """
        if rendered.startswith(prefix):
            return True
        if rendered[:1] in ("'", '"') and rendered[1:].startswith(prefix):
            return True
        return False

    DEDENT_TOKENS = LazyObject(
        lambda: frozenset(["raise", "return", "pass", "break", "continue"]),
        globals(),
        "DEDENT_TOKENS",
    )

    class ReadlineShell(BaseShell, cmd.Cmd):
        """The readline based xonsh shell."""

        def __init__(self, completekey="tab", stdin=None, stdout=None, **kwargs):
            BaseShell.__init__(self, **kwargs)
            # super() doesn't pass the stdin/stdout to Cmd's init method correctly.
            # so calling it explicitly
            cmd.Cmd.__init__(self, completekey=completekey, stdin=stdin, stdout=stdout)

            setup_readline()
            self._current_indent = ""
            self._current_prompt = ""
            self._force_hide = None
            self._complete_only_last_table = {
                # Truth table for completions, keys are:
                # (prefix_begs_quote, prefix_ends_quote, i_ends_quote,
                #  last_starts_with_prefix, i_has_space)
                (True, True, True, True, True): True,
                (True, True, True, True, False): True,
                (True, True, True, False, True): False,
                (True, True, True, False, False): True,
                (True, True, False, True, True): False,
                (True, True, False, True, False): False,
                (True, True, False, False, True): False,
                (True, True, False, False, False): False,
                (True, False, True, True, True): True,
                (True, False, True, True, False): False,
                (True, False, True, False, True): False,
                (True, False, True, False, False): True,
                (True, False, False, True, True): False,
                (True, False, False, True, False): False,
                (True, False, False, False, True): False,
                (True, False, False, False, False): False,
                (False, True, True, True, True): True,
                (False, True, True, True, False): True,
                (False, True, True, False, True): True,
                (False, True, True, False, False): True,
                (False, True, False, True, True): False,
                (False, True, False, True, False): False,
                (False, True, False, False, True): False,
                (False, True, False, False, False): False,
                (False, False, True, True, True): False,
                (False, False, True, True, False): False,
                (False, False, True, False, True): False,
                (False, False, True, False, False): True,
                (False, False, False, True, True): True,
                (False, False, False, True, False): False,
                (False, False, False, False, True): False,
                (False, False, False, False, False): False,
            }
            self.cmdqueue = collections.deque()

        def __del__(self):
            teardown_readline()

        def singleline(self, store_in_history=True, **kwargs):
            """Reads a single line of input. The store_in_history kwarg
            flags whether the input should be stored in readline's in-memory
            history.
            """
            if not store_in_history:  # store current position to remove it later
                try:
                    import readline

                    pos = readline.get_current_history_length() - 1
                except ImportError:
                    store_in_history = True
            events.on_pre_prompt_format.fire()
            if ON_POSIX:
                _ensure_newline()
            prompt = self.prompt
            events.on_pre_prompt.fire()
            rtn = input(prompt)
            events.on_post_prompt.fire()
            if not store_in_history and pos >= 0:
                readline.remove_history_item(pos)
            return rtn

        def parseline(self, line):
            """Overridden to no-op."""
            return "", line, line

        def _querycompletions(self, completions, loc):
            """Returns whether or not we should show completions. 0 means that prefixes
            should not be shown, 1 means that there is a common prefix among all completions
            and they should be shown, while 2 means that there is no common prefix but
            we are under the query limit and they should be shown.
            """
            if commonprefix([c[loc:] for c in completions]):
                return 1
            elif len(completions) <= XSH.env.get("COMPLETION_QUERY_LIMIT"):
                return 2
            msg = f"\nDisplay all {len(completions)} possibilities? "
            msg += "({GREEN}y{RESET} or {RED}n{RESET})"
            self.print_color(msg, end="", flush=True, file=sys.stderr)
            yn = "x"
            while yn not in "yn":
                yn = sys.stdin.read(1)
            show_completions = to_bool(yn)
            print()
            if not show_completions:
                rl_on_new_line()
                return 0
            w, h = shutil.get_terminal_size()
            lines = columnize(completions, width=w)
            more_msg = self.format_color(
                "{YELLOW}==={RESET} more or "
                "{PURPLE}({RESET}q{PURPLE}){RESET}uit "
                "{YELLOW}==={RESET}"
            )
            while len(lines) > h - 1:
                print("".join(lines[: h - 1]), end="", flush=True, file=sys.stderr)
                lines = lines[h - 1 :]
                print(more_msg, end="", flush=True, file=sys.stderr)
                q = sys.stdin.read(1).lower()
                print(flush=True, file=sys.stderr)
                if q == "q":
                    rl_on_new_line()
                    return 0
            print("".join(lines), end="", flush=True, file=sys.stderr)
            rl_on_new_line()
            return 0

        def completedefault(self, prefix, line, begidx, endidx):
            """Implements tab-completion for text."""
            if self.completer is None:
                return []
            rl_completion_suppress_append()  # this needs to be called each time
            _rebind_case_sensitive_completions()
            rl_completion_query_items(val=999999999)
            prev_text = "".join(self.buffer)
            completions, plen = self.completer.complete(
                prefix,
                line,
                begidx,
                endidx,
                ctx=self.ctx,
                multiline_text=prev_text + line,
                cursor_index=len(prev_text) + endidx,
            )
            rtn_completions = _render_completions(completions, prefix, plen)
            # Filter out completions that would shrink readline's Greatest
            # Common Prefix below what was typed (e.g. ``_json`` for prefix
            # ``jso``, issue #6209), while still allowing quoted path forms
            # like ``'file with spaces.txt' `` for an unquoted prefix (#2865).
            filtered = [
                (r, c)
                for r, c in zip(rtn_completions, completions, strict=True)
                if _rendered_keeps_prefix(r, prefix)
            ]
            if filtered:
                rtn_completions, completions = zip(*filtered, strict=True)
            else:
                return []

            rtn = []
            prefix_begs_quote = prefix.startswith("'") or prefix.startswith('"')
            prefix_ends_quote = prefix.endswith("'") or prefix.endswith('"')
            for i in rtn_completions:
                i_ends_quote = i.endswith("'") or i.endswith('"')
                last = i.rsplit(" ", 1)[-1]
                last_starts_prefix = last.startswith(prefix)
                i_has_space = " " in i
                key = (
                    prefix_begs_quote,
                    prefix_ends_quote,
                    i_ends_quote,
                    last_starts_prefix,
                    i_has_space,
                )
                rtn.append(last if self._complete_only_last_table[key] else i)
            # return based on show completions
            show_completions = self._querycompletions(completions, endidx - begidx)
            if show_completions == 0:
                return []
            elif show_completions == 1:
                return rtn
            elif show_completions == 2:
                return rtn
            else:
                raise ValueError("query completions flag not understood.")

        # tab complete on first index too
        completenames = completedefault  # type:ignore

        def _load_remaining_input_into_queue(self):
            buf = b""
            while True:
                r, w, x = select.select([self.stdin], [], [], 1e-6)
                if len(r) == 0:
                    break
                buf += os.read(self.stdin.fileno(), 1024)
            if len(buf) > 0:
                buf = buf.decode().replace("\r\n", "\n").replace("\r", "\n")
                self.cmdqueue.extend(buf.splitlines(keepends=True))

        def postcmd(self, stop, line):
            """Called just before execution of line. For readline, this handles the
            automatic indentation of code blocks.
            """
            try:
                import readline
            except ImportError:
                return stop
            if self.need_more_lines:
                if len(line.strip()) == 0:
                    readline.set_pre_input_hook(None)
                    self._current_indent = ""
                elif ends_with_colon_token(line):
                    ind = line[: len(line) - len(line.lstrip())]
                    ind += XSH.env.get("INDENT")
                    readline.set_pre_input_hook(_insert_text_func(ind, readline))
                    self._current_indent = ind
                elif line.split(maxsplit=1)[0] in DEDENT_TOKENS:
                    env = XSH.env
                    ind = self._current_indent[: -len(env.get("INDENT"))]
                    readline.set_pre_input_hook(_insert_text_func(ind, readline))
                    self._current_indent = ind
                else:
                    ind = line[: len(line) - len(line.lstrip())]
                    if ind != self._current_indent:
                        insert_func = _insert_text_func(ind, readline)
                        readline.set_pre_input_hook(insert_func)
                        self._current_indent = ind
            else:
                readline.set_pre_input_hook(None)
            return stop

        def _cmdloop(self, intro=None):
            """Repeatedly issue a prompt, accept input, parse an initial prefix
            off the received input, and dispatch to action methods, passing them
            the remainder of the line as argument.

            This was forked from Lib/cmd.py from the Python standard library v3.4.3,
            (C) Python Software Foundation, 2015.
            """
            self.preloop()
            try:
                import readline

                if self.use_rawinput and self.completekey:
                    self.old_completer = readline.get_completer()
                    readline.set_completer(self.complete)
                    readline.parse_and_bind(self.completekey + ": complete")
                have_readline = True
            except ImportError:
                have_readline = False
            try:
                if intro is not None:
                    self.intro = intro
                if self.intro:
                    self.stdout.write(str(self.intro) + "\n")
                stop = None
                while not stop:
                    line = None
                    exec_now = False
                    if len(self.cmdqueue) > 0:
                        line = self.cmdqueue.popleft()
                        exec_now = line.endswith("\n")
                    if self.use_rawinput and not exec_now and have_readline:
                        inserter = (
                            None if line is None else _insert_text_func(line, readline)
                        )
                        if inserter is not None:
                            readline.set_pre_input_hook(inserter)
                        try:
                            line = self.singleline()
                        except EOFError:
                            if XSH.env.get("IGNOREEOF"):
                                self.stdout.write('Use "exit" to leave the shell.\n')
                                line = ""
                            else:
                                line = "EOF"
                        if inserter is not None:
                            readline.set_pre_input_hook(None)
                    else:
                        self.print_color(self.prompt, file=self.stdout)
                        if line is not None:
                            os.write(self.stdin.fileno(), line.encode())
                        if not exec_now:
                            line = self.stdin.readline()
                        if len(line) == 0:
                            line = "EOF"
                        else:
                            line = line.rstrip("\r\n")
                        if have_readline and line != "EOF":
                            readline.add_history(line)
                    if not ON_WINDOWS:
                        # select() is not fully functional on windows
                        self._load_remaining_input_into_queue()
                    line = self.precmd(line)
                    stop = self.onecmd(line)
                    stop = self.postcmd(stop, line)
                    if ON_WINDOWS:
                        winutils.enable_virtual_terminal_processing()
                self.postloop()
            finally:
                if self.use_rawinput and self.completekey:
                    try:
                        import readline

                        readline.set_completer(self.old_completer)
                    except ImportError:
                        pass

        def cmdloop(self, intro=None):
            while XSH.exit is None:
                try:
                    self._cmdloop(intro=intro)
                except (KeyboardInterrupt, SystemExit) as e:
                    print(file=self.stdout)  # Gives a newline
                    fix_readline_state_after_ctrl_c()
                    self.reset_buffer()
                    intro = None
                    if isinstance(e, SystemExit):
                        raise

        @property
        def prompt(self):
            """Obtains the current prompt string."""
            global RL_LIB, RL_CAN_RESIZE
            if RL_CAN_RESIZE:
                # This is needed to support some system where line-wrapping doesn't
                # work. This is a bug in upstream Python, or possibly readline.
                RL_LIB.rl_reset_screen_size()
            if self.need_more_lines:
                if self.mlprompt is None:
                    try:
                        self.mlprompt = multiline_prompt(curr=self._current_prompt)
                    except Exception:  # pylint: disable=broad-except
                        print_exception()
                        self.mlprompt = "<multiline prompt error> "
                return self.mlprompt
            env = XSH.env  # pylint: disable=no-member
            p = env.get("PROMPT")
            # clear prompt level cache
            env["PROMPT_FIELDS"].reset()
            try:
                p = self.prompt_formatter(p)
            except Exception:  # pylint: disable=broad-except
                print_exception()
            hide = True if self._force_hide is None else self._force_hide
            p = ansi_partial_color_format(p, style=env.get("XONSH_COLOR_STYLE"), hide=hide)
            self._current_prompt = p
            self.settitle()
            return p

        def format_color(self, string, hide=False, force_string=False, **kwargs):
            """Readline implementation of color formatting. This uses ANSI color
            codes.
            """
            hide = hide if self._force_hide is None else self._force_hide
            style = XSH.env.get("XONSH_COLOR_STYLE")
            return ansi_partial_color_format(string, hide=hide, style=style)

        def print_color(self, string, hide=False, **kwargs):
            if isinstance(string, str):
                s = self.format_color(string, hide=hide)
            else:
                # assume this is a list of (Token, str) tuples and format it
                env = XSH.env
                style_overrides_env = env.get("XONSH_STYLE_OVERRIDES", {})
                self.styler.style_name = env.get("XONSH_COLOR_STYLE")
                self.styler.override(style_overrides_env)
                style_proxy = pyghooks.xonsh_style_proxy(self.styler)
                formatter = pyghooks.XonshTerminal256Formatter(style=style_proxy)
                s = pygments.format(string, formatter).rstrip()
            print(s, **kwargs)

        def color_style_names(self):
            """Returns an iterable of all available style names."""
            return ansi_color_style_names()

        def color_style(self):
            """Returns the current color map."""
            style = XSH.env.get("XONSH_COLOR_STYLE")
            return ansi_color_style(style=style)

        def restore_tty_sanity(self):
            """An interface for resetting the TTY stdin mode. This is highly
            dependent on the shell backend. Also it is mostly optional since
            it only affects ^Z backgrounding behaviour.
            """
            if not ON_POSIX:
                return
            stty, _ = XSH.commands_cache.lazyget("stty", (None, None))
            if stty is None:
                return
            # If available, we should just call the stty utility. This call should
            # not throw even if stty fails. It should also be noted that subprocess
            # calls, like the following, seem to be ineffective:
            #       subprocess.call([stty, 'sane'], shell=True)
            # My guess is that this is because Popen does some crazy redirecting
            # under the covers. This effectively hides the true TTY stdin handle
            # from stty. To get around this we have to use the lower level
            # os.system() function.
            os.system(stty + " sane")

    class ReadlineHistoryAdder(threading.Thread):
        def __init__(self, wait_for_gc=True, *args, **kwargs):
            """Thread responsible for adding inputs from history to the
            current readline instance. May wait for the history garbage
            collector to finish.
            """
            super().__init__(*args, **kwargs)
            self.daemon = True
            self.wait_for_gc = wait_for_gc
            self.start()

        def run(self):
            try:
                import readline
            except ImportError:
                return
            hist = XSH.history
            if hist is None:
                return
            i = 1
            for h in hist.all_items():
                line = h["inp"].rstrip()
                if i == 1:
                    pass
                elif line == readline.get_history_item(i - 1):
                    continue
                readline.add_history(line)
                if RL_LIB is not None:
                    RL_LIB.history_set_pos(i)
                i += 1