Run shell producers inside terminal pipelines and isolate worker-thre…

…ad pipes A shell command piped to an interactive consumer, such as `shell git log | less`, hung: the child inherited the raw pipe descriptor, bypassing the writer that lends the terminal per write, so the consumer stopped on its first terminal access while the producer blocked on the full pipe. do_shell now spawns its child in the pipeline's process group and lends the terminal for the child's lifetime, so the consumer keeps the terminal and Ctrl-C and Ctrl-Z reach both processes as in a shell pipeline. If the pipeline exits before the child can join its group, the child runs in our own group as before. Because such a producer can outlive the consumer that led the group, ProcReader.send_sigint() falls back to the leader's pid as the group id once the leader has been reaped, rather than doing nothing. A pipe started off the main thread could not install job-control signal handlers and failed after Popen, leaving the child unreaped. The terminal pipeline path is now taken only on the main thread; elsewhere the pipeline keeps running in its own session.
tleonhardt · Sep 12, 2026 · Sep 12, 2026 · Sep 12, 2026 · Sep 12, 2026 · Sep 12, 2026
commit a7a733f6314e9285b91bbe6d8b69c36a7e71b66f
@@ -3554,12 +3554,15 @@ def _redirect_output(self, statement: Statement) -> utils.RedirectionSavedState:
 
             terminal_fd = None
             if sys.platform != "win32":
-                for stream in (pipe_stdout, pipe_stderr):
-                    if stream is not None and stream.isatty():
-                        with contextlib.suppress(OSError, ValueError):
-                            if os.tcgetpgrp(stream.fileno()) == os.getpgrp():
-                                terminal_fd = stream.fileno()
-                                break
+                # Job control installs signal handlers, which only the main thread may do.
+                # Elsewhere, keep the pipeline in its own session as before.
+                if threading.current_thread() is threading.main_thread():
+                    for stream in (pipe_stdout, pipe_stderr):
+                        if stream is not None and stream.isatty():
+                            with contextlib.suppress(OSError, ValueError):
+                                if os.tcgetpgrp(stream.fileno()) == os.getpgrp():
+                                    terminal_fd = stream.fileno()
+                                    break
                 if terminal_fd is None:
                     kwargs["start_new_session"] = True
                 else:
@@ -5248,17 +5251,37 @@ def do_shell(self, args: argparse.Namespace) -> None:
         utils.expand_user_in_tokens(tokens)
         expanded_command = " ".join(tokens)
 
-        # Prevent KeyboardInterrupts while in the shell process. The shell process will
-        # still receive the SIGINT since it is in the same process group as us.
-        with self.sigint_protection:
-            # For any stream that is a StdSim, we will use a pipe so we can capture its output
-            proc = subprocess.Popen(  # noqa: S602
-                expanded_command,
-                stdout=subprocess.PIPE if isinstance(self.stdout, utils.StdSim) else self.stdout,  # type: ignore[unreachable]
-                stderr=subprocess.PIPE if isinstance(sys.stderr, utils.StdSim) else sys.stderr,
-                shell=True,
-                **kwargs,
-            )
+        # A terminal pipeline's consumer needs the terminal to drain the pipe, but a shell
+        # command writes into that pipe itself rather than through self.stdout, which lends
+        # the terminal per write. Run the command inside the pipeline's job instead, for as
+        # long as it runs: the consumer keeps the terminal, and Ctrl-C and Ctrl-Z reach both
+        # processes, as they would in a shell pipeline.
+        pipeline = self._cur_pipe_proc_reader
+        pipeline_group = None
+        if pipeline is not None and not isinstance(self.stdout, utils.StdSim):  # type: ignore[unreachable]
+            pipeline_group = pipeline.terminal_group
+
+        # 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 pipeline_group is not None:
+                kwargs["process_group"] = pipeline_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
+                    proc = subprocess.Popen(  # noqa: S602
+                        expanded_command,
+                        stdout=subprocess.PIPE if isinstance(self.stdout, utils.StdSim) else self.stdout,  # type: ignore[unreachable]
+                        stderr=subprocess.PIPE if isinstance(sys.stderr, utils.StdSim) else sys.stderr,
+                        shell=True,
+                        **kwargs,
+                    )
+                    break
+                except PermissionError:
+                    # The pipeline exited before the command could join its group.
+                    if kwargs.pop("process_group", None) is None:
+                        raise
 
             proc_reader = utils.ProcReader(proc, self.stdout, sys.stderr)
             proc_reader.wait()

