Fix terminal handoffs and signaling found in review

- A lend takes the terminal only while cmd2's job owns it. After the shell's bg, a write that
  filled the pipe gave the terminal to the consumer and then to cmd2, and the shell could no
  longer read its terminal. Now a consumer that needs it stops, and cmd2's job stops with it.
- _restore_output() lends the terminal before closing the pipe, so a consumer that sets its
  terminal modes on EOF, such as vim -, is not stopped with SIGTTOU.
- ProcReader's terminal_fd and pipeline arguments work on their own: the watcher starts with
  the reader, _manage_terminal() only installs the Ctrl-Z handler, and wait() reaps a producer
  that joined a pipeline.
- A write interrupted by Ctrl-C after sending some chunks reports them and raises from the next
  write, so BufferedWriter does not send them again.
- do_shell reads the pipeline's group once, and send_sigint() and _signal_pipeline() share one
  group lookup that never signals cmd2's own group.
- A SIGTSTP handler installed outside Python, which signal.getsignal() reports as None, is left
  in place rather than failing to be restored.
- The do_shell stdout choice no longer needs a type: ignore, which made its line too long.

Author: tleonhardt

cmd2/cmd2.py

@@ -3506,24 +3506,27 @@ def _restore_output(self, statement: Statement, saved_redir_state: utils.Redirec
                     # Restore self.stdout first, so that a failure to end the redirection cannot leave it in place
                     redirected_stdout = self.stdout
                     self.stdout = cast(TextIO, saved_redir_state.saved_self_stdout)
-
-                    try:
-                        # If we redirected output to the clipboard
-                        if (
-                            statement.redirector in (constants.REDIRECTION_OVERWRITE, constants.REDIRECTION_APPEND)
-                            and not statement.redirect_to
-                        ):
-                            redirected_stdout.seek(0)
-                            write_to_paste_buffer(redirected_stdout.read())
-                    finally:
-                        with contextlib.suppress(BrokenPipeError):
-                            # Close the file or pipe that stdout was redirected to
-                            redirected_stdout.close()
-
-                    # Check if we need to wait for the process being piped to. Handing the terminal back as it finishes can
-                    # fail, for example after a hangup.
-                    if self._cur_pipe_proc_reader is not None:
-                        self._cur_pipe_proc_reader.wait()
+                    pipe_proc_reader = self._cur_pipe_proc_reader
+
+                    # A terminal pipeline has the terminal before its consumer reads EOF, which may set its terminal modes as
+                    # it does, and keeps it until it exits. Handing the terminal back then can fail, as after a hangup.
+                    with pipe_proc_reader._lend_terminal() if pipe_proc_reader is not None else contextlib.nullcontext():
+                        try:
+                            # If we redirected output to the clipboard
+                            if (
+                                statement.redirector in (constants.REDIRECTION_OVERWRITE, constants.REDIRECTION_APPEND)
+                                and not statement.redirect_to
+                            ):
+                                redirected_stdout.seek(0)
+                                write_to_paste_buffer(redirected_stdout.read())
+                        finally:
+                            with contextlib.suppress(BrokenPipeError):
+                                # Close the file or pipe that stdout was redirected to
+                                redirected_stdout.close()
+
+                        # Check if we need to wait for the process being piped to
+                        if pipe_proc_reader is not None:
+                            pipe_proc_reader.wait()
             finally:
                 # These are restored regardless of whether the command redirected, or whether restoring it failed: a pipeline
                 # left current would keep ppaged() from paging and send Ctrl-C to a process group that is gone.
@@ -5004,32 +5007,34 @@ def do_shell(self, args: argparse.Namespace) -> None:
         # handlers.
         pipeline = self._cur_pipe_proc_reader
         writer = utils._pipeline_writer_of(self.stdout)
-        joined_writer = None
+        job_group = None
         if (
             pipeline is not None
             and writer is not None
             and writer.feeds(pipeline)
             and threading.current_thread() is threading.main_thread()
-            and pipeline._terminal_group is not None
         ):
-            joined_writer = writer
+            # Read the group once: the pipeline's watcher may reap the consumer at any moment, which leaves none to join.
+            job_group = pipeline._terminal_group
+        joined_writer = writer if job_group is not None else None
 
         # Prevent KeyboardInterrupts while in the shell process. The shell process still receives the SIGINT: it is in our
         # process group or in the foreground pipeline's.
         with self.sigint_protection, contextlib.ExitStack() as terminal_stack:
             if pipeline is not None and joined_writer is not None:
-                kwargs["process_group"] = pipeline._terminal_group
+                kwargs["process_group"] = job_group
                 terminal_stack.enter_context(pipeline._lend_terminal())
             while True:
                 try:
                     # For any stream that is a StdSim, we will use a pipe so we can capture its output. A command joining the
                     # pipeline writes to its consumer's pipe directly, after what cmd2 wrote before it. It is spawned inside
                     # the lend, which blocks SIGTTOU, and with the job's Ctrl-Z behavior.
+                    stdout: Any = self.stdout
                     if joined_writer is not None:
                         self.stdout.flush()
-                        stdout: Any = joined_writer.producer_fileno()
-                    else:
-                        stdout = subprocess.PIPE if isinstance(self.stdout, utils.StdSim) else self.stdout  # type: ignore[unreachable]
+                        stdout = joined_writer.producer_fileno()
+                    elif isinstance(stdout, utils.StdSim):
+                        stdout = subprocess.PIPE
                     with (
                         utils._sigttou_mask(block=False) if joined_writer is not None else contextlib.nullcontext(),
                         utils._session_leader_job_stops(pipeline) if joined_writer is not None else contextlib.nullcontext(),
@@ -5058,8 +5063,6 @@ def do_shell(self, args: argparse.Namespace) -> None:
             # its watcher are gone, the same wait relays the command's own stops, such as Ctrl-Z.
             joined_pipeline = pipeline if joined_writer is not None else None
             proc_reader = utils.ProcReader(proc, self.stdout, sys.stderr, pipeline=joined_pipeline)
-            if joined_pipeline is not None:
-                proc_reader._wait_for_exit()
             proc_reader.wait()
 
             # Save the return code of the application for use in a pyscript

cmd2/utils.py

@@ -622,7 +622,9 @@ def _session_leader_job_stops(pipeline: "ProcReader | None" = None) -> Iterator[
     """
     import signal
 
-    if os.getpgrp() != os.getsid(0):
+    # A handler installed outside Python is reported as None, and Python could not reinstall it after the spawn. So it stays,
+    # and the pipeline's watcher continues any stop in the job, having no handler to relay it to.
+    if os.getpgrp() != os.getsid(0) or signal.getsignal(signal.SIGTSTP) is None:
         yield
         return
     # A stop the pipeline's watcher relayed to cmd2 meanwhile would be discarded, and the watcher would wait for cmd2 to resume
@@ -700,6 +702,10 @@ def __init__(
         if self._proc.stderr is not None:
             self._err_thread.start()
 
+        # A terminal pipeline is reaped only by its watcher, which relays its stops once _manage_terminal() sets up job control
+        if terminal_fd is not None:
+            threading.Thread(name="pipe_job", target=self._wait_for_job, args=(terminal_fd,), daemon=True).start()
+
     def send_sigint(self) -> None:
         """Send a SIGINT to the process similar to if <Ctrl>+C were pressed."""
         import signal
@@ -710,19 +716,8 @@ def send_sigint(self) -> None:
             self._proc.send_signal(signal.CTRL_BREAK_EVENT)
         else:
             # Since cmd2 uses shell=True in its Popen calls, we need to send the SIGINT to the whole process group to make sure
-            # it propagates further than the shell. Once reaped, the process's ID may already belong to another process.
-            group_id = None
-            if self._proc.returncode is None:
-                with contextlib.suppress(ProcessLookupError):
-                    group_id = os.getpgid(self._proc.pid)
-            if group_id is None:
-                group_id = self._joined_group()
-            if group_id is None:
-                return
-            # Never re-signal our own group: other ProcReader callers may share it and already received Ctrl-C.
-            if group_id != os.getpgrp():
-                with contextlib.suppress(ProcessLookupError, PermissionError):
-                    os.killpg(group_id, signal.SIGINT)
+            # it propagates further than the shell.
+            self._signal_pipeline(signal.SIGINT)
 
     def terminate(self) -> None:
         """Terminate the process."""
@@ -764,8 +759,16 @@ def _signal_pipeline(self, signum: int) -> bool:
         The group may be gone, or hold only processes cmd2 may not signal: a zombie on
         macOS, or a program running as another user, such as sudo.
         """
-        group_id = self._proc.pid if self._proc.returncode is None else self._joined_group()
+        group_id = None
+        # Once reaped, the process's ID may already belong to another process.
+        if self._proc.returncode is None:
+            with contextlib.suppress(ProcessLookupError):
+                group_id = os.getpgid(self._proc.pid)
         if group_id is None:
+            group_id = self._joined_group()
+        # Never signal our own group: that would stop or continue cmd2 itself, and other ProcReader callers may share it and
+        # already received Ctrl-C.
+        if group_id is None or group_id == os.getpgrp():
             return False
         try:
             os.killpg(group_id, signum)
@@ -781,7 +784,7 @@ def _set_foreground_group(terminal_fd: int, group_id: int) -> None:
 
     @contextlib.contextmanager
     def _manage_terminal(self) -> Iterator[None]:
-        """Watch the pipeline and suspend the shell's whole job on the main thread."""
+        """Suspend the shell's whole job on the main thread, and relay the pipeline's stops to it."""
         import signal
 
         terminal_fd = self._terminal_fd
@@ -830,18 +833,22 @@ def suspend_job(signum: int, frame: Any) -> None:
                         self._suspension_lock.release()
                     self._job_resumed.set()
 
-        self._stop_handler = suspend_job
-        signal.signal(signal.SIGTSTP, suspend_job)
+        # A handler installed outside Python, such as by an application embedding it, is reported as None, and Python could not
+        # reinstall it. So it stays: Ctrl-Z does only what it does, and the watcher continues any stop in the pipeline's job,
+        # having no handler to relay it to.
+        if previous_handler is not None:
+            self._stop_handler = suspend_job
+            signal.signal(signal.SIGTSTP, suspend_job)
         try:
-            threading.Thread(name="pipe_job", target=self._wait_for_job, args=(terminal_fd,), daemon=True).start()
             yield
         finally:
             # The watcher may outlive job control, if waiting for the pipeline failed. Without suspend_job(), a stop it relayed
             # would stop cmd2 with nothing to resume either. So detach it first, under the lock a suspension holds, and it
             # relays no more.
             with self._suspension():
                 self._detached = True
-                signal.signal(signal.SIGTSTP, previous_handler)
+                if previous_handler is not None:
+                    signal.signal(signal.SIGTSTP, previous_handler)
 
     @contextlib.contextmanager
     def _suspension(self) -> Iterator[None]:
@@ -875,7 +882,11 @@ def _lend_terminal(self) -> Iterator[None]:
         with _sigttou_mask(block=True):
             with self._terminal_lock:
                 try:
-                    self._set_foreground_group(terminal_fd, self._proc.pid)
+                    # Lend only a terminal cmd2's job owns. Once the shell's bg has continued cmd2 in the background, the shell
+                    # owns it, and taking it would leave the shell unable to read its terminal. A consumer that needs it then
+                    # stops, and the watcher relays that stop to cmd2's job, as a shell's own background pipeline would stop.
+                    if os.tcgetpgrp(terminal_fd) in (self._original_group, self._proc.pid):
+                        self._set_foreground_group(terminal_fd, self._proc.pid)
                 except OSError as error:
                     # The group can disappear before the watcher has recorded its exit. Linux reports a group that no longer
                     # exists as EPERM.
@@ -996,10 +1007,11 @@ def _suspend_with_cmd2(self, terminal_fd: int, seen: int) -> None:
         with self._suspension():
             if self._suspensions != seen or self._detached:
                 return
-            # Only suspend_job() resumes a pipeline stopped for a relay. Should command code have replaced it as SIGTSTP's
-            # handler, or ignored the signal, the pipeline would stay stopped for good. So it goes on, as when cmd2 ignores
-            # Ctrl-Z. Like a suspension, that deals with every stop reported before it.
-            if signal.getsignal(signal.SIGTSTP) is not self._stop_handler:
+            # Only suspend_job() resumes a pipeline stopped for a relay. Should it not be SIGTSTP's handler, because command
+            # code replaced it or ignored the signal, or job control left a handler installed outside Python, the pipeline
+            # would stay stopped for good. So it goes on, as when cmd2 ignores Ctrl-Z. Like a suspension, that deals with
+            # every stop reported before it.
+            if self._stop_handler is None or signal.getsignal(signal.SIGTSTP) is not self._stop_handler:
                 self._signal_pipeline(signal.SIGCONT)
                 self._suspensions += 1
                 return
@@ -1078,6 +1090,9 @@ def wait(self) -> None:
         if self._terminal_fd is not None:
             with self._lend_terminal():
                 self._wait_for_exit()
+        elif self._pipeline is not None:
+            # Only this wait reaps a producer in a pipeline's job. Until it does, the reader threads wait for its return code.
+            self._wait_for_exit()
         if self._out_thread.is_alive():
             self._out_thread.join()
         if self._err_thread.is_alive():
@@ -1281,6 +1296,8 @@ def __init__(self, fd: int, reader: ProcReader, interruptible: Callable[[], bool
         self._relay_lock = threading.Lock()
         self._poller = select.poll()
         self._poller.register(fd, select.POLLOUT)
+        # Set when Ctrl-C interrupted a write that had sent part of its data: see write()
+        self._interrupted = False
 
     def fileno(self) -> int:
         """Return a descriptor for subprocesses, such as a shell command's stdout.
@@ -1334,10 +1351,18 @@ def write(self, b: Any) -> int:
         on its own for the handler to run. The descriptor itself stays blocking: a shell
         command inherits it, and a producer that found it non-blocking would fail with
         EAGAIN once the pipe filled.
+
+        Ctrl-C can raise KeyboardInterrupt between two chunks, once some are in the pipe.
+        BufferedWriter takes an exception to mean that nothing was written, and would send
+        those chunks again. So such a write reports what it sent, and the next one, which
+        BufferedWriter makes straight away for the rest, raises the interrupt instead.
         """
         import select
         import signal
 
+        if self._interrupted:
+            self._interrupted = False
+            raise KeyboardInterrupt
         view = memoryview(b).cast("B")
         fd = super().fileno()
         written = 0
@@ -1367,6 +1392,10 @@ def write(self, b: Any) -> int:
                 if self._reader._proc.returncode in (-signal.SIGINT, 128 + signal.SIGINT):
                     raise KeyboardInterrupt from None
             raise
+        except KeyboardInterrupt:
+            if not written:
+                raise
+            self._interrupted = True
         return written
 
 

tests/test_cmd2.py

@@ -485,6 +485,41 @@ def popen(*args, **kwargs):
     assert spawned_while_lent == [True, False]
 
 
+@pytest.mark.skipif(sys.platform == "win32", reason="POSIX process groups")
+def test_shell_reads_the_pipelines_group_once(base_app) -> None:
+    """The pipeline's watcher can reap the consumer at any moment, after which it has no group to join.
+
+    Deciding to join from one reading and joining with another could spawn the command in
+    cmd2's own group while treating it as the pipeline's: Ctrl-Z would then stop cmd2 itself.
+    """
+    import contextlib
+    import subprocess
+    from unittest import mock
+
+    leader = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(30)"], process_group=0)
+    spawned = []
+    real_popen = subprocess.Popen
+
+    def popen(*args, **kwargs):
+        spawned.append(kwargs)
+        return real_popen(*args, **kwargs)
+
+    pipeline = mock.Mock(_lend_terminal=contextlib.nullcontext, _suspension=contextlib.nullcontext)
+    # The consumer is reaped right after the first reading.
+    type(pipeline)._terminal_group = mock.PropertyMock(side_effect=[leader.pid, None, None])
+    base_app._cur_pipe_proc_reader = pipeline
+    base_app.stdout, read_fd = pipeline_stdout(pipeline)
+    try:
+        with mock.patch("subprocess.Popen", popen):
+            base_app.do_shell("echo joined")
+        assert spawned[0]["process_group"] == leader.pid
+        base_app.stdout.close()
+        assert read_all(read_fd) == b"joined\n"
+    finally:
+        leader.kill()
+        leader.wait()
+
+
 @pytest.mark.skipif(sys.platform == "win32", reason="POSIX shell executable")
 def test_shell_permission_error_unrelated_to_pipeline(base_app, tmp_path, monkeypatch) -> None:
     unusable_shell = tmp_path / "shell"
@@ -1114,7 +1149,7 @@ def test_restore_output_resets_pipe_state_when_the_wait_fails(base_app) -> None:
     saved_stdout = base_app.stdout
     saved = cmd2.utils.RedirectionSavedState(saved_stdout, None, False)
     saved.redirecting = True
-    reader = mock.Mock()
+    reader = mock.Mock(_lend_terminal=contextlib.nullcontext)
     reader.wait.side_effect = OSError(errno.EIO, "terminal hung up")
     base_app._cur_pipe_proc_reader = reader
     base_app._redirecting = True
@@ -1128,6 +1163,42 @@ def test_restore_output_resets_pipe_state_when_the_wait_fails(base_app) -> None:
     assert base_app._redirecting is False
 
 
+def test_restore_output_lends_the_terminal_before_the_consumer_reads_eof(base_app) -> None:
+    """A consumer may set its terminal modes once it reads EOF, as `vim -` does after reading its input.
+
+    From the background, that stops it with SIGTTOU, and on macOS the call then fails with
+    EINTR as it is continued. So the pipeline has the terminal before its pipe closes.
+    """
+    import contextlib
+
+    statement = base_app.statement_parser.parse("help | vim -")
+    saved = cmd2.utils.RedirectionSavedState(base_app.stdout, None, False)
+    saved.redirecting = True
+    lent = []
+    events = []
+
+    @contextlib.contextmanager
+    def lend_terminal():
+        lent.append(True)
+        try:
+            yield
+        finally:
+            lent.pop()
+
+    reader = mock.Mock(_lend_terminal=lend_terminal)
+    reader.wait.side_effect = lambda: events.append(("wait", bool(lent)))
+    redirected = mock.Mock()
+    redirected.close.side_effect = lambda: events.append(("close", bool(lent)))
+    base_app._cur_pipe_proc_reader = reader
+    base_app._redirecting = True
+    base_app.stdout = redirected
+
+    base_app._restore_output(statement, saved)
+
+    assert events == [("close", True), ("wait", True)]
+    assert not lent
+
+
 @pytest.mark.parametrize("failure", ["clipboard", "close"])
 def test_restore_output_restores_stdout_when_ending_the_redirection_fails(base_app, mocker, failure) -> None:
     """A failed clipboard write, or a pipe that fails to close, must not leave stdout redirected.