diff -Nru python-invoke-1.3.0+ds/debian/changelog python-invoke-1.4.1+ds/debian/changelog
--- python-invoke-1.3.0+ds/debian/changelog	2019-11-23 16:46:49.000000000 +0000
+++ python-invoke-1.4.1+ds/debian/changelog	2020-07-08 19:59:58.000000000 +0000
@@ -1,3 +1,11 @@
+python-invoke (1.4.1+ds-0.1) unstable; urgency=medium
+
+  * Non-maintainer upload.
+  * New upstream release.
+  * Fix uscan configuration to automatically repack orig tarball.
+
+ -- Antoine Beaupré <anarcat@debian.org>  Wed, 08 Jul 2020 15:59:58 -0400
+
 python-invoke (1.3.0+ds-0.1) unstable; urgency=low
 
   * Non-maintainer upload.
diff -Nru python-invoke-1.3.0+ds/debian/watch python-invoke-1.4.1+ds/debian/watch
--- python-invoke-1.3.0+ds/debian/watch	2019-10-11 10:32:03.000000000 +0000
+++ python-invoke-1.4.1+ds/debian/watch	2020-07-08 19:59:58.000000000 +0000
@@ -1,3 +1,3 @@
-version=3
-opts="uversionmangle=s/\.(b|rc)/~$1/,dversionmangle=s/\+dfsg\d+$//" \
+version=4
+opts="repacksuffix=+ds,uversionmangle=s/\.(b|rc)/~$1/,dversionmangle=s/\+ds\d*$//" \
 https://github.com/pyinvoke/invoke/tags .*/(\d[\d\.]+)\.tar\.gz
diff -Nru python-invoke-1.3.0+ds/dev-requirements.txt python-invoke-1.4.1+ds/dev-requirements.txt
--- python-invoke-1.3.0+ds/dev-requirements.txt	2019-10-11 10:31:04.000000000 +0000
+++ python-invoke-1.4.1+ds/dev-requirements.txt	2020-01-30 00:49:50.000000000 +0000
@@ -3,9 +3,8 @@
 releases>=0.6.1,<2.0
 alabaster==0.7.12
 # Testing (explicit dependencies to get around a Travis/pip issue)
-# NOTE: pytest-relaxed currently only works with pytest >=3, <3.3
-pytest==3.2.5
-pytest-relaxed==1.1.4
+pytest==4.6.3
+pytest-relaxed==1.1.5
 pytest-cov==2.5.1
 mock==1.0.1
 flake8==3.7.8
diff -Nru python-invoke-1.3.0+ds/integration/runners.py python-invoke-1.4.1+ds/integration/runners.py
--- python-invoke-1.3.0+ds/integration/runners.py	2019-10-11 10:31:04.000000000 +0000
+++ python-invoke-1.4.1+ds/integration/runners.py	2020-01-30 00:49:50.000000000 +0000
@@ -126,12 +126,12 @@
 
     class timeouts:
         def does_not_fire_when_command_quick(self):
-            assert Local(Context()).run("sleep 1", timeout=5)
+            assert run("sleep 1", timeout=5)
 
         def triggers_exception_when_command_slow(self):
             before = time.time()
             with raises(CommandTimedOut) as info:
-                Local(Context()).run("sleep 5", timeout=0.5)
+                run("sleep 5", timeout=0.5)
             after = time.time()
             # Fudge real time check a bit, <=0.5 typically fails due to
             # overhead etc. May need raising further to avoid races? Meh.