@@ -587,12 +587,15 @@ def send_sigint(self) -> None:
             # the whole process group to make sure it propagates further than the shell
             try:
                 group_id = os.getpgid(self._proc.pid)
-                # Pipelines have their own group. Never re-signal our own group:
-                # other ProcReader callers may share it and already received Ctrl-C.
-                if group_id != os.getpgrp():
-                    os.killpg(group_id, signal.SIGINT)
             except ProcessLookupError:
-                return
+                # Pipelines lead their own group. A shell command that joined it, such
+                # as `shell sleep 100 | head -1`, can outlive the reaped consumer.
+                group_id = self._proc.pid
+            # 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):
+                    os.killpg(group_id, signal.SIGINT)
 
     def terminate(self) -> None:
         """Terminate the process."""
@@ -605,6 +608,13 @@ def terminate(self) -> None:
             with contextlib.suppress(ProcessLookupError):
                 os.kill(self._proc.pid, signal.SIGTERM)
 
+    @property
+    def terminal_group(self) -> int | None:
+        """Process group of a running terminal pipeline, which a producer may join, or None."""
+        if self._terminal_fd is None or self._proc.returncode is not None:
+            return None
+        return self._proc.pid
+
     @staticmethod
     def _set_foreground_group(terminal_fd: int, group_id: int) -> None:
         """Transfer the terminal without stopping this background thread with SIGTTOU."""

@@ -20,20 +20,22 @@
 
 
 @pytest.mark.parametrize(
-    ("finish", "stop_job", "shell_child", "launcher"),
+    ("finish", "stop_job", "shell_child", "launcher", "producer"),
     [
-        pytest.param("interrupts", True, False, "direct", id="direct-signals-and-job-control"),
-        pytest.param("interrupts", True, True, "sh", id="wrapper-signals-and-job-control"),
-        pytest.param("exit_sigint", False, False, "direct", id="interrupt-busy-producer"),
-        pytest.param("exit_sigint", True, True, "uv", id="uv-stop-and-interrupt-busy-producer"),
-        pytest.param("read_input", False, False, "direct", id="nested-prompt"),
-        pytest.param("shell_input", False, True, "direct", id="shell-input"),
-        pytest.param("direct_input", False, False, "sh", id="direct-input-with-toolbar-off"),
-        pytest.param("direct_input", False, False, "exec", id="direct-input-in-orphaned-session"),
-        pytest.param("interrupts", False, False, "exec", id="orphaned-job-control"),
+        pytest.param("interrupts", True, False, "direct", "command", id="direct-signals-and-job-control"),
+        pytest.param("interrupts", True, True, "sh", "command", id="wrapper-signals-and-job-control"),
+        pytest.param("exit_sigint", False, False, "direct", "command", id="interrupt-busy-producer"),
+        pytest.param("exit_sigint", True, True, "uv", "command", id="uv-stop-and-interrupt-busy-producer"),
+        pytest.param("exit_sigint", False, False, "direct", "shell", id="interrupt-busy-shell-producer"),
+        pytest.param("exit_sigint", True, True, "sh", "shell", id="wrapper-stop-and-interrupt-busy-shell-producer"),
+        pytest.param("read_input", False, False, "direct", "command", id="nested-prompt"),
+        pytest.param("shell_input", False, True, "direct", "command", id="shell-input"),
+        pytest.param("direct_input", False, False, "sh", "command", id="direct-input-with-toolbar-off"),
+        pytest.param("direct_input", False, False, "exec", "command", id="direct-input-in-orphaned-session"),
+        pytest.param("interrupts", False, False, "exec", "command", id="orphaned-job-control"),
     ],
 )
-def test_pipeline_stops_with_cmd2_and_returns_terminal(tmp_path, finish, stop_job, shell_child, launcher) -> None:
+def test_pipeline_stops_with_cmd2_and_returns_terminal(tmp_path, finish, stop_job, shell_child, launcher, producer) -> None:
     import fcntl
     import pty
     import struct