diff -Nru python-invoke-1.3.0+ds/integration/_support/regression.py python-invoke-1.4.1+ds/integration/_support/regression.py
--- python-invoke-1.3.0+ds/integration/_support/regression.py	1970-01-01 00:00:00.000000000 +0000
+++ python-invoke-1.4.1+ds/integration/_support/regression.py	2020-01-30 00:49:50.000000000 +0000
@@ -0,0 +1,34 @@
+"""
+Barebones regression-catching script that looks for ephemeral run() failures.
+
+Intended to be run from top level of project via ``inv regression``. In an
+ideal world this would be truly part of the integration test suite, but:
+
+- something about the outer invoke or pytest environment seems to prevent such
+  issues from appearing reliably (see eg issue #660)
+- it can take quite a while to run, even compared to other integration tests.
+"""
+
+
+import sys
+
+from invoke import task
+
+
+@task
+def check(c):
+    count = 0
+    failures = []
+    for _ in range(0, 1000):
+        count += 1
+        try:
+            # 'ls' chosen as an arbitrary, fast-enough-for-looping but
+            # does-some-real-work example (where eg 'sleep' is less useful)
+            response = c.run("ls", hide=True)
+            if not response.ok:
+                failures.append(response)
+        except Exception as e:
+            failures.append(e)
+    if failures:
+        print("run() FAILED {}/{} times!".format(len(failures), count))
+        sys.exit(1)
diff -Nru python-invoke-1.3.0+ds/invoke/config.py python-invoke-1.4.1+ds/invoke/config.py
--- python-invoke-1.3.0+ds/invoke/config.py	2019-10-11 10:31:04.000000000 +0000
+++ python-invoke-1.4.1+ds/invoke/config.py	2020-01-30 00:49:50.000000000 +0000
@@ -469,29 +469,31 @@
             # default" that could go here. Alternately, make _more_ of these
             # default to None?
             "run": {
-                "warn": False,
-                "hide": None,
-                "shell": shell,
-                "pty": False,
-                "fallback": True,
-                "env": {},
-                "replace_env": False,
+                "asynchronous": False,
+                "disown": False,
+                "dry": False,
                 "echo": False,
+                "echo_stdin": None,
                 "encoding": None,
-                "out_stream": None,
+                "env": {},
                 "err_stream": None,
+                "fallback": True,
+                "hide": None,
                 "in_stream": None,
+                "out_stream": None,
+                "pty": False,
+                "replace_env": False,
+                "shell": shell,
+                "warn": False,
                 "watchers": [],
-                "echo_stdin": None,
-                "dry": False,
             },
             # This doesn't live inside the 'run' tree; otherwise it'd make it
             # somewhat harder to extend/override in Fabric 2 which has a split
             # local/remote runner situation.
             "runners": {"local": Local},
             "sudo": {
-                "prompt": "[sudo] password: ",
                 "password": None,
+                "prompt": "[sudo] password: ",
                 "user": None,
             },
             "tasks": {
diff -Nru python-invoke-1.3.0+ds/invoke/exceptions.py python-invoke-1.4.1+ds/invoke/exceptions.py
--- python-invoke-1.3.0+ds/invoke/exceptions.py	2019-10-11 10:31:04.000000000 +0000
+++ python-invoke-1.4.1+ds/invoke/exceptions.py	2020-01-30 00:49:50.000000000 +0000
@@ -134,6 +134,10 @@
 
 
 class CommandTimedOut(Failure):
+    """
+    Raised when a subprocess did not exit within a desired timeframe.
+    """
+
     def __init__(self, result, timeout):
         super(CommandTimedOut, self).__init__(result)
         self.timeout = timeout
diff -Nru python-invoke-1.3.0+ds/invoke/__init__.py python-invoke-1.4.1+ds/invoke/__init__.py
--- python-invoke-1.3.0+ds/invoke/__init__.py	2019-10-11 10:31:04.000000000 +0000
+++ python-invoke-1.4.1+ds/invoke/__init__.py	2020-01-30 00:49:50.000000000 +0000
@@ -23,7 +23,7 @@
 from .loader import FilesystemLoader  # noqa
 from .parser import Argument, Parser, ParserContext, ParseResult  # noqa
 from .program import Program  # noqa
-from .runners import Runner, Local, Failure, Result  # noqa
+from .runners import Runner, Local, Failure, Result, Promise  # noqa
 from .tasks import task, call, Call, Task  # noqa
 from .terminals import pty_size  # noqa
 from .watchers import FailingResponder, Responder, StreamWatcher  # noqa
@@ -31,7 +31,7 @@
 
 def run(command, **kwargs):
     """
-    Run ``command`` in a local subprocess and return a `.Result` object.
+    Run ``command`` in a subprocess and return a `.Result` object.
 
     See `.Runner.run` for API details.
 
@@ -46,3 +46,23 @@
     .. versionadded:: 1.0
     """
     return Context().run(command, **kwargs)
+
+
+def sudo(command, **kwargs):
+    """
+    Run ``command`` in a ``sudo`` subprocess and return a `.Result` object.
+
+    See `.Context.sudo` for API details, such as the ``password`` kwarg.
+
+    .. note::
+        This function is a convenience wrapper around Invoke's `.Context` and
+        `.Runner` APIs.
+
+        Specifically, it creates an anonymous `.Context` instance and calls its
+        `~.Context.sudo` method, which in turn defaults to using a `.Local`
+        runner subclass for command execution (plus sudo-related bits &
+        pieces).
+
+    .. versionadded:: 1.4
+    """
+    return Context().sudo(command, **kwargs)
diff -Nru python-invoke-1.3.0+ds/invoke/program.py python-invoke-1.4.1+ds/invoke/program.py
--- python-invoke-1.3.0+ds/invoke/program.py	2019-10-11 10:31:04.000000000 +0000
+++ python-invoke-1.4.1+ds/invoke/program.py	2020-01-30 00:49:50.000000000 +0000
@@ -223,7 +223,7 @@
             If ``None`` (default), uses the first word in ``argv`` verbatim (as
             with ``name`` above, except not capitalized).
 
-        :param list binary_names:
+        :param binary_names:
             List of binary name strings, for use in completion scripts.
 
             This list ensures that the shell completion scripts generated by
diff -Nru python-invoke-1.3.0+ds/invoke/runners.py python-invoke-1.4.1+ds/invoke/runners.py
--- python-invoke-1.3.0+ds/invoke/runners.py	2019-10-11 10:31:04.000000000 +0000
+++ python-invoke-1.4.1+ds/invoke/runners.py	2020-01-30 00:49:50.000000000 +0000
@@ -99,11 +99,27 @@
         #: A list of `.StreamWatcher` instances for use by `respond`. Is filled
         #: in at runtime by `run`.
         self.watchers = []
+        # Optional timeout timer placeholder
         self._timer = None
+        # Async flags (initialized for 'finally' referencing in case something
+        # goes REAL bad during options parsing)
+        self._asynchronous = False
+        self._disowned = False
 
     def run(self, command, **kwargs):
         """
-        Execute ``command``, returning an instance of `Result`.
+        Execute ``command``, returning an instance of `Result` once complete.
+
+        By default, this method is synchronous (it only returns once the
+        subprocess has completed), and allows interactive keyboard
+        communication with the subprocess.
+
+        It can instead behave asynchronously (returning early & requiring
+        interaction with the resulting object to manage subprocess lifecycle)
+        if you specify ``asynchronous=True``. Furthermore, you can completely
+        disassociate the subprocess from Invoke's control (allowing it to
+        persist on its own after Python exits) by saying ``disown=True``. See
+        the per-kwarg docs below for details on both of these.
 
         .. note::
             All kwargs will default to the values found in this instance's
@@ -175,6 +191,62 @@
             ``pty=True``. Whether this has any effect depends on the specific
             `Runner` subclass being invoked. Default: ``True``.
 
+        :param bool asynchronous:
+            When set to ``True`` (default ``False``), enables asynchronous
+            behavior, as follows:
+
+            - Connections to the controlling terminal are disabled, meaning you
+              will not see the subprocess output and it will not respond to
+              your keyboard input - similar to ``hide=True`` and
+              ``in_stream=False`` (though explicitly given
+              ``(out|err|in)_stream`` file-like objects will still be honored
+              as normal).
+            - `.run` returns immediately after starting the subprocess, and its
+              return value becomes an instance of `Promise` instead of
+              `Result`.
+            - `Promise` objects are primarily useful for their `~Promise.join`
+              method, which blocks until the subprocess exits (similar to
+              threading APIs) and either returns a final `~Result` or raises an
+              exception, just as a synchronous ``run`` would.
+
+                - As with threading and similar APIs, users of
+                  ``asynchronous=True`` should make sure to ``join`` their
+                  `Promise` objects to prevent issues with interpreter
+                  shutdown.
+                - One easy way to handle such cleanup is to use the `Promise`
+                  as a context manager - it will automatically ``join`` at the
+                  exit of the context block.
+
+            .. versionadded:: 1.4
+
+        :param bool disown:
+            When set to ``True`` (default ``False``), returns immediately like
+            ``asynchronous=True``, but does not perform any background work
+            related to that subprocess (it is completely ignored). This allows
+            subprocesses using shell backgrounding or similar techniques (e.g.
+            trailing ``&``, ``nohup``) to persist beyond the lifetime of the
+            Python process running Invoke.
+
+            .. note::
+                If you're unsure whether you want this or ``asynchronous``, you
+                probably want ``asynchronous``!
+
+            Specifically, ``disown=True`` has the following behaviors:
+
+            - The return value is ``None`` instead of a `Result` or subclass.
+            - No I/O worker threads are spun up, so you will have no access to
+              the subprocess' stdout/stderr, your stdin will not be forwarded,
+              ``(out|err|in)_stream`` will be ignored, and features like
+              ``watchers`` will not function.
+            - No exit code is checked for, so you will not receive any errors
+              if the subprocess fails to exit cleanly.
+            - ``pty=True`` may not function correctly (subprocesses may not run
+              at all; this seems to be a potential bug in Python's
+              ``pty.fork``) unless your command line includes tools such as
+              ``nohup`` or (the shell builtin) ``disown``.
+
+            .. versionadded:: 1.4
+
         :param bool echo:
             Controls whether `.run` prints the command string to local stdout
             prior to executing it. Default: ``False``.
@@ -265,7 +337,7 @@
 
         :param timeout:
             Cause the runner to submit an interrupt to the subprocess and raise
-            `CommandTimedOut`, if the command takes longer than ``timeout``
+            `.CommandTimedOut`, if the command takes longer than ``timeout``
             seconds to execute. Defaults to ``None``, meaning no timeout.
 
             .. versionadded:: 1.3
@@ -290,143 +362,118 @@
         try:
             return self._run_body(command, **kwargs)
         finally:
-            self.stop()
-            self.stop_timer()
+            if not (self._asynchronous or self._disowned):
+                self._stop_everything()
 
-    def _run_body(self, command, **kwargs):
-        # Normalize kwargs w/ config
-        opts, out_stream, err_stream, in_stream = self._run_opts(kwargs)
-        shell = opts["shell"]
+    def _stop_everything(self):
+        # TODO 2.0: as probably noted elsewhere, stop_timer should become part
+        # of stop() and then we can nix this. Ugh!
+        self.stop()
+        self.stop_timer()
+
+    def _setup(self, command, kwargs):
+        """
+        Prepare data on ``self`` so we're ready to start running.
+        """
+        # Normalize kwargs w/ config; sets self.opts, self.streams
+        self._unify_kwargs_with_config(kwargs)
         # Environment setup
-        env = self.generate_env(opts["env"], opts["replace_env"])
-        # Echo running command
-        if opts["echo"]:
+        self.env = self.generate_env(
+            self.opts["env"], self.opts["replace_env"]
+        )
+        # Arrive at final encoding if neither config nor kwargs had one
+        self.encoding = self.opts["encoding"] or self.default_encoding()
+        # Echo running command (wants to be early to be included in dry-run)
+        if self.opts["echo"]:
             print("\033[1;37m{}\033[0m".format(command))
+        # Prepare common result args.
+        # TODO: I hate this. Needs a deeper separate think about tweaking
+        # Runner.generate_result in a way that isn't literally just this same
+        # two-step process, and which also works w/ downstream.
+        self.result_kwargs = dict(
+            command=command,
+            shell=self.opts["shell"],
+            env=self.env,
+            pty=self.using_pty,
+            hide=self.opts["hide"],
+            encoding=self.encoding,
+        )
+
+    def _run_body(self, command, **kwargs):
+        # Prepare all the bits n bobs.
+        self._setup(command, kwargs)
         # If dry-run, stop here.
-        if opts["dry"]:
+        if self.opts["dry"]:
             return self.generate_result(
-                command=command,
-                stdout="",
-                stderr="",
-                exited=0,
-                pty=self.using_pty,
+                **dict(self.result_kwargs, stdout="", stderr="", exited=0)
             )
         # Start executing the actual command (runs in background)
-        self.start(command, shell, env)
-        self.start_timer(opts["timeout"])
-        # Arrive at final encoding if neither config nor kwargs had one
-        self.encoding = opts["encoding"] or self.default_encoding()
-        # Set up IO thread parameters (format - body_func: {kwargs})
-        stdout, stderr = [], []
-        thread_args = {
-            self.handle_stdout: {
-                "buffer_": stdout,
-                "hide": "stdout" in opts["hide"],
-                "output": out_stream,
-            }
-        }
-        # After opt processing above, in_stream will be a real stream obj or
-        # False, so we can truth-test it. We don't even create a stdin-handling
-        # thread if it's False, meaning user indicated stdin is nonexistent or
-        # problematic.
-        if in_stream:
-            thread_args[self.handle_stdin] = {
-                "input_": in_stream,
-                "output": out_stream,
-                "echo": opts["echo_stdin"],
-            }
-        if not self.using_pty:
-            thread_args[self.handle_stderr] = {
-                "buffer_": stderr,
-                "hide": "stderr" in opts["hide"],
-                "output": err_stream,
-            }
-        # Kick off IO threads
-        self.threads = {}
-        exceptions = []
-        for target, kwargs in six.iteritems(thread_args):
-            t = ExceptionHandlingThread(target=target, kwargs=kwargs)
-            self.threads[target] = t
-            t.start()
-        # Wait for completion, then tie things off & obtain result
-        # And make sure we perform that tying off even if things asplode.
-        exception = None
-        while True:
-            try:
-                self.wait()
-                break  # done waiting!
-            # NOTE: we handle all this now instead of at
-            # actual-exception-handling time because otherwise the stdout/err
-            # reader threads may block until the subprocess exits.
-            # TODO: honor other signals sent to our own process and transmit
-            # them to the subprocess before handling 'normally'.
-            except KeyboardInterrupt as e:
-                self.send_interrupt(e)
-                # NOTE: no break; we want to return to self.wait() since we
-                # can't know if subprocess is actually terminating due to this
-                # or not (think REPLs-within-shells, editors, other interactive
-                # use cases)
-            except BaseException as e:  # Want to handle SystemExit etc still
-                # Store exception for post-shutdown reraise
-                exception = e
-                # Break out of return-to-wait() loop - we want to shut down
-                break
-        # Inform stdin-mirroring worker to stop its eternal looping
-        self.program_finished.set()
-        # Join threads, setting a timeout if necessary
-        for target, thread in six.iteritems(self.threads):
-            thread.join(self._thread_join_timeout(target))
-            e = thread.exception()
-            if e is not None:
-                exceptions.append(e)
-        # If we got a main-thread exception while wait()ing, raise it now that
-        # we've closed our worker threads.
-        if exception is not None:
-            raise exception
-        # Strip out WatcherError from any thread exceptions; they are bundled
-        # into Failure handling at the end.
-        watcher_errors = []
-        thread_exceptions = []
-        for exception in exceptions:
-            real = exception.value
-            if isinstance(real, WatcherError):
-                watcher_errors.append(real)
-            else:
-                thread_exceptions.append(exception)
+        self.start(command, self.opts["shell"], self.env)
+        # If disowned, we just stop here - no threads, no timer, no error
+        # checking, nada.
+        if self._disowned:
+            return
+        # Stand up & kick off IO, timer threads
+        self.start_timer(self.opts["timeout"])
+        self.threads, self.stdout, self.stderr = self.create_io_threads()
+        for thread in self.threads.values():
+            thread.start()
+        # Wrap up or promise that we will, depending
+        return self.make_promise() if self._asynchronous else self._finish()
+
+    def make_promise(self):
+        """
+        Return a `Promise` allowing async control of the rest of lifecycle.
+
+        .. versionadded:: 1.4
+        """
+        return Promise(self)
+
+    def _finish(self):
+        # Wait for subprocess to run, forwarding signals as we get them.
+        try:
+            while True:
+                try:
+                    self.wait()
+                    break  # done waiting!
+                # Don't locally stop on ^C, only forward it:
+                # - if remote end really stops, we'll naturally stop after
+                # - if remote end does not stop (eg REPL, editor) we don't want
+                # to stop prematurely
+                except KeyboardInterrupt as e:
+                    self.send_interrupt(e)
+                # TODO: honor other signals sent to our own process and
+                # transmit them to the subprocess before handling 'normally'.
+        # Make sure we tie off our worker threads, even if something exploded.
+        # Any exceptions that raised during self.wait() above will appear after
+        # this block.
+        finally:
+            # Inform stdin-mirroring worker to stop its eternal looping
+            self.program_finished.set()
+            # Join threads, storing inner exceptions, & set a timeout if
+            # necessary. (Segregate WatcherErrors as they are "anticipated
+            # errors" that want to show up at the end during creation of
+            # Failure objects.)
+            watcher_errors = []
+            thread_exceptions = []
+            for target, thread in six.iteritems(self.threads):
+                thread.join(self._thread_join_timeout(target))
+                exception = thread.exception()
+                if exception is not None:
+                    real = exception.value
+                    if isinstance(real, WatcherError):
+                        watcher_errors.append(real)
+                    else:
+                        thread_exceptions.append(exception)
         # If any exceptions appeared inside the threads, raise them now as an
         # aggregate exception object.
+        # NOTE: this is kept outside the 'finally' so that main-thread
+        # exceptions are raised before worker-thread exceptions; they're more
+        # likely to be Big Serious Problems.
         if thread_exceptions:
             raise ThreadException(thread_exceptions)
-        # At this point, we had enough success that we want to be returning or
-        # raising detailed info about our execution; so we generate a Result.
-        stdout = "".join(stdout)
-        stderr = "".join(stderr)
-        if WINDOWS:
-            # "Universal newlines" - replace all standard forms of
-            # newline with \n. This is not technically Windows related
-            # (\r as newline is an old Mac convention) but we only apply
-            # the translation for Windows as that's the only platform
-            # it is likely to matter for these days.
-            stdout = stdout.replace("\r\n", "\n").replace("\r", "\n")
-            stderr = stderr.replace("\r\n", "\n").replace("\r", "\n")
-        # Get return/exit code, unless there were WatcherErrors to handle.
-        # NOTE: In that case, returncode() may block waiting on the process
-        # (which may be waiting for user input). Since most WatcherError
-        # situations lack a useful exit code anyways, skipping this doesn't
-        # really hurt any.
-        exited = None if watcher_errors else self.returncode()
-        # Obtain actual result
-        result = self.generate_result(
-            command=command,
-            shell=shell,
-            env=env,
-            stdout=stdout,
-            stderr=stderr,
-            exited=exited,
-            pty=self.using_pty,
-            hide=opts["hide"],
-            encoding=self.encoding,
-        )
+        # Collate stdout/err, calculate exited, and get final result obj
+        result = self._collate_result(watcher_errors)
         # Any presence of WatcherError from the threads indicates a watcher was
         # upset and aborted execution; make a generic Failure out of it and
         # raise that.
@@ -435,20 +482,21 @@
             # threads...as unlikely as that would normally be.
             raise Failure(result, reason=watcher_errors[0])
         # If a timeout was requested and the subprocess did time out, shout.
-        timeout = opts["timeout"]
+        timeout = self.opts["timeout"]
         if timeout is not None and self.timed_out:
             raise CommandTimedOut(result, timeout=timeout)
-        if not (result or opts["warn"]):
+        if not (result or self.opts["warn"]):
             raise UnexpectedExit(result)
         return result
 
-    def _run_opts(self, kwargs):
+    def _unify_kwargs_with_config(self, kwargs):
         """
         Unify `run` kwargs with config options to arrive at local options.
 
-        :returns:
-            Four-tuple of ``(opts_dict, stdout_stream, stderr_stream,
-            stdin_stream)``.
+        Sets:
+
+        - ``self.opts`` - opts dict
+        - ``self.streams`` - map of stream names to stream target values
         """
         opts = {}
         for key, value in six.iteritems(self.context.config.run):
@@ -463,30 +511,71 @@
         if kwargs:
             err = "run() got an unexpected keyword argument '{}'"
             raise TypeError(err.format(list(kwargs.keys())[0]))
+        # Update disowned, async flags
+        self._asynchronous = opts["asynchronous"]
+        self._disowned = opts["disown"]
+        if self._asynchronous and self._disowned:
+            err = "Cannot give both 'asynchronous' and 'disown' at the same time!"  # noqa
+            raise ValueError(err)
         # If hide was True, turn off echoing
         if opts["hide"] is True:
             opts["echo"] = False
         # Conversely, ensure echoing is always on when dry-running
         if opts["dry"] is True:
             opts["echo"] = True
+        # Always hide if async
+        if self._asynchronous:
+            opts["hide"] = True
         # Then normalize 'hide' from one of the various valid input values,
-        # into a stream-names tuple.
-        opts["hide"] = normalize_hide(opts["hide"])
+        # into a stream-names tuple. Also account for the streams.
+        out_stream, err_stream = opts["out_stream"], opts["err_stream"]
+        opts["hide"] = normalize_hide(opts["hide"], out_stream, err_stream)
         # Derive stream objects
-        out_stream = opts["out_stream"]
         if out_stream is None:
             out_stream = sys.stdout
-        err_stream = opts["err_stream"]
         if err_stream is None:
             err_stream = sys.stderr
         in_stream = opts["in_stream"]
         if in_stream is None:
-            in_stream = sys.stdin
+            # If in_stream hasn't been overridden, and we're async, we don't
+            # want to read from sys.stdin (otherwise the default) - so set
+            # False instead.
+            in_stream = False if self._asynchronous else sys.stdin
         # Determine pty or no
         self.using_pty = self.should_use_pty(opts["pty"], opts["fallback"])
         if opts["watchers"]:
             self.watchers = opts["watchers"]
-        return opts, out_stream, err_stream, in_stream
+        # Set data
+        self.opts = opts
+        self.streams = {"out": out_stream, "err": err_stream, "in": in_stream}
+
+    def _collate_result(self, watcher_errors):
+        # At this point, we had enough success that we want to be returning or
+        # raising detailed info about our execution; so we generate a Result.
+        stdout = "".join(self.stdout)
+        stderr = "".join(self.stderr)
+        if WINDOWS:
+            # "Universal newlines" - replace all standard forms of
+            # newline with \n. This is not technically Windows related
+            # (\r as newline is an old Mac convention) but we only apply
+            # the translation for Windows as that's the only platform
+            # it is likely to matter for these days.
+            stdout = stdout.replace("\r\n", "\n").replace("\r", "\n")
+            stderr = stderr.replace("\r\n", "\n").replace("\r", "\n")
+        # Get return/exit code, unless there were WatcherErrors to handle.
+        # NOTE: In that case, returncode() may block waiting on the process
+        # (which may be waiting for user input). Since most WatcherError
+        # situations lack a useful exit code anyways, skipping this doesn't
+        # really hurt any.
+        exited = None if watcher_errors else self.returncode()
+        # TODO: as noted elsewhere, I kinda hate this. Consider changing
+        # generate_result()'s API in next major rev so we can tidy up.
+        result = self.generate_result(
+            **dict(
+                self.result_kwargs, stdout=stdout, stderr=stderr, exited=exited
+            )
+        )
+        return result
 
     def _thread_join_timeout(self, target):
         # Add a timeout to out/err thread joins when it looks like they're not
@@ -504,6 +593,45 @@
             return 1
         return None
 
+    def create_io_threads(self):
+        """
+        Create and return a dictionary of IO thread worker objects.
+
+        Caller is expected to handle persisting and/or starting the wrapped
+        threads.
+        """
+        stdout, stderr = [], []
+        # Set up IO thread parameters (format - body_func: {kwargs})
+        thread_args = {
+            self.handle_stdout: {
+                "buffer_": stdout,
+                "hide": "stdout" in self.opts["hide"],
+                "output": self.streams["out"],
+            }
+        }
+        # After opt processing above, in_stream will be a real stream obj or
+        # False, so we can truth-test it. We don't even create a stdin-handling
+        # thread if it's False, meaning user indicated stdin is nonexistent or
+        # problematic.
+        if self.streams["in"]:
+            thread_args[self.handle_stdin] = {
+                "input_": self.streams["in"],
+                "output": self.streams["out"],
+                "echo": self.opts["echo_stdin"],
+            }
+        if not self.using_pty:
+            thread_args[self.handle_stderr] = {
+                "buffer_": stderr,
+                "hide": "stderr" in self.opts["hide"],
+                "output": self.streams["err"],
+            }
+        # Kick off IO threads
+        threads = {}
+        for target, kwargs in six.iteritems(thread_args):
+            t = ExceptionHandlingThread(target=target, kwargs=kwargs)
+            threads[target] = t
+        return threads, stdout, stderr
+
     def generate_result(self, **kwargs):
         """
         Create & return a suitable `Result` instance from the given ``kwargs``.
@@ -999,7 +1127,6 @@
         # TODO 2.0: merge with stop() (i.e. make stop() something users extend
         # and call super() in, instead of completely overriding, then just move
         # this into the default implementation of stop().
-        # TODO: this
         if self._timer:
             self._timer.cancel()
 
@@ -1137,7 +1264,8 @@
                 # Use execve for bare-minimum "exec w/ variable # args + env"
                 # behavior. No need for the 'p' (use PATH to find executable)
                 # for now.
-                # TODO: see if subprocess is using equivalent of execvp...
+                # NOTE: stdlib subprocess (actually its posix flavor, which is
+                # written in C) uses either execve or execv, depending.
                 os.execve(shell, [shell, "-c", command], env)
         else:
             self.process = Popen(
@@ -1191,8 +1319,16 @@
             return self.process.returncode
 
     def stop(self):
-        # No explicit close-out required (so far).
-        pass
+        # If we opened a PTY for child communications, make sure to close() it,
+        # otherwise long-running Invoke-using processes exhaust their file
+        # descriptors eventually.
+        if self.using_pty:
+            try:
+                os.close(self.parent_fd)
+            except Exception:
+                # If something weird happened preventing the close, there's
+                # nothing to be done about it now...
+                pass
 
 
 class Result(object):
@@ -1370,22 +1506,88 @@
         return encode_output(text, self.encoding)
 
 
-def normalize_hide(val):
+class Promise(Result):
+    """
+    A promise of some future `Result`, yielded from asynchronous execution.
+
+    This class' primary API member is `join`; instances may also be used as
+    context managers, which will automatically call `join` when the block
+    exits. In such cases, the context manager yields ``self``.
+
+    `Promise` also exposes copies of many `Result` attributes, specifically
+    those that derive from `~Runner.run` kwargs and not the result of command
+    execution. For example, ``command`` is replicated here, but ``stdout`` is
+    not.
+
+    .. versionadded:: 1.4
+    """
+
+    def __init__(self, runner):
+        """
+        Create a new promise.
+
+        :param runner:
+            An in-flight `Runner` instance making this promise.
+
+            Must already have started the subprocess and spun up IO threads.
+        """
+        self.runner = runner
+        # Basically just want exactly this (recently refactored) kwargs dict.
+        # TODO: consider proxying vs copying, but prob wait for refactor
+        for key, value in self.runner.result_kwargs.items():
+            setattr(self, key, value)
+
+    def join(self):
+        """
+        Block until associated subprocess exits, returning/raising the result.
+
+        This acts identically to the end of a synchronously executed ``run``,
+        namely that:
+
+        - various background threads (such as IO workers) are themselves
+          joined;
+        - if the subprocess exited normally, a `Result` is returned;
+        - in any other case (unforeseen exceptions, IO sub-thread
+          `.ThreadException`, `.Failure`, `.WatcherError`) the relevant
+          exception is raised here.
+
+        See `~Runner.run` docs, or those of the relevant classes, for further
+        details.
+        """
+        try:
+            return self.runner._finish()
+        finally:
+            self.runner._stop_everything()
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        self.join()
+
+
+def normalize_hide(val, out_stream=None, err_stream=None):
+    # Normalize to list-of-stream-names
     hide_vals = (None, False, "out", "stdout", "err", "stderr", "both", True)
     if val not in hide_vals:
         err = "'hide' got {!r} which is not in {!r}"
         raise ValueError(err.format(val, hide_vals))
     if val in (None, False):
-        hide = ()
+        hide = []
     elif val in ("both", True):
-        hide = ("stdout", "stderr")
+        hide = ["stdout", "stderr"]
     elif val == "out":
-        hide = ("stdout",)
+        hide = ["stdout"]
     elif val == "err":
-        hide = ("stderr",)
+        hide = ["stderr"]
     else:
-        hide = (val,)
-    return hide
+        hide = [val]
+    # Revert any streams that have been overridden from the default value
+    if out_stream is not None and "stdout" in hide:
+        hide.remove("stdout")
+    if err_stream is not None and "stderr" in hide:
+        hide.remove("stderr")
+    return tuple(hide)
 
 
 def default_encoding():
diff -Nru python-invoke-1.3.0+ds/invoke/util.py python-invoke-1.4.1+ds/invoke/util.py
--- python-invoke-1.3.0+ds/invoke/util.py	2019-10-11 10:31:04.000000000 +0000
+++ python-invoke-1.4.1+ds/invoke/util.py	2020-01-30 00:49:50.000000000 +0000
@@ -271,8 +271,7 @@
         # NOTE: it seems highly unlikely that a thread could still be
         # is_alive() but also have encountered an exception. But hey. Why not
         # be thorough?
-        alive = self.is_alive() and (self.exc_info is None)
-        return not alive
+        return (not self.is_alive()) and self.exc_info is not None
 
     def __repr__(self):
         # TODO: beef this up more
diff -Nru python-invoke-1.3.0+ds/invoke/_version.py python-invoke-1.4.1+ds/invoke/_version.py
--- python-invoke-1.3.0+ds/invoke/_version.py	2019-10-11 10:31:04.000000000 +0000
+++ python-invoke-1.4.1+ds/invoke/_version.py	2020-01-30 00:49:50.000000000 +0000
@@ -1,2 +1,2 @@
-__version_info__ = (1, 3, 0)
+__version_info__ = (1, 4, 1)
 __version__ = ".".join(map(str, __version_info__))
diff -Nru python-invoke-1.3.0+ds/LICENSE python-invoke-1.4.1+ds/LICENSE
--- python-invoke-1.3.0+ds/LICENSE	2019-10-11 10:31:04.000000000 +0000
+++ python-invoke-1.4.1+ds/LICENSE	2020-01-30 00:49:50.000000000 +0000
@@ -1,4 +1,4 @@
-Copyright (c) 2019 Jeff Forcier.
+Copyright (c) 2020 Jeff Forcier.
 All rights reserved.
 
 Redistribution and use in source and binary forms, with or without
diff -Nru python-invoke-1.3.0+ds/MANIFEST.in python-invoke-1.4.1+ds/MANIFEST.in
--- python-invoke-1.3.0+ds/MANIFEST.in	2019-10-11 10:31:04.000000000 +0000
+++ python-invoke-1.4.1+ds/MANIFEST.in	2020-01-30 00:49:50.000000000 +0000
@@ -7,4 +7,5 @@
 include dev-requirements.txt
 include tasks-requirements.txt
 recursive-include tests *
-recursive-exclude tests *.pyc *.pyo
+recursive-exclude * *.pyc *.pyo
+recursive-exclude **/__pycache__ *
diff -Nru python-invoke-1.3.0+ds/setup.cfg python-invoke-1.4.1+ds/setup.cfg
--- python-invoke-1.3.0+ds/setup.cfg	2019-10-11 10:31:04.000000000 +0000
+++ python-invoke-1.4.1+ds/setup.cfg	2020-01-30 00:49:50.000000000 +0000
@@ -8,4 +8,7 @@
 
 [tool:pytest]
 testpaths = tests
-python_files = *
\ No newline at end of file
+python_files = *
+filterwarnings =
+    once::Warning
+    ignore::DeprecationWarning
diff -Nru python-invoke-1.3.0+ds/sites/docs/invoke.rst python-invoke-1.4.1+ds/sites/docs/invoke.rst
--- python-invoke-1.3.0+ds/sites/docs/invoke.rst	2019-10-11 10:31:04.000000000 +0000
+++ python-invoke-1.4.1+ds/sites/docs/invoke.rst	2020-01-30 00:49:50.000000000 +0000
@@ -94,6 +94,18 @@
 
     Enable debug output.
 
+.. option:: --dry
+
+    Echo commands instead of actually running them; specifically, causes any
+    ``run`` calls to:
+
+    - Act as if the ``echo`` option has been turned on, printing the
+      command-to-be-run to stdout;
+    - Skip actual subprocess invocation (returning before any of that machinery
+      starts running);
+    - Return a dummy `~invoke.runners.Result` object with 'blank' values (empty
+      stdout/err strings, ``0`` exit code, etc).
+
 .. option:: -D, --list-depth=INT
 
     Limit :option:`--list` display to the specified number of levels, e.g.
diff -Nru python-invoke-1.3.0+ds/sites/www/changelog.rst python-invoke-1.4.1+ds/sites/www/changelog.rst
--- python-invoke-1.3.0+ds/sites/www/changelog.rst	2019-10-11 10:31:04.000000000 +0000
+++ python-invoke-1.4.1+ds/sites/www/changelog.rst	2020-01-30 00:49:50.000000000 +0000
@@ -2,6 +2,52 @@
 Changelog
 =========
 
+- :release:`1.4.1 <2020-01-29>`
+- :release:`1.3.1 <2020-01-29>`
+- :support:`586` Explicitly strip out ``__pycache__`` (and for good measure,
+  ``.py[co]``, which previously we only stripped from the ``tests/`` folder) in
+  our ``MANIFEST.in``, since at least some earlier releases erroneously
+  included such. Credit to Martijn Pieters for the report and Floris Lambrechts
+  for the patch.
+- :bug:`660` Fix an issue with `~invoke.run` & friends having intermittent
+  problems at exit time (symptom was typically about the exit code value being
+  ``None`` instead of an integer; often with an exception trace). Thanks to
+  Frank Lazzarini for the report and to the numerous others who provided
+  reproduction cases.
+- :bug:`518` Close pseudoterminals opened by the `~invoke.runners.Local` class
+  during ``run(..., pty=True)``. Previously, these were only closed
+  incidentally at process shutdown, causing file descriptor leakage in
+  long-running processes. Thanks to Jonathan Paulson for the report.
+- :release:`1.4.0 <2020-01-03>`
+- :bug:`637 major` A corner case in `~invoke.context.Context.run` caused
+  overridden streams to be unused if those streams were also set to be hidden
+  (eg ``run(command, hide=True, out_stream=StringIO())`` would result in no
+  writes to the ``StringIO`` object).
+
+  This has been fixed - hiding for a given stream is now ignored if that stream
+  has been set to some non-``None`` (and in the case of ``in_stream``,
+  non-``False``) value.
+- :bug:`- major` As part of feature work on :issue:`682`, we noticed that the
+  `~invoke.runners.Result` return value from `~invoke.context.Context.run` was
+  inconsistent between dry-run and regular modes; for example, the dry-run
+  version of the object lacked updated values for ``hide``, ``encoding`` and
+  ``env``. This has been fixed.
+- :feature:`682` (originally reported as :issue:`194`) Add asynchronous
+  behavior to `~invoke.runners.Runner.run`:
+
+  - Basic asynchronicity, where the method returns as soon as the subprocess
+    has started running, and that return value is an object with methods
+    allowing access to the final result.
+  - "Disowning" subprocesses entirely, which not only returns immediately but
+    also omits background threading, allowing the subprocesses to outlive
+    Invoke's own process.
+
+  See the updated API docs for the `~invoke.runners.Runner` for details on the
+  new ``asynchronous`` and ``disown`` kwargs enabling this behavior. Thanks to
+  ``@MinchinWeb`` for the original report.
+- :feature:`-` Never accompanied the top-level singleton `~invoke.run` (which
+  simply wraps an anonymous `~invoke.context.Context`'s ``run`` method) with
+  its logical sibling, `~invoke.sudo` - this has been remedied.
 - :release:`1.3.0 <2019-08-06>`
 - :feature:`324` Add basic dry-run support, in the form of a new
   :option:`--dry` CLI option and matching ``run.dry`` config setting, which
diff -Nru python-invoke-1.3.0+ds/tasks.py python-invoke-1.4.1+ds/tasks.py
--- python-invoke-1.3.0+ds/tasks.py	2019-10-11 10:31:04.000000000 +0000
+++ python-invoke-1.4.1+ds/tasks.py	2020-01-30 00:49:50.000000000 +0000
@@ -72,10 +72,23 @@
     return coverage_(c, report=report, opts=opts, tester=test)
 
 
+@task
+def regression(c, jobs=8):
+    """
+    Run an expensive, hard-to-test-in-pytest run() regression checker.
+
+    :param int jobs: Number of jobs to run, in total. Ideally num of CPUs.
+    """
+    os.chdir("integration/_support")
+    cmd = "seq {} | parallel -n0 --halt=now,fail=1 inv -c regression check"
+    c.run(cmd.format(jobs))
+
+
 ns = Collection(
     test,
     coverage,
     integration,
+    regression,
     vendorize,
     release,
     www,
diff -Nru python-invoke-1.3.0+ds/tests/concurrency.py python-invoke-1.4.1+ds/tests/concurrency.py
--- python-invoke-1.3.0+ds/tests/concurrency.py	2019-10-11 10:31:04.000000000 +0000
+++ python-invoke-1.4.1+ds/tests/concurrency.py	2020-01-30 00:49:50.000000000 +0000
@@ -32,14 +32,20 @@
             assert isinstance(wrapper.value, AttributeError)
 
         def exhibits_is_dead_flag(self):
+            # Spin up a thread that will except internally (can't put() on a
+            # None object)
             t = EHThread(target=self.worker, args=[None])
             t.start()
             t.join()
+            # Excepted -> it's dead
             assert t.is_dead
+            # Spin up a happy thread that can exit peacefully (it's not "dead",
+            # though...maybe we should change that terminology)
             t = EHThread(target=self.worker, args=[Queue()])
             t.start()
             t.join()
-            assert t.is_dead
+            # Not dead, just uh...sleeping?
+            assert not t.is_dead
 
     class via_subclassing:
         def setup(self):
@@ -73,11 +79,17 @@
             assert isinstance(wrapper.value, AttributeError)
 
         def exhibits_is_dead_flag(self):
+            # Spin up a thread that will except internally (can't put() on a
+            # None object)
             t = self.klass(queue=None)
             t.start()
             t.join()
+            # Excepted -> it's dead
             assert t.is_dead
+            # Spin up a happy thread that can exit peacefully (it's not "dead",
+            # though...maybe we should change that terminology)
             t = self.klass(queue=Queue())
             t.start()
             t.join()
-            assert t.is_dead
+            # Not dead, just uh...sleeping?
+            assert not t.is_dead
diff -Nru python-invoke-1.3.0+ds/tests/config.py python-invoke-1.4.1+ds/tests/config.py
--- python-invoke-1.3.0+ds/tests/config.py	2019-10-11 10:31:04.000000000 +0000
+++ python-invoke-1.4.1+ds/tests/config.py	2020-01-30 00:49:50.000000000 +0000
@@ -92,6 +92,8 @@
             # which override them, e.g. runner tests around warn=True, etc).
             expected = {
                 "run": {
+                    "asynchronous": False,
+                    "disown": False,
                     "dry": False,
                     "echo": False,
                     "echo_stdin": None,
@@ -641,7 +643,7 @@
             c._load_yml = Mock(side_effect=IOError(2, "aw nuts"))
             c.set_runtime_path("is-a.yml")  # Triggers use of _load_yml
             c.load_runtime()
-            mock_debug.assert_has_call("Didn't see any is-a.yml, skipping.")
+            mock_debug.assert_any_call("Didn't see any is-a.yml, skipping.")
 
         @raises(IOError)
         def non_missing_file_IOErrors_are_raised(self):
diff -Nru python-invoke-1.3.0+ds/tests/init.py python-invoke-1.4.1+ds/tests/init.py
--- python-invoke-1.3.0+ds/tests/init.py	2019-10-11 10:31:04.000000000 +0000
+++ python-invoke-1.4.1+ds/tests/init.py	2020-01-30 00:49:50.000000000 +0000
@@ -2,6 +2,8 @@
 
 import six
 
+from mock import patch
+
 import invoke
 import invoke.collection
 import invoke.exceptions
@@ -58,6 +60,9 @@
         def runner_class(self):
             assert invoke.Runner is invoke.runners.Runner
 
+        def promise_class(self):
+            assert invoke.Promise is invoke.runners.Promise
+
         def failure_class(self):
             assert invoke.Failure is invoke.runners.Failure
 
@@ -104,3 +109,18 @@
         def Call(self):
             # Starting to think we shouldn't bother with lowercase-c call...
             assert invoke.Call is invoke.tasks.Call
+
+    class offers_singletons:
+        @patch("invoke.Context")
+        def run(self, Context):
+            result = invoke.run("foo", bar="biz")
+            ctx = Context.return_value
+            ctx.run.assert_called_once_with("foo", bar="biz")
+            assert result is ctx.run.return_value
+
+        @patch("invoke.Context")
+        def sudo(self, Context):
+            result = invoke.sudo("foo", bar="biz")
+            ctx = Context.return_value
+            ctx.sudo.assert_called_once_with("foo", bar="biz")
+            assert result is ctx.sudo.return_value
diff -Nru python-invoke-1.3.0+ds/tests/merge_dicts.py python-invoke-1.4.1+ds/tests/merge_dicts.py
--- python-invoke-1.3.0+ds/tests/merge_dicts.py	2019-10-11 10:31:04.000000000 +0000
+++ python-invoke-1.4.1+ds/tests/merge_dicts.py	2020-01-30 00:49:50.000000000 +0000
@@ -101,10 +101,11 @@
         assert proj["foo"]["bar"]["biz"] == "proj value"
 
     def merge_file_types_by_reference(self):
-        d1 = {}
-        d2 = {"foo": open(__file__)}
-        merge_dicts(d1, d2)
-        assert d1["foo"].closed is False
+        with open(__file__) as fd:
+            d1 = {}
+            d2 = {"foo": fd}
+            merge_dicts(d1, d2)
+            assert d1["foo"].closed is False
 
 
 class copy_dict_:
diff -Nru python-invoke-1.3.0+ds/tests/runners.py python-invoke-1.4.1+ds/tests/runners.py
--- python-invoke-1.3.0+ds/tests/runners.py	2019-10-11 10:31:04.000000000 +0000
+++ python-invoke-1.4.1+ds/tests/runners.py	2020-01-30 00:49:50.000000000 +0000
@@ -3,6 +3,7 @@
 import struct
 import sys
 import termios
+import threading
 import types
 
 from io import BytesIO
@@ -15,19 +16,20 @@
 from mock import patch, Mock, call
 
 from invoke import (
-    Runner,
-    Local,
     CommandTimedOut,
-    Context,
     Config,
+    Context,
     Failure,
-    ThreadException,
-    SubprocessPipeError,
+    Local,
+    Promise,
     Responder,
-    WatcherError,
-    UnexpectedExit,
-    StreamWatcher,
     Result,
+    Runner,
+    StreamWatcher,
+    SubprocessPipeError,
+    ThreadException,
+    UnexpectedExit,
+    WatcherError,
 )
 from invoke.runners import default_encoding
 from invoke.terminals import WINDOWS
@@ -120,6 +122,8 @@
 
 
 class Runner_:
+    _stop_methods = ["generate_result", "stop", "stop_timer"]
+
     # NOTE: these copies of _run and _runner form the base case of "test Runner
     # subclasses via self._run/_runner helpers" functionality. See how e.g.
     # Local_ uses the same approach but bakes in the dummy class used.
@@ -483,6 +487,13 @@
             assert sys.stdout.getvalue() == ""
 
         @trap
+        def overridden_out_is_never_hidden(self):
+            out = StringIO()
+            self._runner(out="sup").run(_, out_stream=out, hide=True)
+            assert out.getvalue() == "sup"
+            assert sys.stdout.getvalue() == ""
+
+        @trap
         def err_can_be_overridden(self):
             "err_stream can be overridden"
             err = StringIO()
@@ -491,6 +502,13 @@
             assert sys.stderr.getvalue() == ""
 
         @trap
+        def overridden_err_is_never_hidden(self):
+            err = StringIO()
+            self._runner(err="sup").run(_, err_stream=err, hide=True)
+            assert err.getvalue() == "sup"
+            assert sys.stderr.getvalue() == ""
+
+        @trap
         def pty_defaults_to_sys(self):
             self._runner(out="sup").run(_, pty=True)
             assert sys.stdout.getvalue() == "sup"
@@ -1385,6 +1403,97 @@
                 runner.run(_)
             runner.stop.assert_called_once_with()
 
+    class asynchronous:
+        def returns_Promise_immediately_and_finishes_on_join(self):
+            # Dummy subclass with controllable process_is_finished flag
+            class _Finisher(_Dummy):
+                _finished = False
+
+                @property
+                def process_is_finished(self):
+                    return self._finished
+
+            runner = _Finisher(Context())
+            # Set up mocks and go
+            runner.start = Mock()
+            for method in self._stop_methods:
+                setattr(runner, method, Mock())
+            result = runner.run(_, asynchronous=True)
+            # Got a Promise (its attrs etc are in its own test subsuite)
+            assert isinstance(result, Promise)
+            # Started, but did not stop (as would've happened for disown)
+            assert runner.start.called
+            for method in self._stop_methods:
+                assert not getattr(runner, method).called
+            # Set proc completion flag to truthy and join()
+            runner._finished = True
+            result.join()
+            for method in self._stop_methods:
+                assert getattr(runner, method).called
+
+        @trap
+        def hides_output(self):
+            # Run w/ faux subproc stdout/err data, but async
+            self._runner(out="foo", err="bar").run(_, asynchronous=True).join()
+            # Expect that default out/err streams did not get printed to.
+            assert sys.stdout.getvalue() == ""
+            assert sys.stderr.getvalue() == ""
+
+        def does_not_forward_stdin(self):
+            class MockedHandleStdin(_Dummy):
+                pass
+
+            MockedHandleStdin.handle_stdin = Mock()
+            runner = self._runner(klass=MockedHandleStdin)
+            runner.run(_, asynchronous=True).join()
+            # As with the main test for setting this to False, we know that
+            # when stdin is disabled, the handler is never even called (no
+            # thread is created for it).
+            assert not MockedHandleStdin.handle_stdin.called
+
+        def leaves_overridden_streams_alone(self):
+            # NOTE: technically a duplicate test of the generic tests for #637
+            # re: intersect of hide and overridden streams. But that's an
+            # implementation detail so this is still valuable.
+            klass = self._mock_stdin_writer()
+            out, err, in_ = StringIO(), StringIO(), StringIO("hallo")
+            runner = self._runner(out="foo", err="bar", klass=klass)
+            runner.run(
+                _,
+                asynchronous=True,
+                out_stream=out,
+                err_stream=err,
+                in_stream=in_,
+            ).join()
+            assert out.getvalue() == "foo"
+            assert err.getvalue() == "bar"
+            assert klass.write_proc_stdin.called  # lazy
+
+    class disown:
+        @patch.object(threading.Thread, "start")
+        def starts_and_returns_None_but_does_nothing_else(self, thread_start):
+            runner = Runner(Context())
+            runner.start = Mock()
+            not_called = self._stop_methods + ["wait"]
+            for method in not_called:
+                setattr(runner, method, Mock())
+            result = runner.run(_, disown=True)
+            # No Result object!
+            assert result is None
+            # Subprocess kicked off
+            assert runner.start.called
+            # No timer or IO threads started
+            assert not thread_start.called
+            # No wait or shutdown related Runner methods called
+            for method in not_called:
+                assert not getattr(runner, method).called
+
+        def cannot_be_given_alongside_asynchronous(self):
+            with raises(ValueError) as info:
+                self._runner().run(_, asynchronous=True, disown=True)
+            sentinel = "Cannot give both 'asynchronous' and 'disown'"
+            assert sentinel in str(info.value)
+
 
 class _FastLocal(Local):
     # Neuter this for same reason as in _Dummy above
@@ -1475,6 +1584,12 @@
                 assert e.type == OSError
                 assert str(e.value) == "wat"
 
+        @mock_pty(os_close_error=True)
+        def stop_mutes_errors_on_pty_close(self):
+            # Another doesn't-blow-up test, this time around os.close() of the
+            # pty itself (due to os_close_error=True)
+            self._run(_, pty=True)
+
         class fallback:
             @mock_pty(isatty=False)
             def can_be_overridden_by_kwarg(self):
@@ -1647,3 +1762,63 @@
         def encodes_with_result_encoding(self, encode):
             Result(stdout="foo", encoding="utf-16").tail("stdout")
             encode.assert_called_once_with("\n\nfoo", "utf-16")
+
+
+class Promise_:
+    def exposes_read_only_run_params(self):
+        runner = _runner()
+        promise = runner.run(
+            _, pty=True, encoding="utf-17", shell="sea", asynchronous=True
+        )
+        assert promise.command == _
+        assert promise.pty is True
+        assert promise.encoding == "utf-17"
+        assert promise.shell == "sea"
+        assert not hasattr(promise, "stdout")
+        assert not hasattr(promise, "stderr")
+
+    class join:
+        # NOTE: high level Runner lifecycle mechanics of join() (re: wait(),
+        # process_is_finished() etc) are tested in main suite.
+
+        def returns_Result_on_success(self):
+            result = _runner().run(_, asynchronous=True).join()
+            assert isinstance(result, Result)
+            # Sanity
+            assert result.command == _
+            assert result.exited == 0
+
+        def raises_main_thread_exception_on_kaboom(self):
+            runner = _runner(klass=_GenericExceptingRunner)
+            with raises(_GenericException):
+                runner.run(_, asynchronous=True).join()
+
+        def raises_subthread_exception_on_their_kaboom(self):
+            class Kaboom(_Dummy):
+                def handle_stdout(self, **kwargs):
+                    raise OhNoz()
+
+            runner = _runner(klass=Kaboom)
+            promise = runner.run(_, asynchronous=True)
+            with raises(ThreadException) as info:
+                promise.join()
+            assert isinstance(info.value.exceptions[0].value, OhNoz)
+
+        def raises_Failure_on_failure(self):
+            runner = _runner(exits=1)
+            promise = runner.run(_, asynchronous=True)
+            with raises(Failure):
+                promise.join()
+
+    class context_manager:
+        def calls_join_or_wait_on_close_of_block(self):
+            promise = _runner().run(_, asynchronous=True)
+            promise.join = Mock()
+            with promise:
+                pass
+            promise.join.assert_called_once_with()
+
+        def yields_self(self):
+            promise = _runner().run(_, asynchronous=True)
+            with promise as value:
+                assert value is promise
diff -Nru python-invoke-1.3.0+ds/tests/_util.py python-invoke-1.4.1+ds/tests/_util.py
--- python-invoke-1.3.0+ds/tests/_util.py	2019-10-11 10:31:04.000000000 +0000
+++ python-invoke-1.4.1+ds/tests/_util.py	2020-01-30 00:49:50.000000000 +0000
@@ -185,6 +185,7 @@
     skip_asserts=False,
     insert_os=False,
     be_childish=False,
+    os_close_error=False,
 ):
     # Windows doesn't have ptys, so all the pty tests should be
     # skipped anyway...
@@ -203,10 +204,10 @@
         def wrapper(*args, **kwargs):
             args = list(args)
             pty, os, ioctl = args.pop(), args.pop(), args.pop()
-            # Don't actually fork, but pretend we did & that main thread is
-            # also the child (pid 0) to trigger execve call; & give 'parent fd'
-            # of 1 (stdout).
-            pty.fork.return_value = (12345 if be_childish else 0), 1
+            # Don't actually fork, but pretend we did (with "our" pid differing
+            # depending on be_childish) & give 'parent fd' of 3 (typically,
+            # first allocated non-stdin/out/err FD)
+            pty.fork.return_value = (12345 if be_childish else 0), 3
             # We don't really need to care about waiting since not truly
             # forking/etc, so here we just return a nonzero "pid" + sentinel
             # wait-status value (used in some tests about WIFEXITED etc)
@@ -221,7 +222,7 @@
             err_file = BytesIO(b(err))
 
             def fakeread(fileno, count):
-                fd = {1: out_file, 2: err_file}[fileno]
+                fd = {3: out_file, 2: err_file}[fileno]
                 ret = fd.read(count)
                 # If asked, fake a Linux-platform trailing I/O error.
                 if not ret and trailing_error:
@@ -229,15 +230,18 @@
                 return ret
 
             os.read.side_effect = fakeread
+            if os_close_error:
+                os.close.side_effect = IOError
             if insert_os:
                 args.append(os)
+
+            # Do the thing!!!
             f(*args, **kwargs)
+
             # Short-circuit if we raised an error in fakeread()
             if trailing_error:
                 return
             # Sanity checks to make sure the stuff we mocked, actually got ran!
-            # TODO: inject our mocks back into the tests so they can make their
-            # own assertions if desired
             pty.fork.assert_called_with()
             # Skip rest of asserts if we pretended to be the child
             if be_childish:
@@ -250,6 +254,8 @@
                     assert getattr(os, name).called
                 # Ensure at least one of the exit status getters was called
                 assert os.WEXITSTATUS.called or os.WTERMSIG.called
+                # Ensure something closed the pty FD
+                os.close.assert_called_once_with(3)
 
         return wrapper
 
diff -Nru python-invoke-1.3.0+ds/.travis.yml python-invoke-1.4.1+ds/.travis.yml
--- python-invoke-1.3.0+ds/.travis.yml	2019-10-11 10:31:04.000000000 +0000
+++ python-invoke-1.4.1+ds/.travis.yml	2020-01-30 00:49:50.000000000 +0000
@@ -1,6 +1,6 @@
 language: python
 sudo: required
-dist: trusty
+dist: xenial
 cache:
   directories:
     - $HOME/.cache/pip
@@ -9,15 +9,18 @@
   - "3.4"
   - "3.5"
   - "3.6"
-  - "3.7-dev"
-  - "nightly"
+  - "3.7"
+  - "3.8-dev"
   - "pypy"
   - "pypy3"
 matrix:
-  # NOTE: comment this out if we have to reinstate any allow_failures, because
-  # of https://github.com/travis-ci/travis-ci/issues/1696 (multiple
-  # notifications)
-  fast_finish: true
+  allow_failures:
+    - python: "3.8-dev"
+# WHY does this have to be in before_install and not install? o_O
+before_install:
+  # Used by 'inv regression' (more performant/safe/likely to expose real issues
+  # than in-Python threads...)
+  - sudo apt-get -y install parallel
 install:
   # For some reason Travis' build envs have wildly different pip/setuptools
   # versions between minor Python versions, and this can cause many hilarious
@@ -47,11 +50,17 @@
 script:
   # Execute full test suite + coverage, as the new sudo-capable user
   - inv travis.sudo-coverage
+  # Perform extra "not feasible inside pytest for no obvious reason" tests
+  - inv regression
   # Websites build OK? (Not on PyPy3, Sphinx is all "who the hell are you?" =/
   - "if [[ $TRAVIS_PYTHON_VERSION != 'pypy3' ]]; then inv sites; fi"
   # Doctests in websites OK? (Same caveat as above...)
   - "if [[ $TRAVIS_PYTHON_VERSION != 'pypy3' ]]; then inv www.doctest; fi"
   # Did we break setup.py?
+  # NOTE: sometime in 2019 travis grew a bizarre EnvironmentError problem
+  # around inability to overwrite/remote __pycache__ dirs...this attempts to
+  # workaround
+  - "find . -type d -name __pycache__ | sudo xargs rm -rf"
   - inv travis.test-installation --package=invoke --sanity="inv --list"
   # Test distribution builds, including some package_data based stuff
   # (completion script printing)