@@ -88,8 +90,10 @@ def test_pipeline_stops_with_cmd2_and_returns_terminal(tmp_path, finish, stop_jo
     application = tmp_path / "application.py"
     application_pid = tmp_path / "application.pid"
     interrupt_request = tmp_path / "interrupt.request"
+    last_result = tmp_path / "last_result"
     application.write_text(
         "from cmd2 import Cmd, ToolbarMode\n"
+        "from cmd2.plugin import CommandFinalizationData\n"
         "import getpass, os, pathlib, signal, threading, time\n"
         "signal.signal(signal.SIGTSTP, signal.SIG_DFL)\n"
         f"pathlib.Path({str(application_pid)!r}).write_text(str(os.getpid()))\n"
@@ -110,7 +114,11 @@ def test_pipeline_stops_with_cmd2_and_returns_terminal(tmp_path, finish, stop_jo
         "        os.write(2, b'RAW> ')\n"
         "        assert os.read(0, 7) == b'direct\\n'\n"
         "        self.poutput('INPUT_COMPLETE')\n"
+        "    def record_result(self, data: CommandFinalizationData) -> CommandFinalizationData:\n"
+        f"        pathlib.Path({str(last_result)!r}).write_text(repr(self.last_result))\n"
+        "        return data\n"
         f"app = App(bottom_toolbar_mode=ToolbarMode.{'OFF' if finish == 'direct_input' else 'RESERVED'})\n"
+        "app.register_cmdfinalization_hook(app.record_result)\n"
         "app.prompt = 'TEST> '\n"
         "app.debug = True\n"
         "app.main_session.bottom_toolbar = 'STATUS'\n"
@@ -206,6 +214,21 @@ def stopped(*pids: int) -> bool:
         assert job_group > 1
         wait_until(lambda: os.tcgetpgrp(master) == job_group)
         command = "busy" if finish == "exit_sigint" else "help -v"
+        if producer == "shell":
+            # A shell command writes into the pipe itself rather than through cmd2's
+            # stdout, so cmd2 cannot lend the terminal write by write. Like seq or git
+            # log, this producer dies from SIGINT rather than handling it.
+            busy_script = tmp_path / "busy.py"
+            busy_script.write_text(
+                "import os, signal, sys, time\n"
+                "signal.signal(signal.SIGINT, signal.SIG_DFL)\n"
+                "os.write(2, b'BUSY_READY\\n')\n"
+                "sys.stdout.write('x' * 262144)\n"
+                "sys.stdout.flush()\n"
+                "time.sleep(30)\n",
+                encoding="utf-8",
+            )
+            command = f"shell {shlex.quote(sys.executable)} {shlex.quote(str(busy_script))}"
         if finish == "read_input":
             command = "ask"
         elif finish == "direct_input":
@@ -287,6 +310,11 @@ def stopped(*pids: int) -> bool:
         if finish != "exit_sigint":
             send("q")
         wait_until(lambda: os.tcgetpgrp(master) == job_group and "TEST>" in "\n".join(screen.display))
+        if producer == "shell":
+            # Ctrl-C reached the producer directly, as in a shell pipeline. It did not
+            # merely die of a broken pipe once the pager was gone.
+            wait_until(last_result.exists)
+            assert last_result.read_text() == repr(-signal.SIGINT)
         start = len(transcript)
         send("help quit\n")
         wait_until(lambda: "Exit this application" in transcript[start:])
@@ -309,3 +337,66 @@ def stopped(*pids: int) -> bool:
         os.close(master)
         process.kill()
         process.wait(timeout=5)
+
+
+def test_pipeline_from_worker_thread_stays_isolated(tmp_path) -> None:
+    """A pipe started off the main thread cannot install job-control handlers.
+
+    It must fall back to running the pipeline in its own session, as before, rather
+    than failing after Popen and leaving the child unreaped.
+    """
+    import pty
+
+    shell = shutil.which("bash")
+    if shell is None:
+        pytest.skip("requires an interactive bash shell")
+    application = tmp_path / "application.py"
+    application.write_text(
+        "from cmd2 import Cmd, ToolbarMode\n"
+        "import os, threading\n"
+        "app = Cmd(bottom_toolbar_mode=ToolbarMode.OFF)\n"
+        "outcome = []\n"
+        "worker = threading.Thread(target=lambda: outcome.append(app.onecmd_plus_hooks('help quit | cat')))\n"
+        "worker.start()\n"
+        "worker.join()\n"
+        "try:\n"
+        "    reaped = os.waitpid(-1, os.WNOHANG)\n"
+        "except ChildProcessError:\n"
+        "    reaped = None\n"
+        "os.write(1, f'WORKER_DONE {outcome} {reaped}\\n'.encode())\n",
+        encoding="utf-8",
+    )
+    master, slave = pty.openpty()
+    bootstrap = (
+        "import os, fcntl, termios; os.setsid(); "
+        "fcntl.ioctl(0, termios.TIOCSCTTY, 0); "
+        "os.execv(os.environ['TEST_SHELL'], ['bash', '--noprofile', '--norc', '-i'])"
+    )
+    env = dict(os.environ, TERM="xterm-256color", PS1="OUTER> ", TEST_SHELL=shell, SHELL=shell)
+    env["PYTHONPATH"] = str(Path(__file__).resolve().parents[1])
+    process = subprocess.Popen([sys.executable, "-c", bootstrap], stdin=slave, stdout=slave, stderr=slave, env=env)
+    os.close(slave)
+    decoder = codecs.getincrementaldecoder("utf-8")("replace")
+    transcript = ""
+
+    def wait_until(predicate):
+        nonlocal transcript
+        deadline = time.monotonic() + 10
+        while time.monotonic() < deadline:
+            if select.select([master], [], [], 0.05)[0]:
+                transcript += decoder.decode(os.read(master, 65536))
+            if predicate():
+                return
+        pytest.fail(f"terminal condition timed out:\n{transcript}")
+
+    try:
+        wait_until(lambda: "OUTER> " in transcript)
+        os.write(master, f"{shlex.quote(sys.executable)} {shlex.quote(str(application))}\n".encode())
+        # The whole line: a partial read must not satisfy the wait before the reap result arrives.
+        wait_until(lambda: re.search(r"WORKER_DONE .*\r\n", transcript) is not None)
+        assert "Exit this application" in transcript
+        assert "WORKER_DONE [False] None" in transcript
+    finally:
+        os.close(master)
+        process.kill()
+        process.wait(timeout=5)
@@ -223,7 +223,7 @@ def test_proc_reader_send_sigint(pr_none) -> None:
 @pytest.mark.skipif(sys.platform == "win32", reason="POSIX process groups")
 def test_proc_reader_does_not_resignal_its_own_group(pr_none) -> None:
     try:
-        with mock.patch("os.getpgid", return_value=os.getpgrp()), mock.patch("os.killpg") as killpg:
+        with mock.patch("os.getpgrp", return_value=pr_none._proc.pid), mock.patch("os.killpg") as killpg:
             pr_none.send_sigint()
         killpg.assert_not_called()
     finally:
@@ -232,11 +232,40 @@ def test_proc_reader_does_not_resignal_its_own_group(pr_none) -> None:
 
 
 @pytest.mark.skipif(sys.platform == "win32", reason="POSIX process groups")
-def test_proc_reader_sigint_after_consumer_exit() -> None:
-    reader = cu.ProcReader(mock.Mock(stdout=None, stderr=None), sys.stdout, sys.stderr)
-    with mock.patch("os.getpgid", side_effect=ProcessLookupError), mock.patch("os.killpg") as killpg:
+def test_proc_reader_sigint_after_pipeline_exit() -> None:
+    reader = cu.ProcReader(mock.Mock(pid=os.getpid() + 1, stdout=None, stderr=None), sys.stdout, sys.stderr)
+    with (
+        mock.patch("os.getpgid", side_effect=ProcessLookupError),
+        mock.patch("os.killpg", side_effect=ProcessLookupError) as killpg,
+    ):
         reader.send_sigint()
-    killpg.assert_not_called()
+    killpg.assert_called_once_with(reader._proc.pid, signal.SIGINT)
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="POSIX process groups")
+def test_proc_reader_sigint_reaches_group_after_leader_exit() -> None:
+    """A shell producer joins the pipeline's group and can outlive the consumer that led it."""
+    import subprocess
+
+    # A terminal pipeline leads its own group within our session, so a producer may join it.
+    leader = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(30)"], process_group=0)
+    reader = cu.ProcReader(leader, sys.stdout, sys.stderr)
+    member_code = (
+        "import signal, time; signal.signal(signal.SIGINT, signal.SIG_DFL); print('ready', flush=True); time.sleep(30)"
+    )
+    member = subprocess.Popen([sys.executable, "-c", member_code], stdout=subprocess.PIPE, process_group=leader.pid)
+    try:
+        assert member.stdout is not None
+        assert member.stdout.readline().strip() == b"ready"
+        reader.terminate()
+        reader.wait()
+        assert leader.returncode == -signal.SIGTERM
+
+        reader.send_sigint()
+        assert member.wait(timeout=5) == -signal.SIGINT
+    finally:
+        member.kill()
+        member.wait()
 
 
 def test_proc_reader_terminate(pr_none) -> None: